From 901c06a113c32c458396dd8d80765d0a3b6af081 Mon Sep 17 00:00:00 2001 From: chen Date: Wed, 29 Jul 2026 14:42:44 +0800 Subject: [PATCH 1/3] fix: serialise summary-cache fingerprint read-compare-write under a module lock --- services/summary_cache.py | 346 +++++++++++++++++++----- services/workspace_context.py | 59 ++-- services/workspace_db.py | 28 +- services/workspace_listing.py | 50 ++-- services/workspace_tabs.py | 37 +-- tests/test_summary_cache_concurrency.py | 210 ++++++++++++++ 6 files changed, 558 insertions(+), 172 deletions(-) create mode 100644 tests/test_summary_cache_concurrency.py diff --git a/services/summary_cache.py b/services/summary_cache.py index fce9049..929926c 100644 --- a/services/summary_cache.py +++ b/services/summary_cache.py @@ -13,6 +13,8 @@ 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 @@ -21,6 +23,10 @@ _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" @@ -107,6 +113,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 +151,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 +166,12 @@ 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 _read_cache_file(path: Path | str) -> dict[str, Any] | None: + with _summary_cache_lock: + return _read_cache_file_unlocked(path) + + +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 +183,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 +205,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 +236,51 @@ 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_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.""" + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = _get_cached_projects_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_cached_projects_unlocked(fingerprint) + if hit is not None: + return hit + _set_cached_projects_unlocked(fingerprint, built[0], built[1]) + return built + + +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 +291,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 +319,49 @@ 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.""" + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = _get_cached_composer_id_to_ws_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_cached_composer_id_to_ws_unlocked(fingerprint) + if hit is not None: + return hit + _set_cached_composer_id_to_ws_unlocked(fingerprint, built) + return built + + +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,17 +385,26 @@ def get_cached_invalid_workspace_aliases( return validated -def set_cached_invalid_workspace_aliases( +def get_cached_invalid_workspace_aliases( fingerprint: dict[str, Any], - aliases: dict[str, str], -) -> None: - """Persist invalid-workspace alias map under *fingerprint*. +) -> dict[str, str] | None: + """Load cached invalid-workspace alias map when the fingerprint matches. Args: - fingerprint: Invalidation fingerprint paired with *aliases*. - aliases: ``{invalid_id: replacement_id}`` from alias inference. + fingerprint: Storage mtime/rules digest. + + Returns: + ``{invalid_id: replacement_id}`` on hit, else ``None``. """ - _write_cache_file( + 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, @@ -295,25 +413,55 @@ def set_cached_invalid_workspace_aliases( ) +def set_cached_invalid_workspace_aliases( + fingerprint: dict[str, Any], + aliases: dict[str, str], +) -> None: + """Persist invalid-workspace alias map under *fingerprint*. + + Args: + fingerprint: Invalidation fingerprint paired with *aliases*. + aliases: ``{invalid_id: replacement_id}`` from alias inference. + """ + 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.""" + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = _get_cached_invalid_workspace_aliases_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_cached_invalid_workspace_aliases_unlocked(fingerprint) + if hit is not None: + return hit + _set_cached_invalid_workspace_aliases_unlocked(fingerprint, built) + return built + + def _tab_summaries_path(workspace_id: str) -> Path: safe = hashlib.sha256(workspace_id.encode("utf-8")).hexdigest()[:16] 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 +475,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 +523,32 @@ 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.""" + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = _get_cached_tab_summaries_unlocked(fingerprint, workspace_id) + if hit is not None: + return hit + + payload, status = build_fn() + + with _summary_cache_lock: + fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) + hit = _get_cached_tab_summaries_unlocked(fingerprint, workspace_id) + if hit is not None: + return hit + if status == 200: + _set_cached_tab_summaries_unlocked(fingerprint, workspace_id, payload, status) + return payload, status diff --git a/services/workspace_context.py b/services/workspace_context.py index de77e3d..9d865fa 100644 --- a/services/workspace_context.py +++ b/services/workspace_context.py @@ -179,46 +179,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..fe66ef6 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( + if nocache_enabled(request_nocache=nocache): + return build_composer_id_to_workspace_id(workspace_path, workspace_entries) + + def build() -> dict[str, str]: + return build_composer_id_to_workspace_id(workspace_path, workspace_entries) + + 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..7079fde 100644 --- a/services/workspace_listing.py +++ b/services/workspace_listing.py @@ -25,10 +25,8 @@ 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 ( @@ -69,35 +67,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..36de7de 100644 --- a/services/workspace_tabs.py +++ b/services/workspace_tabs.py @@ -44,10 +44,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 +54,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 +64,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 +526,23 @@ 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( + if nocache_enabled(request_nocache=nocache): + return _build_workspace_tab_summaries_uncached( + workspace_id, workspace_path, rules, workspace_entries, nocache=nocache, + ) + + def build() -> tuple[dict[str, Any], int]: + return _build_workspace_tab_summaries_uncached( + workspace_id, workspace_path, rules, workspace_entries, nocache=nocache, + ) + + 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..90039a1 --- /dev/null +++ b/tests/test_summary_cache_concurrency.py @@ -0,0 +1,210 @@ +"""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, +) + + +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() + self.cache_patch = patch.object(summary_cache, "CACHE_DIR", Path(self.tmp.name)) + self.cache_patch.start() + summary_cache.PROJECTS_CACHE_FILE = Path(self.tmp.name) / "projects.json" + summary_cache.COMPOSER_MAP_CACHE_FILE = ( + Path(self.tmp.name) / "composer-id-to-ws.json" + ) + summary_cache.INVALID_WORKSPACE_ALIASES_CACHE_FILE = ( + Path(self.tmp.name) / "invalid-workspace-aliases.json" + ) + + def tearDown(self): + self.cache_patch.stop() + self.tmp.cleanup() + + def test_blocked_build_recheck_returns_peer_cache(self): + """Lost-update: peer write while build is blocked must win on recheck.""" + ws_root = os.path.join(self.tmp.name, "ws") + os.makedirs(ws_root) + workspace_path, workspace_entries = _make_workspace_fixture(ws_root) + + 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() + build_count = 0 + build_count_lock = threading.Lock() + + def blocked_build() -> tuple[list[dict[str, object]], list[dict[str, object]]]: + nonlocal build_count + with build_count_lock: + build_count += 1 + count = build_count + build_started.set() + self.assertTrue( + allow_build_finish.wait(timeout=5.0), + msg="timed out waiting to finish blocked build", + ) + if count == 1: + return stale_projects, [] + return peer_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=5.0), + 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): + ws_root = os.path.join(self.tmp.name, "ws-warm") + os.makedirs(ws_root) + workspace_path, workspace_entries = _make_workspace_fixture(ws_root) + 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=5.0) + 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=5.0) + 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() From c801f6fb21366b5429c1ced08d1afc57beb588e5 Mon Sep 17 00:00:00 2001 From: chen Date: Wed, 29 Jul 2026 14:56:43 +0800 Subject: [PATCH 2/3] fix: patch cache paths in concurrency tests and reuse nocache builders --- services/workspace_db.py | 6 +++--- services/workspace_tabs.py | 7 +++---- tests/test_summary_cache_concurrency.py | 23 +++++++++++++---------- 3 files changed, 19 insertions(+), 17 deletions(-) diff --git a/services/workspace_db.py b/services/workspace_db.py index fe66ef6..1ccd386 100644 --- a/services/workspace_db.py +++ b/services/workspace_db.py @@ -440,12 +440,12 @@ def build_composer_id_to_workspace_id_cached( nocache_enabled, ) - if nocache_enabled(request_nocache=nocache): - return build_composer_id_to_workspace_id(workspace_path, workspace_entries) - 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, diff --git a/services/workspace_tabs.py b/services/workspace_tabs.py index 36de7de..5c8ab8d 100644 --- a/services/workspace_tabs.py +++ b/services/workspace_tabs.py @@ -526,16 +526,15 @@ def list_workspace_tab_summaries( but ``tabs`` entries carry no ``bubbles`` field. """ workspace_entries = collect_workspace_entries(workspace_path) - if nocache_enabled(request_nocache=nocache): - return _build_workspace_tab_summaries_uncached( - workspace_id, workspace_path, rules, workspace_entries, nocache=nocache, - ) 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, diff --git a/tests/test_summary_cache_concurrency.py b/tests/test_summary_cache_concurrency.py index 90039a1..7080ddd 100644 --- a/tests/test_summary_cache_concurrency.py +++ b/tests/test_summary_cache_concurrency.py @@ -42,18 +42,21 @@ def _make_workspace_fixture(root: str) -> tuple[str, list[dict[str, object]]]: class TestSummaryCacheConcurrency(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() - self.cache_patch = patch.object(summary_cache, "CACHE_DIR", Path(self.tmp.name)) - self.cache_patch.start() - summary_cache.PROJECTS_CACHE_FILE = Path(self.tmp.name) / "projects.json" - summary_cache.COMPOSER_MAP_CACHE_FILE = ( - Path(self.tmp.name) / "composer-id-to-ws.json" - ) - summary_cache.INVALID_WORKSPACE_ALIASES_CACHE_FILE = ( - Path(self.tmp.name) / "invalid-workspace-aliases.json" - ) + 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): - self.cache_patch.stop() + for patcher in reversed(self._patches): + patcher.stop() self.tmp.cleanup() def test_blocked_build_recheck_returns_peer_cache(self): From e89f46b30d40f94a742459d85a05ee3b42ce5dd6 Mon Sep 17 00:00:00 2001 From: chen Date: Wed, 29 Jul 2026 15:28:40 +0800 Subject: [PATCH 3/3] refactor: extract shared get_or_build_cached and tidy concurrency tests --- services/summary_cache.py | 117 ++++++++++++------------ services/workspace_context.py | 2 - services/workspace_listing.py | 1 - services/workspace_tabs.py | 1 - tests/test_summary_cache_concurrency.py | 33 +++---- 5 files changed, 72 insertions(+), 82 deletions(-) diff --git a/services/summary_cache.py b/services/summary_cache.py index 929926c..b6028ef 100644 --- a/services/summary_cache.py +++ b/services/summary_cache.py @@ -17,7 +17,7 @@ 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 @@ -34,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. @@ -166,11 +168,6 @@ def _read_cache_file_unlocked(path: Path | str) -> dict[str, Any] | None: return None -def _read_cache_file(path: Path | str) -> dict[str, Any] | None: - with _summary_cache_lock: - return _read_cache_file_unlocked(path) - - def _write_cache_file_unlocked(path: Path | str, payload: dict[str, Any]) -> None: p = Path(path) try: @@ -252,17 +249,19 @@ def set_cached_projects( _set_cached_projects_unlocked(fingerprint, projects, warnings) -def get_or_build_cached_projects( +def _get_or_build_cached( 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.""" + 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_cached_projects_unlocked(fingerprint) + hit = get_unlocked(fingerprint) if hit is not None: return hit @@ -270,13 +269,32 @@ def get_or_build_cached_projects( with _summary_cache_lock: fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) - hit = _get_cached_projects_unlocked(fingerprint) + hit = get_unlocked(fingerprint) if hit is not None: return hit - _set_cached_projects_unlocked(fingerprint, built[0], built[1]) + 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: @@ -341,21 +359,14 @@ def get_or_build_cached_composer_id_to_ws( build_fn: Callable[[], dict[str, str]], ) -> dict[str, str]: """Return cached composer map or build once under double-checked locking.""" - with _summary_cache_lock: - fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) - hit = _get_cached_composer_id_to_ws_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_cached_composer_id_to_ws_unlocked(fingerprint) - if hit is not None: - return hit - _set_cached_composer_id_to_ws_unlocked(fingerprint, built) - return built + 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( @@ -435,21 +446,14 @@ def get_or_build_cached_invalid_workspace_aliases( build_fn: Callable[[], dict[str, str]], ) -> dict[str, str]: """Return cached alias map or build once under double-checked locking.""" - with _summary_cache_lock: - fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) - hit = _get_cached_invalid_workspace_aliases_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_cached_invalid_workspace_aliases_unlocked(fingerprint) - if hit is not None: - return hit - _set_cached_invalid_workspace_aliases_unlocked(fingerprint, built) - return built + 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, + ) def _tab_summaries_path(workspace_id: str) -> Path: @@ -536,19 +540,14 @@ def get_or_build_cached_tab_summaries( build_fn: Callable[[], tuple[dict[str, Any], int]], ) -> tuple[dict[str, Any], int]: """Return cached tab summaries or build once under double-checked locking.""" - with _summary_cache_lock: - fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) - hit = _get_cached_tab_summaries_unlocked(fingerprint, workspace_id) - if hit is not None: - return hit - - payload, status = build_fn() - - with _summary_cache_lock: - fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules) - hit = _get_cached_tab_summaries_unlocked(fingerprint, workspace_id) - if hit is not None: - return hit - if status == 200: - _set_cached_tab_summaries_unlocked(fingerprint, workspace_id, payload, status) - return payload, status + 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 9d865fa..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, diff --git a/services/workspace_listing.py b/services/workspace_listing.py index 7079fde..b88b6d1 100644 --- a/services/workspace_listing.py +++ b/services/workspace_listing.py @@ -32,7 +32,6 @@ 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, diff --git a/services/workspace_tabs.py b/services/workspace_tabs.py index 5c8ab8d..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 diff --git a/tests/test_summary_cache_concurrency.py b/tests/test_summary_cache_concurrency.py index 7080ddd..38c502f 100644 --- a/tests/test_summary_cache_concurrency.py +++ b/tests/test_summary_cache_concurrency.py @@ -22,6 +22,8 @@ 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") @@ -59,11 +61,14 @@ def tearDown(self): 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.""" - ws_root = os.path.join(self.tmp.name, "ws") - os.makedirs(ws_root) - workspace_path, workspace_entries = _make_workspace_fixture(ws_root) + workspace_path, workspace_entries = self._setup_workspace("ws") peer_projects = [ {"id": "peer", "name": "Peer", "conversationCount": 1, "lastModified": "x"}, @@ -73,22 +78,14 @@ def test_blocked_build_recheck_returns_peer_cache(self): ] build_started = threading.Event() allow_build_finish = threading.Event() - build_count = 0 - build_count_lock = threading.Lock() def blocked_build() -> tuple[list[dict[str, object]], list[dict[str, object]]]: - nonlocal build_count - with build_count_lock: - build_count += 1 - count = build_count build_started.set() self.assertTrue( - allow_build_finish.wait(timeout=5.0), + allow_build_finish.wait(timeout=_WAIT_TIMEOUT_S), msg="timed out waiting to finish blocked build", ) - if count == 1: - return stale_projects, [] - return peer_projects, [] + return stale_projects, [] errors: list[str] = [] results: list[tuple[list[dict[str, object]], list[dict[str, object]]]] = [] @@ -109,7 +106,7 @@ def thread_a() -> None: def thread_b() -> None: try: self.assertTrue( - build_started.wait(timeout=5.0), + build_started.wait(timeout=_WAIT_TIMEOUT_S), msg="thread B started before thread A entered build_fn", ) results.append( @@ -141,9 +138,7 @@ def thread_b() -> None: self.assertEqual(on_disk.get("projects"), peer_projects) def test_concurrent_get_or_build_returns_consistent_results(self): - ws_root = os.path.join(self.tmp.name, "ws-warm") - os.makedirs(ws_root) - workspace_path, workspace_entries = _make_workspace_fixture(ws_root) + workspace_path, workspace_entries = self._setup_workspace("ws-warm") warm_projects = [ {"id": "warm", "name": "Warm", "conversationCount": 2, "lastModified": "z"}, ] @@ -160,7 +155,7 @@ def test_concurrent_get_or_build_returns_consistent_results(self): def reader() -> None: try: - barrier.wait(timeout=5.0) + barrier.wait(timeout=_WAIT_TIMEOUT_S) hit = get_or_build_cached_projects( workspace_path, workspace_entries, # type: ignore[arg-type] @@ -195,7 +190,7 @@ def test_get_cached_projects_remains_thread_safe(self): def reader() -> None: try: - barrier.wait(timeout=5.0) + barrier.wait(timeout=_WAIT_TIMEOUT_S) hits.append(get_cached_projects(fp)) except Exception as exc: errors.append(str(exc))