refactor: extract shared get_or_build_cached and tidy concurrency tests

Author: clean6378-max-it

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,31 +249,52 @@ 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
 
     built = build_fn()
 
     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,
+    )

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,

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,

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

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))