diff --git a/services/summary_cache.py b/services/summary_cache.py index fce9049..b6028ef 100644 --- a/services/summary_cache.py +++ b/services/summary_cache.py @@ -13,14 +13,20 @@ import json import logging import os +import threading +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path -from typing import Any +from typing import Any, TypeVar from utils.exclusion_rules import RuleTokens _logger = logging.getLogger(__name__) +# Serialises fingerprint compute, cache read-compare, and cache write so threaded +# WSGI workers cannot lost-update a peer's fresh cache entry (see design guide). +_summary_cache_lock = threading.Lock() + CACHE_VERSION = 1 CACHE_DIR = Path.home() / ".cache" / "cursor-chat-browser" PROJECTS_CACHE_FILE = CACHE_DIR / "projects.json" @@ -28,6 +34,8 @@ INVALID_WORKSPACE_ALIASES_CACHE_FILE = CACHE_DIR / "invalid-workspace-aliases.json" TAB_SUMMARIES_PREFIX = "tab-summaries-" +T = TypeVar("T") + def nocache_enabled(*, request_nocache: bool = False) -> bool: """Return whether summary-cache reads should be bypassed. @@ -107,6 +115,26 @@ def _entry_mtimes(entry: dict[str, Any]) -> list[list[str | int]]: } +def _workspace_storage_fingerprint( + workspace_path: str, + workspace_entries: list[dict[str, Any]], + rules: list[RuleTokens], +) -> dict[str, Any]: + """Fingerprint workspace storage, resolving global-db and cli-chats paths.""" + from services.workspace_db import global_storage_db_path + from utils.workspace_path import get_cli_chats_path + + gdb = global_storage_db_path(workspace_path) + cli_path = get_cli_chats_path() + return fingerprint_workspace_storage( + workspace_path, + workspace_entries, + global_db_path=gdb if os.path.isfile(gdb) else None, + rules=rules, + cli_chats_path=cli_path if os.path.isdir(cli_path) else None, + ) + + def _normalize_fingerprint(fp: dict[str, Any]) -> dict[str, Any]: """Normalize fingerprint for comparison (JSON round-trip uses lists, not tuples).""" normalized = dict(fp) @@ -125,7 +153,7 @@ def _fingerprint_equal(a: object, b: dict[str, Any]) -> bool: return _normalize_fingerprint(a) == _normalize_fingerprint(b) -def _read_cache_file(path: Path | str) -> dict[str, Any] | None: +def _read_cache_file_unlocked(path: Path | str) -> dict[str, Any] | None: p = Path(path) if not p.is_file(): return None @@ -140,7 +168,7 @@ def _read_cache_file(path: Path | str) -> dict[str, Any] | None: return None -def _write_cache_file(path: Path | str, payload: dict[str, Any]) -> None: +def _write_cache_file_unlocked(path: Path | str, payload: dict[str, Any]) -> None: p = Path(path) try: p.parent.mkdir(parents=True, exist_ok=True) @@ -152,19 +180,15 @@ def _write_cache_file(path: Path | str, payload: dict[str, Any]) -> None: _logger.warning("Summary cache write failed for %s: %s", path, e) -def get_cached_projects( - fingerprint: dict[str, Any], -) -> tuple[list[dict[str, Any]], list[dict[str, Any]]] | None: - """Load cached workspace project list when the fingerprint matches. +def _write_cache_file(path: Path | str, payload: dict[str, Any]) -> None: + with _summary_cache_lock: + _write_cache_file_unlocked(path, payload) - Args: - fingerprint: Storage mtime/rules digest from - :func:`fingerprint_workspace_storage`. - Returns: - ``(projects, warnings)`` on hit, else ``None``. - """ - data = _read_cache_file(PROJECTS_CACHE_FILE) +def _get_cached_projects_unlocked( + fingerprint: dict[str, Any], +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]] | None: + data = _read_cache_file_unlocked(PROJECTS_CACHE_FILE) if not data: return None if not _fingerprint_equal(data.get("fingerprint"), fingerprint): @@ -178,19 +202,28 @@ def get_cached_projects( return projects, warnings -def set_cached_projects( +def get_cached_projects( fingerprint: dict[str, Any], - projects: list[dict[str, Any]], - warnings: list[dict[str, Any]], -) -> None: - """Write workspace project list and warnings to the disk cache. +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]] | None: + """Load cached workspace project list when the fingerprint matches. Args: - fingerprint: Invalidation fingerprint paired with the payload. - projects: Sidebar project dicts. - warnings: Parse warnings emitted while building *projects*. + fingerprint: Storage mtime/rules digest from + :func:`fingerprint_workspace_storage`. + + Returns: + ``(projects, warnings)`` on hit, else ``None``. """ - _write_cache_file( + with _summary_cache_lock: + return _get_cached_projects_unlocked(fingerprint) + + +def _set_cached_projects_unlocked( + fingerprint: dict[str, Any], + projects: list[dict[str, Any]], + warnings: list[dict[str, Any]], +) -> None: + _write_cache_file_unlocked( PROJECTS_CACHE_FILE, { "fingerprint": fingerprint, @@ -200,18 +233,72 @@ def set_cached_projects( ) -def get_cached_composer_id_to_ws( +def set_cached_projects( fingerprint: dict[str, Any], -) -> dict[str, str] | None: - """Load cached composer-id → workspace-id map when the fingerprint matches. + projects: list[dict[str, Any]], + warnings: list[dict[str, Any]], +) -> None: + """Write workspace project list and warnings to the disk cache. Args: - fingerprint: Storage mtime/rules digest. - - Returns: - Mapping on hit, else ``None``. + fingerprint: Invalidation fingerprint paired with the payload. + projects: Sidebar project dicts. + warnings: Parse warnings emitted while building *projects*. """ - data = _read_cache_file(COMPOSER_MAP_CACHE_FILE) + with _summary_cache_lock: + _set_cached_projects_unlocked(fingerprint, projects, warnings) + + +def _get_or_build_cached( + workspace_path: str, + workspace_entries: list[dict[str, Any]], + rules: list[RuleTokens], + *, + build_fn: Callable[[], T], + get_unlocked: Callable[[dict[str, Any]], T | None], + set_unlocked: Callable[[dict[str, Any], T], None], + should_cache: Callable[[T], bool] | None = None, +) -> T: + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = get_unlocked(fingerprint) + if hit is not None: + return hit + + built = build_fn() + + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = get_unlocked(fingerprint) + if hit is not None: + return hit + if should_cache is None or should_cache(built): + set_unlocked(fingerprint, built) + return built + + +def get_or_build_cached_projects( + workspace_path: str, + workspace_entries: list[dict[str, Any]], + rules: list[RuleTokens], + *, + build_fn: Callable[[], tuple[list[dict[str, Any]], list[dict[str, Any]]]], +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """Return cached projects or build once under double-checked locking.""" + return _get_or_build_cached( + workspace_path, + workspace_entries, + rules, + build_fn=build_fn, + get_unlocked=_get_cached_projects_unlocked, + set_unlocked=lambda fp, built: _set_cached_projects_unlocked(fp, built[0], built[1]), + ) + + +def _get_cached_composer_id_to_ws_unlocked( + fingerprint: dict[str, Any], +) -> dict[str, str] | None: + data = _read_cache_file_unlocked(COMPOSER_MAP_CACHE_FILE) if not data: return None if not _fingerprint_equal(data.get("fingerprint"), fingerprint): @@ -222,17 +309,26 @@ def get_cached_composer_id_to_ws( return {str(k): str(v) for k, v in mapping.items()} -def set_cached_composer_id_to_ws( +def get_cached_composer_id_to_ws( fingerprint: dict[str, Any], - mapping: dict[str, str], -) -> None: - """Persist composer-id → workspace-id map under *fingerprint*. +) -> dict[str, str] | None: + """Load cached composer-id → workspace-id map when the fingerprint matches. Args: - fingerprint: Invalidation fingerprint paired with *mapping*. - mapping: Composer UUID to workspace folder name. + fingerprint: Storage mtime/rules digest. + + Returns: + Mapping on hit, else ``None``. """ - _write_cache_file( + with _summary_cache_lock: + return _get_cached_composer_id_to_ws_unlocked(fingerprint) + + +def _set_cached_composer_id_to_ws_unlocked( + fingerprint: dict[str, Any], + mapping: dict[str, str], +) -> None: + _write_cache_file_unlocked( COMPOSER_MAP_CACHE_FILE, { "fingerprint": fingerprint, @@ -241,18 +337,42 @@ def set_cached_composer_id_to_ws( ) -def get_cached_invalid_workspace_aliases( +def set_cached_composer_id_to_ws( fingerprint: dict[str, Any], -) -> dict[str, str] | None: - """Load cached invalid-workspace alias map when the fingerprint matches. + mapping: dict[str, str], +) -> None: + """Persist composer-id → workspace-id map under *fingerprint*. Args: - fingerprint: Storage mtime/rules digest. - - Returns: - ``{invalid_id: replacement_id}`` on hit, else ``None``. + fingerprint: Invalidation fingerprint paired with *mapping*. + mapping: Composer UUID to workspace folder name. """ - data = _read_cache_file(INVALID_WORKSPACE_ALIASES_CACHE_FILE) + with _summary_cache_lock: + _set_cached_composer_id_to_ws_unlocked(fingerprint, mapping) + + +def get_or_build_cached_composer_id_to_ws( + workspace_path: str, + workspace_entries: list[dict[str, Any]], + rules: list[RuleTokens], + *, + build_fn: Callable[[], dict[str, str]], +) -> dict[str, str]: + """Return cached composer map or build once under double-checked locking.""" + return _get_or_build_cached( + workspace_path, + workspace_entries, + rules, + build_fn=build_fn, + get_unlocked=_get_cached_composer_id_to_ws_unlocked, + set_unlocked=_set_cached_composer_id_to_ws_unlocked, + ) + + +def _get_cached_invalid_workspace_aliases_unlocked( + fingerprint: dict[str, Any], +) -> dict[str, str] | None: + data = _read_cache_file_unlocked(INVALID_WORKSPACE_ALIASES_CACHE_FILE) if not data: return None if not _fingerprint_equal(data.get("fingerprint"), fingerprint): @@ -276,6 +396,34 @@ def get_cached_invalid_workspace_aliases( return validated +def get_cached_invalid_workspace_aliases( + fingerprint: dict[str, Any], +) -> dict[str, str] | None: + """Load cached invalid-workspace alias map when the fingerprint matches. + + Args: + fingerprint: Storage mtime/rules digest. + + Returns: + ``{invalid_id: replacement_id}`` on hit, else ``None``. + """ + with _summary_cache_lock: + return _get_cached_invalid_workspace_aliases_unlocked(fingerprint) + + +def _set_cached_invalid_workspace_aliases_unlocked( + fingerprint: dict[str, Any], + aliases: dict[str, str], +) -> None: + _write_cache_file_unlocked( + INVALID_WORKSPACE_ALIASES_CACHE_FILE, + { + "fingerprint": fingerprint, + "invalid_workspace_aliases": aliases, + }, + ) + + def set_cached_invalid_workspace_aliases( fingerprint: dict[str, Any], aliases: dict[str, str], @@ -286,12 +434,25 @@ def set_cached_invalid_workspace_aliases( fingerprint: Invalidation fingerprint paired with *aliases*. aliases: ``{invalid_id: replacement_id}`` from alias inference. """ - _write_cache_file( - INVALID_WORKSPACE_ALIASES_CACHE_FILE, - { - "fingerprint": fingerprint, - "invalid_workspace_aliases": aliases, - }, + with _summary_cache_lock: + _set_cached_invalid_workspace_aliases_unlocked(fingerprint, aliases) + + +def get_or_build_cached_invalid_workspace_aliases( + workspace_path: str, + workspace_entries: list[dict[str, Any]], + rules: list[RuleTokens], + *, + build_fn: Callable[[], dict[str, str]], +) -> dict[str, str]: + """Return cached alias map or build once under double-checked locking.""" + return _get_or_build_cached( + workspace_path, + workspace_entries, + rules, + build_fn=build_fn, + get_unlocked=_get_cached_invalid_workspace_aliases_unlocked, + set_unlocked=_set_cached_invalid_workspace_aliases_unlocked, ) @@ -300,20 +461,11 @@ def _tab_summaries_path(workspace_id: str) -> Path: return CACHE_DIR / f"{TAB_SUMMARIES_PREFIX}{safe}.json" -def get_cached_tab_summaries( +def _get_cached_tab_summaries_unlocked( fingerprint: dict[str, Any], workspace_id: str, ) -> tuple[dict[str, Any], int] | None: - """Load cached tab-summary response for one workspace when fingerprint matches. - - Args: - fingerprint: Storage mtime/rules digest. - workspace_id: Workspace folder name the payload belongs to. - - Returns: - ``(payload, status)`` on hit, else ``None``. - """ - data = _read_cache_file(_tab_summaries_path(workspace_id)) + data = _read_cache_file_unlocked(_tab_summaries_path(workspace_id)) if not data: return None if data.get("workspace_id") != workspace_id: @@ -327,6 +479,40 @@ def get_cached_tab_summaries( return payload, status +def get_cached_tab_summaries( + fingerprint: dict[str, Any], + workspace_id: str, +) -> tuple[dict[str, Any], int] | None: + """Load cached tab-summary response for one workspace when fingerprint matches. + + Args: + fingerprint: Storage mtime/rules digest. + workspace_id: Workspace folder name the payload belongs to. + + Returns: + ``(payload, status)`` on hit, else ``None``. + """ + with _summary_cache_lock: + return _get_cached_tab_summaries_unlocked(fingerprint, workspace_id) + + +def _set_cached_tab_summaries_unlocked( + fingerprint: dict[str, Any], + workspace_id: str, + payload: dict[str, Any], + status: int, +) -> None: + _write_cache_file_unlocked( + _tab_summaries_path(workspace_id), + { + "workspace_id": workspace_id, + "fingerprint": fingerprint, + "payload": payload, + "status": status, + }, + ) + + def set_cached_tab_summaries( fingerprint: dict[str, Any], workspace_id: str, @@ -341,12 +527,27 @@ def set_cached_tab_summaries( payload: JSON body returned to clients. status: HTTP status code paired with *payload*. """ - _write_cache_file( - _tab_summaries_path(workspace_id), - { - "workspace_id": workspace_id, - "fingerprint": fingerprint, - "payload": payload, - "status": status, - }, + with _summary_cache_lock: + _set_cached_tab_summaries_unlocked(fingerprint, workspace_id, payload, status) + + +def get_or_build_cached_tab_summaries( + workspace_path: str, + workspace_entries: list[dict[str, Any]], + rules: list[RuleTokens], + workspace_id: str, + *, + build_fn: Callable[[], tuple[dict[str, Any], int]], +) -> tuple[dict[str, Any], int]: + """Return cached tab summaries or build once under double-checked locking.""" + return _get_or_build_cached( + workspace_path, + workspace_entries, + rules, + build_fn=build_fn, + get_unlocked=lambda fp: _get_cached_tab_summaries_unlocked(fp, workspace_id), + set_unlocked=lambda fp, built: _set_cached_tab_summaries_unlocked( + fp, workspace_id, built[0], built[1], + ), + should_cache=lambda built: built[1] == 200, ) diff --git a/services/workspace_context.py b/services/workspace_context.py index de77e3d..576bf95 100644 --- a/services/workspace_context.py +++ b/services/workspace_context.py @@ -2,7 +2,6 @@ from __future__ import annotations -import os import sqlite3 from dataclasses import dataclass, replace from typing import Any @@ -16,7 +15,6 @@ build_composer_id_to_workspace_id_cached, collect_invalid_workspace_ids, collect_workspace_entries, - global_storage_db_path, load_bubble_map, load_project_layouts_map, safe_fetchall, @@ -179,46 +177,37 @@ def resolve_invalid_workspace_aliases_cached( return {} from services.summary_cache import ( - fingerprint_workspace_storage, - get_cached_invalid_workspace_aliases, + get_or_build_cached_invalid_workspace_aliases, nocache_enabled, - set_cached_invalid_workspace_aliases, ) - from utils.workspace_path import get_cli_chats_path - gdb = global_storage_db_path(workspace_path) - cli_path = get_cli_chats_path() - fingerprint = fingerprint_workspace_storage( + def build() -> dict[str, str]: + layouts = ( + project_layouts_map + if project_layouts_map is not None + else load_project_layouts_map(global_db) + ) + composer_rows = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL) + return infer_invalid_workspace_aliases( + composer_rows=composer_rows, + project_layouts_map=layouts, + project_name_map=ctx.project_name_to_workspace_id, + workspace_path_map=ctx.workspace_path_to_id, + workspace_entries=ctx.workspace_entries, + bubble_map={}, + composer_id_to_ws=ctx.composer_id_to_workspace_id, + invalid_workspace_ids=ctx.invalid_workspace_ids, + ) + + if nocache_enabled(request_nocache=nocache): + return build() + + return get_or_build_cached_invalid_workspace_aliases( workspace_path, ctx.workspace_entries, - global_db_path=gdb if os.path.isfile(gdb) else None, - rules=rules, - cli_chats_path=cli_path if os.path.isdir(cli_path) else None, - ) - if not nocache_enabled(request_nocache=nocache): - cached = get_cached_invalid_workspace_aliases(fingerprint) - if cached is not None: - return cached - - layouts = ( - project_layouts_map - if project_layouts_map is not None - else load_project_layouts_map(global_db) - ) - composer_rows = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL) - aliases = infer_invalid_workspace_aliases( - composer_rows=composer_rows, - project_layouts_map=layouts, - project_name_map=ctx.project_name_to_workspace_id, - workspace_path_map=ctx.workspace_path_to_id, - workspace_entries=ctx.workspace_entries, - bubble_map={}, - composer_id_to_ws=ctx.composer_id_to_workspace_id, - invalid_workspace_ids=ctx.invalid_workspace_ids, + rules, + build_fn=build, ) - if not nocache_enabled(request_nocache=nocache): - set_cached_invalid_workspace_aliases(fingerprint, aliases) - return aliases def with_invalid_workspace_aliases( diff --git a/services/workspace_db.py b/services/workspace_db.py index 2c43a0d..1ccd386 100644 --- a/services/workspace_db.py +++ b/services/workspace_db.py @@ -436,30 +436,22 @@ def build_composer_id_to_workspace_id_cached( ) -> dict[str, str]: """Like :func:`build_composer_id_to_workspace_id` with optional disk cache.""" from services.summary_cache import ( - fingerprint_workspace_storage, - get_cached_composer_id_to_ws, + get_or_build_cached_composer_id_to_ws, nocache_enabled, - set_cached_composer_id_to_ws, ) - from utils.workspace_path import get_cli_chats_path - gdb = global_storage_db_path(workspace_path) - cli_path = get_cli_chats_path() - fingerprint = fingerprint_workspace_storage( + def build() -> dict[str, str]: + return build_composer_id_to_workspace_id(workspace_path, workspace_entries) + + if nocache_enabled(request_nocache=nocache): + return build() + + return get_or_build_cached_composer_id_to_ws( workspace_path, workspace_entries, - global_db_path=gdb if os.path.isfile(gdb) else None, - rules=rules, - cli_chats_path=cli_path if os.path.isdir(cli_path) else None, + rules, + build_fn=build, ) - if not nocache_enabled(request_nocache=nocache): - cached = get_cached_composer_id_to_ws(fingerprint) - if cached is not None: - return cached - mapping = build_composer_id_to_workspace_id(workspace_path, workspace_entries) - if not nocache_enabled(request_nocache=nocache): - set_cached_composer_id_to_ws(fingerprint, mapping) - return mapping @contextmanager diff --git a/services/workspace_listing.py b/services/workspace_listing.py index 7a31a9d..b88b6d1 100644 --- a/services/workspace_listing.py +++ b/services/workspace_listing.py @@ -25,16 +25,13 @@ parse_composer_data_row, ) from services.summary_cache import ( - fingerprint_workspace_storage, - get_cached_projects, + get_or_build_cached_projects, nocache_enabled, - set_cached_projects, ) from services.workspace_context import resolve_invalid_workspace_aliases_cached from services.workspace_db import ( COMPOSER_ROWS_WITH_HEADERS_SQL, collect_workspace_entries, - global_storage_db_path, load_project_layouts_map, open_global_db, safe_fetchall, @@ -69,35 +66,35 @@ def list_workspace_projects( :meth:`models.ParseWarningCollector.to_api_list`; empty when no skips. """ effective_nocache = nocache_enabled(request_nocache=nocache) - workspace_entries: list[dict[str, Any]] | None = None - if not effective_nocache: - workspace_entries = collect_workspace_entries(workspace_path) - gdb = global_storage_db_path(workspace_path) - cli_path = get_cli_chats_path() - fingerprint = fingerprint_workspace_storage( + if effective_nocache: + orch = prepare_workspace_orchestration( workspace_path, - workspace_entries, - global_db_path=gdb if os.path.isfile(gdb) else None, - rules=rules, - cli_chats_path=cli_path if os.path.isdir(cli_path) else None, + rules, + nocache=True, + ) + return _build_workspace_projects_uncached( + workspace_path, rules, orch, nocache=True, + ) + + workspace_entries = collect_workspace_entries(workspace_path) + + def build() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + orch = prepare_workspace_orchestration( + workspace_path, + rules, + nocache=False, + workspace_entries=workspace_entries, + ) + return _build_workspace_projects_uncached( + workspace_path, rules, orch, nocache=False, ) - cached = get_cached_projects(fingerprint) - if cached is not None: - return cached - orch = prepare_workspace_orchestration( + return get_or_build_cached_projects( workspace_path, + workspace_entries, rules, - nocache=effective_nocache, - workspace_entries=workspace_entries, - ) - - projects, warnings = _build_workspace_projects_uncached( - workspace_path, rules, orch, nocache=effective_nocache, + build_fn=build, ) - if not effective_nocache: - set_cached_projects(orch.fingerprint, projects, warnings) - return projects, warnings def _build_workspace_projects_uncached( diff --git a/services/workspace_tabs.py b/services/workspace_tabs.py index e19afff..3b2db88 100644 --- a/services/workspace_tabs.py +++ b/services/workspace_tabs.py @@ -3,7 +3,6 @@ import hashlib import json import logging -import os import sqlite3 from collections.abc import Iterator, Mapping from datetime import datetime @@ -44,10 +43,8 @@ parse_composer_data_row, ) from services.summary_cache import ( - fingerprint_workspace_storage, - get_cached_tab_summaries, + get_or_build_cached_tab_summaries, nocache_enabled, - set_cached_tab_summaries, ) from services.workspace_context import ( resolve_workspace_context_cached, @@ -56,7 +53,6 @@ from services.workspace_db import ( COMPOSER_ROWS_WITH_HEADERS_SQL, collect_workspace_entries, - global_storage_db_path, load_bubble_map, load_bubbles_for_composer, load_code_block_diff_map, @@ -67,7 +63,6 @@ open_global_db, safe_fetchall, ) -from utils.workspace_path import get_cli_chats_path from services.workspace_resolver import ( lookup_workspace_display_name, matching_workspace_ids_for_folder, @@ -530,26 +525,22 @@ def list_workspace_tab_summaries( but ``tabs`` entries carry no ``bubbles`` field. """ workspace_entries = collect_workspace_entries(workspace_path) - gdb = global_storage_db_path(workspace_path) - cli_path = get_cli_chats_path() - fingerprint = fingerprint_workspace_storage( + + def build() -> tuple[dict[str, Any], int]: + return _build_workspace_tab_summaries_uncached( + workspace_id, workspace_path, rules, workspace_entries, nocache=nocache, + ) + + if nocache_enabled(request_nocache=nocache): + return build() + + return get_or_build_cached_tab_summaries( workspace_path, workspace_entries, - global_db_path=gdb if os.path.isfile(gdb) else None, - rules=rules, - cli_chats_path=cli_path if os.path.isdir(cli_path) else None, - ) - if not nocache_enabled(request_nocache=nocache): - cached = get_cached_tab_summaries(fingerprint, workspace_id) - if cached is not None: - return cached - - payload, status = _build_workspace_tab_summaries_uncached( - workspace_id, workspace_path, rules, workspace_entries, nocache=nocache, + rules, + workspace_id, + build_fn=build, ) - if status == 200 and not nocache_enabled(request_nocache=nocache): - set_cached_tab_summaries(fingerprint, workspace_id, payload, status) - return payload, status def _build_workspace_tab_summaries_uncached( diff --git a/tests/test_summary_cache_concurrency.py b/tests/test_summary_cache_concurrency.py new file mode 100644 index 0000000..38c502f --- /dev/null +++ b/tests/test_summary_cache_concurrency.py @@ -0,0 +1,208 @@ +"""Concurrency regression tests for summary-cache fingerprint read-compare-write.""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path +from unittest.mock import patch + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if REPO_ROOT not in sys.path: + sys.path.insert(0, REPO_ROOT) + +from services import summary_cache +from services.summary_cache import ( + get_or_build_cached_projects, + get_cached_projects, +) + +_WAIT_TIMEOUT_S = 5.0 + + +def _make_workspace_fixture(root: str) -> tuple[str, list[dict[str, object]]]: + entry_dir = os.path.join(root, "entry1") + os.makedirs(entry_dir) + db_path = os.path.join(entry_dir, "state.vscdb") + with open(db_path, "wb") as f: + f.write(b"x") + workspace_path = root + entries: list[dict[str, object]] = [ + { + "name": "entry1", + "workspaceJsonPath": os.path.join(entry_dir, "workspace.json"), + }, + ] + return workspace_path, entries + + +class TestSummaryCacheConcurrency(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + cache_dir = Path(self.tmp.name) + self._patches = [ + patch.object(summary_cache, "CACHE_DIR", cache_dir), + patch.object( + summary_cache, + "PROJECTS_CACHE_FILE", + cache_dir / "projects.json", + ), + ] + for patcher in self._patches: + patcher.start() + + def tearDown(self): + for patcher in reversed(self._patches): + patcher.stop() + self.tmp.cleanup() + + def _setup_workspace(self, name: str) -> tuple[str, list[dict[str, object]]]: + ws_root = os.path.join(self.tmp.name, name) + os.makedirs(ws_root) + return _make_workspace_fixture(ws_root) + + def test_blocked_build_recheck_returns_peer_cache(self): + """Lost-update: peer write while build is blocked must win on recheck.""" + workspace_path, workspace_entries = self._setup_workspace("ws") + + peer_projects = [ + {"id": "peer", "name": "Peer", "conversationCount": 1, "lastModified": "x"}, + ] + stale_projects = [ + {"id": "stale", "name": "Stale", "conversationCount": 1, "lastModified": "y"}, + ] + build_started = threading.Event() + allow_build_finish = threading.Event() + + def blocked_build() -> tuple[list[dict[str, object]], list[dict[str, object]]]: + build_started.set() + self.assertTrue( + allow_build_finish.wait(timeout=_WAIT_TIMEOUT_S), + msg="timed out waiting to finish blocked build", + ) + return stale_projects, [] + + errors: list[str] = [] + results: list[tuple[list[dict[str, object]], list[dict[str, object]]]] = [] + + def thread_a() -> None: + try: + results.append( + get_or_build_cached_projects( + workspace_path, + workspace_entries, # type: ignore[arg-type] + [], + build_fn=blocked_build, # type: ignore[arg-type] + ), + ) + except Exception as exc: + errors.append(f"thread A: {exc}") + + def thread_b() -> None: + try: + self.assertTrue( + build_started.wait(timeout=_WAIT_TIMEOUT_S), + msg="thread B started before thread A entered build_fn", + ) + results.append( + get_or_build_cached_projects( + workspace_path, + workspace_entries, # type: ignore[arg-type] + [], + build_fn=lambda: (peer_projects, []), # type: ignore[arg-type] + ), + ) + except Exception as exc: + errors.append(f"thread B: {exc}") + finally: + allow_build_finish.set() + + with ThreadPoolExecutor(max_workers=2) as pool: + fut_a = pool.submit(thread_a) + fut_b = pool.submit(thread_b) + for fut in as_completed([fut_a, fut_b]): + fut.result() + + self.assertEqual(errors, [], "\n".join(errors)) + self.assertEqual(len(results), 2) + for projects, _warnings in results: + self.assertEqual(projects, peer_projects) + + with summary_cache.PROJECTS_CACHE_FILE.open(encoding="utf-8") as f: + on_disk = json.load(f) + self.assertEqual(on_disk.get("projects"), peer_projects) + + def test_concurrent_get_or_build_returns_consistent_results(self): + workspace_path, workspace_entries = self._setup_workspace("ws-warm") + warm_projects = [ + {"id": "warm", "name": "Warm", "conversationCount": 2, "lastModified": "z"}, + ] + fingerprint = summary_cache._workspace_storage_fingerprint( + workspace_path, + workspace_entries, # type: ignore[arg-type] + [], + ) + summary_cache.set_cached_projects(fingerprint, warm_projects, []) + + barrier = threading.Barrier(8) + errors: list[str] = [] + collected: list[tuple[list[dict[str, object]], list[dict[str, object]]]] = [] + + def reader() -> None: + try: + barrier.wait(timeout=_WAIT_TIMEOUT_S) + hit = get_or_build_cached_projects( + workspace_path, + workspace_entries, # type: ignore[arg-type] + [], + build_fn=lambda: (_ for _ in ()).throw( # type: ignore[arg-type] + AssertionError("build_fn must not run on warm cache"), + ), + ) + collected.append(hit) + except Exception as exc: + errors.append(str(exc)) + + with ThreadPoolExecutor(max_workers=8) as pool: + futures = [pool.submit(reader) for _ in range(8)] + for fut in as_completed(futures): + fut.result() + + self.assertEqual(errors, [], "\n".join(errors)) + self.assertEqual(len(collected), 8) + for projects, warnings in collected: + self.assertEqual(projects, warm_projects) + self.assertEqual(warnings, []) + + def test_get_cached_projects_remains_thread_safe(self): + fp = {"version": 1, "workspace_path": "/ws", "global_db_mtime_ns": 100} + projects = [{"id": "a", "name": "A", "conversationCount": 1, "lastModified": "x"}] + summary_cache.set_cached_projects(fp, projects, []) + + barrier = threading.Barrier(12) + errors: list[str] = [] + hits: list[tuple[list[dict[str, object]], list[dict[str, object]]] | None] = [] + + def reader() -> None: + try: + barrier.wait(timeout=_WAIT_TIMEOUT_S) + hits.append(get_cached_projects(fp)) + except Exception as exc: + errors.append(str(exc)) + + with ThreadPoolExecutor(max_workers=12) as pool: + futures = [pool.submit(reader) for _ in range(12)] + for fut in as_completed(futures): + fut.result() + + self.assertEqual(errors, [], "\n".join(errors)) + self.assertTrue(all(hit is not None and hit[0] == projects for hit in hits)) + + +if __name__ == "__main__": + unittest.main()