From 7f2abfd396ce4d63d92f518cd3ef8e0831779606 Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Tue, 1 Sep 2026 13:54:35 +0200 Subject: [PATCH] fix(api): isolate ticket response build locks --- planfile/api/server.py | 51 +++++++++------ tests/test_sqlite_ticket_index.py | 100 ++++++++++++++++++++++++++++++ 2 files changed, 131 insertions(+), 20 deletions(-) diff --git a/planfile/api/server.py b/planfile/api/server.py index 948f18c..a241eca 100644 --- a/planfile/api/server.py +++ b/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: diff --git a/tests/test_sqlite_ticket_index.py b/tests/test_sqlite_ticket_index.py index 1815cad..ae5aff9 100644 --- a/tests/test_sqlite_ticket_index.py +++ b/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)