diff --git a/planfile/api/server.py b/planfile/api/server.py index 4cc5999..ea5b42d 100644 --- a/planfile/api/server.py +++ b/planfile/api/server.py @@ -12,10 +12,10 @@ import sqlite3 import time from collections import deque -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, contextmanager from datetime import UTC, datetime from pathlib import Path -from threading import RLock +from threading import BoundedSemaphore, RLock from typing import Any, Literal from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -360,7 +360,11 @@ 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_RESPONSE_BUILD_LOCKS = { + workload: tuple(RLock() for _ in range(32)) + for workload in ("full", "operational", "summary", "archive-operational") +} +_TICKET_ARCHIVE_OPERATIONAL_BUILD_SLOTS = BoundedSemaphore(4) _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 @@ -479,9 +483,56 @@ def _ticket_list_response_cached_bytes() -> int: 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) - ] + _, sprint, filters, _, limit, view = query_key + workload = ( + "archive-operational" + if _ticket_list_is_heavy_archive( + sprint=sprint, + filters=filters, + limit=limit, + view=view, + ) + else str(view) + ) + locks = _TICKET_LIST_RESPONSE_BUILD_LOCKS[workload] + return locks[hash(query_key) % len(locks)] + + +def _ticket_list_is_heavy_archive( + *, + sprint: str, + filters, + limit: int | None, + view: str, +) -> bool: + return ( + sprint == "all" + and view == "operational" + and not filters + and limit is not None + and limit >= 500 + ) + + +@contextmanager +def _ticket_list_workload_slot( + *, + sprint: str, + filters: dict, + limit: int | None, + view: Literal["full", "operational", "summary"], +): + """Bound CPU-heavy archive projections while reserving queue-read capacity.""" + if _ticket_list_is_heavy_archive( + sprint=sprint, + filters=filters, + limit=limit, + view=view, + ): + with _TICKET_ARCHIVE_OPERATIONAL_BUILD_SLOTS: + yield + return + yield def _cache_ticket_list_response( @@ -585,7 +636,12 @@ def _ticket_list_response( # a burst of websocket-driven dashboard refreshes builds one 5+ MB response, # not one copy per browser tab. 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_build_lock(query_key), _ticket_list_workload_slot( + sprint=sprint, + filters=filters, + limit=limit, + view=view, + ): 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: diff --git a/tests/test_sqlite_ticket_index.py b/tests/test_sqlite_ticket_index.py index ff28889..fc3332a 100644 --- a/tests/test_sqlite_ticket_index.py +++ b/tests/test_sqlite_ticket_index.py @@ -307,6 +307,100 @@ def response(): assert calls == 1 +def test_slow_archive_operational_build_never_blocks_summary_status( + tmp_path, monkeypatch +): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + pf.create_ticket(name="Workload isolation") + 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_operational_response + + def operational_response(**kwargs): + if not kwargs["filters"]: + slow_started.set() + assert finish_slow.wait(timeout=5) + return original(**kwargs) + + monkeypatch.setattr( + pf.store, + "indexed_ticket_operational_response", + operational_response, + ) + + def slow_response(): + return server._ticket_list_response( + pf, + sprint="all", + filters={}, + offset=0, + limit=1000, + view="operational", + ) + + def status_response(): + return server._ticket_list_response( + pf, + sprint="all", + filters={"source": "status"}, + offset=0, + limit=1, + view="summary", + ) + + def filtered_operational_response(): + return server._ticket_list_response( + pf, + sprint="all", + filters={"status": "open"}, + offset=0, + limit=500, + view="operational", + ) + + with ThreadPoolExecutor(max_workers=3) as pool: + slow = pool.submit(slow_response) + assert slow_started.wait(timeout=5) + status = pool.submit(status_response) + filtered = pool.submit(filtered_operational_response) + try: + assert status.result(timeout=1).status_code == 200 + assert filtered.result(timeout=1).status_code == 200 + finally: + finish_slow.set() + assert slow.result(timeout=5).status_code == 200 + + +def test_heavy_archive_operational_builds_have_bounded_concurrency(): + active = 0 + maximum = 0 + lock = threading.Lock() + + def build(): + nonlocal active, maximum + with server._ticket_list_workload_slot( + sprint="all", + filters={}, + limit=1000, + view="operational", + ): + with lock: + active += 1 + maximum = max(maximum, active) + time.sleep(0.05) + with lock: + active -= 1 + + with ThreadPoolExecutor(max_workers=12) as pool: + list(pool.map(lambda _number: build(), range(12))) + + assert maximum == 4 + + def test_sqlite_index_serves_get_without_reparsing_sprint_files(tmp_path, monkeypatch): pf = Planfile(str(tmp_path)) _disable_archive(pf)