fix(api): isolate ticket response build locks

planfile/api/server.py

@@ -360,6 +360,7 @@ def _validate_completion_receipt(receipt: dict[str, Any] | None, ticket_id: str)
 _TICKET_LIST_RESPONSE_CACHE: dict[tuple, tuple[bytes, int, int]] = {}
 _TICKET_LIST_RESPONSE_CACHE_LIMIT = 4
 _TICKET_LIST_RESPONSE_CACHE_LOCK = RLock()
+_TICKET_LIST_RESPONSE_BUILD_LOCKS = tuple(RLock() for _ in range(32))
 _TICKET_LIST_LATEST: dict[tuple, tuple[float, bytes, int, int]] = {}
 _TICKET_LIST_RESPONSE_CACHE_DEFAULT_BYTES = 256 * 1024 * 1024
 _TICKET_LIST_RESPONSE_CACHE_MIN_BYTES = 1024 * 1024
@@ -474,6 +475,13 @@ def _ticket_list_response_cached_bytes() -> int:
     return sum(len(body) for body in bodies.values())
 
 
+def _ticket_list_response_build_lock(query_key: tuple) -> RLock:
+    """Coalesce identical cache misses without blocking unrelated queue views."""
+    return _TICKET_LIST_RESPONSE_BUILD_LOCKS[
+        hash(query_key) % len(_TICKET_LIST_RESPONSE_BUILD_LOCKS)
+    ]
+
+
 def _cache_ticket_list_response(
     *,
     query_key: tuple,
@@ -483,22 +491,23 @@ def _cache_ticket_list_response(
     count: int,
 ) -> None:
     """Retain only bounded ticket projections; full archives can be hundreds of MB."""
-    for existing_key in tuple(_TICKET_LIST_RESPONSE_CACHE):
-        if existing_key[:-1] == query_key:
-            _TICKET_LIST_RESPONSE_CACHE.pop(existing_key, None)
-    _TICKET_LIST_LATEST.pop(query_key, None)
-
-    byte_limit = _ticket_list_response_cache_byte_limit()
-    if len(body) > byte_limit:
-        return
-    if (
-        len(_TICKET_LIST_RESPONSE_CACHE) >= _TICKET_LIST_RESPONSE_CACHE_LIMIT
-        or _ticket_list_response_cached_bytes() + len(body) > byte_limit
-    ):
-        _TICKET_LIST_RESPONSE_CACHE.clear()
-        _TICKET_LIST_LATEST.clear()
-    _TICKET_LIST_RESPONSE_CACHE[versioned_key] = (body, total, count)
-    _TICKET_LIST_LATEST[query_key] = (time.monotonic(), body, total, count)
+    with _TICKET_LIST_RESPONSE_CACHE_LOCK:
+        for existing_key in tuple(_TICKET_LIST_RESPONSE_CACHE):
+            if existing_key[:-1] == query_key:
+                _TICKET_LIST_RESPONSE_CACHE.pop(existing_key, None)
+        _TICKET_LIST_LATEST.pop(query_key, None)
+
+        byte_limit = _ticket_list_response_cache_byte_limit()
+        if len(body) > byte_limit:
+            return
+        if (
+            len(_TICKET_LIST_RESPONSE_CACHE) >= _TICKET_LIST_RESPONSE_CACHE_LIMIT
+            or _ticket_list_response_cached_bytes() + len(body) > byte_limit
+        ):
+            _TICKET_LIST_RESPONSE_CACHE.clear()
+            _TICKET_LIST_LATEST.clear()
+        _TICKET_LIST_RESPONSE_CACHE[versioned_key] = (body, total, count)
+        _TICKET_LIST_LATEST[query_key] = (time.monotonic(), body, total, count)
 
 
 def _bounded_stale_index_response(
@@ -565,9 +574,10 @@ def _ticket_list_response(
     # FastAPI runs this sync endpoint in a worker pool. Serialize cache misses so
     # a burst of websocket-driven dashboard refreshes builds one 5+ MB response,
     # not one copy per browser tab.
-    with _TICKET_LIST_RESPONSE_CACHE_LOCK:
-        query_key = (str(pf.store.project_dir), sprint, tuple(sorted(filters.items())), offset, limit, view)
-        latest = _TICKET_LIST_LATEST.get(query_key)
+    query_key = (str(pf.store.project_dir), sprint, tuple(sorted(filters.items())), offset, limit, view)
+    with _ticket_list_response_build_lock(query_key):
+        with _TICKET_LIST_RESPONSE_CACHE_LOCK:
+            latest = _TICKET_LIST_LATEST.get(query_key)
         if allow_stale and latest is not None and time.monotonic() - latest[0] < _DASHBOARD_STALE_WINDOW_SECONDS:
             _, body, total, count = latest
             return Response(
@@ -582,7 +592,8 @@ def _ticket_list_response(
             )
         signature = _ticket_snapshot_signature(pf, sprint)
         key = query_key + (signature,)
-        cached = _TICKET_LIST_RESPONSE_CACHE.get(key)
+        with _TICKET_LIST_RESPONSE_CACHE_LOCK:
+            cached = _TICKET_LIST_RESPONSE_CACHE.get(key)
         if cached is not None:
             body, total, count = cached
         else:

tests/test_sqlite_ticket_index.py

@@ -117,6 +117,106 @@ def delayed_records():
     assert replacement == [{"id": "PLF-2", "name": "Replacement"}]
 
 
+def test_unrelated_ticket_queries_do_not_share_a_response_build_lock(
+    tmp_path, monkeypatch
+):
+    pf = Planfile(str(tmp_path))
+    _disable_archive(pf)
+    pf.create_ticket(name="Concurrent query")
+    pf.store.configure_ticket_index(True)
+    server._TICKET_LIST_RESPONSE_CACHE.clear()
+    server._TICKET_LIST_LATEST.clear()
+    slow_started = threading.Event()
+    finish_slow = threading.Event()
+    original = pf.store.indexed_ticket_summaries
+
+    def summaries(**kwargs):
+        if kwargs["filters"].get("source") == "slow":
+            slow_started.set()
+            assert finish_slow.wait(timeout=5)
+        return original(**kwargs)
+
+    monkeypatch.setattr(pf.store, "indexed_ticket_summaries", summaries)
+    slow_filters = {"source": "slow"}
+    fast_number = 0
+    while True:
+        fast_filters = {"source": f"fast-{fast_number}"}
+        slow_key = (str(pf.store.project_dir), "all", tuple(slow_filters.items()), 0, 1, "summary")
+        fast_key = (str(pf.store.project_dir), "all", tuple(fast_filters.items()), 0, 1, "summary")
+        if server._ticket_list_response_build_lock(slow_key) is not server._ticket_list_response_build_lock(fast_key):
+            break
+        fast_number += 1
+
+    def response(filters):
+        return server._ticket_list_response(
+            pf,
+            sprint="all",
+            filters=filters,
+            offset=0,
+            limit=1,
+            view="summary",
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        slow = pool.submit(response, slow_filters)
+        assert slow_started.wait(timeout=5)
+        fast = pool.submit(response, fast_filters)
+        try:
+            assert fast.result(timeout=1).status_code == 200
+        finally:
+            finish_slow.set()
+        assert slow.result(timeout=5).status_code == 200
+
+
+def test_identical_ticket_queries_still_share_one_response_build(tmp_path, monkeypatch):
+    pf = Planfile(str(tmp_path))
+    _disable_archive(pf)
+    pf.create_ticket(name="Shared query")
+    pf.store.configure_ticket_index(True)
+    server._TICKET_LIST_RESPONSE_CACHE.clear()
+    server._TICKET_LIST_LATEST.clear()
+    first_started = threading.Event()
+    finish_first = threading.Event()
+    calls = 0
+    calls_lock = threading.Lock()
+    original = pf.store.indexed_ticket_summaries
+
+    def summaries(**kwargs):
+        nonlocal calls
+        with calls_lock:
+            calls += 1
+            current_call = calls
+        if current_call == 1:
+            first_started.set()
+            assert finish_first.wait(timeout=5)
+        return original(**kwargs)
+
+    monkeypatch.setattr(pf.store, "indexed_ticket_summaries", summaries)
+
+    def response():
+        return server._ticket_list_response(
+            pf,
+            sprint="all",
+            filters={"source": "shared"},
+            offset=0,
+            limit=1,
+            view="summary",
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        first = pool.submit(response)
+        assert first_started.wait(timeout=5)
+        second = pool.submit(response)
+        time.sleep(0.1)
+        try:
+            assert calls == 1
+        finally:
+            finish_first.set()
+        assert first.result(timeout=5).status_code == 200
+        assert second.result(timeout=5).status_code == 200
+    assert calls == 1
+
+
 def test_sqlite_index_serves_get_without_reparsing_sprint_files(tmp_path, monkeypatch):
     pf = Planfile(str(tmp_path))
     _disable_archive(pf)