From df8385b325c7272a656a271dec6583b57b4b5ebd Mon Sep 17 00:00:00 2001 From: timmy Date: Thu, 13 Aug 2026 01:22:42 +0000 Subject: [PATCH] fix: keep cold Find Work reliable across workers (Closes #695) --- README.md | 5 ++- src/main.py | 10 +++-- tests/test_gitea_work_search.py | 76 +++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index a5900cd..0a057d6 100644 --- a/README.md +++ b/README.md @@ -373,7 +373,10 @@ account and never contain the Gitea token. Find Work uses the same worker-shared `STACKCHAIN_AVAILABLE_ISSUE_SNAPSHOT_DB` overrides `STACKCHAIN_STATE_DIR/available-issue-snapshot.sqlite3`. One expiring lease bounds each catalog scan across the deployment, shared retry metadata prevents worker-by-worker retry -bursts, and confirmed claims are removed from every worker's retained catalog. Releasing +bursts, and confirmed claims are removed from every worker's retained catalog. On a cold +catalog, non-owner workers wait within the bounded foreground deadline for the lease owner +to publish; if that owner releases or abandons the lease, a waiter can take over the scan +within its remaining request budget. Releasing an assignment invalidates the shared catalog so the newly available issue can be discovered by the next authoritative scan. Each bounded opaque revision token includes the store generation, so a token from a different deployment or before replacement of the store diff --git a/src/main.py b/src/main.py index b8b16b5..d628bf3 100644 --- a/src/main.py +++ b/src/main.py @@ -2767,14 +2767,18 @@ async def _available_issue_snapshot() -> tuple[list[dict], bool, bool, bool]: if _available_issue_snapshot_value is not None: return _available_issue_snapshot_value, True, True, False if not local_refresh: - for _ in range(50): + wait_deadline = time.monotonic() + WORK_PAGE_TIMEOUT_SECONDS + while time.monotonic() < wait_deadline: await asyncio.sleep(0.02) shared = await asyncio.to_thread(_available_issue_snapshot_store.load) if shared.items is not None: return shared.items, False, False, False if not shared.refreshing: - break - raise RuntimeError("available issue catalog refresh is owned by another worker") + local_refresh = await _start_available_issue_refresh() + if local_refresh: + break + if not local_refresh: + raise RuntimeError("available issue catalog refresh is owned by another worker") try: return await asyncio.shield(_available_issue_snapshot_task), False, False, False except Exception: diff --git a/tests/test_gitea_work_search.py b/tests/test_gitea_work_search.py index e1f80af..c25d9df 100644 --- a/tests/test_gitea_work_search.py +++ b/tests/test_gitea_work_search.py @@ -354,6 +354,82 @@ async def test_available_issue_endpoint_reuses_catalog_published_by_another_work } +@pytest.mark.anyio +async def test_available_issue_endpoint_waits_for_slow_cold_catalog_from_another_worker( + monkeypatch, tmp_path +): + now = 100.0 + path = tmp_path / "shared-available.sqlite3" + owner_store = AvailableIssueSnapshotStore(path, clock=lambda: now) + worker_store = AvailableIssueSnapshotStore(path, clock=lambda: now) + owner = owner_store.try_acquire_refresh(lease_seconds=6) + monkeypatch.setattr(main, "_available_issue_snapshot_store", worker_store) + + sleep_calls = 0 + + async def publish_after_existing_one_second_window(_delay): + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls == 60: + owner_store.publish( + owner, + items=[{"repository": "stackchain/api", "number": 8}], + ) + + async def must_not_scan(): + raise AssertionError("a valid cross-worker lease must prevent a duplicate scan") + + monkeypatch.setattr(main.asyncio, "sleep", publish_after_existing_one_second_window) + monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", must_not_scan) + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get("/api/v1/available-issues?page=1") + + assert sleep_calls > 50 + assert response.status_code == 200 + assert response.json()["items"] == [ + {"repository": "stackchain/api", "number": 8} + ] + + +@pytest.mark.anyio +async def test_available_issue_endpoint_takes_over_released_cold_refresh( + monkeypatch, tmp_path +): + now = 100.0 + path = tmp_path / "shared-available.sqlite3" + owner_store = AvailableIssueSnapshotStore(path, clock=lambda: now) + worker_store = AvailableIssueSnapshotStore(path, clock=lambda: now) + owner = owner_store.try_acquire_refresh(lease_seconds=6) + monkeypatch.setattr(main, "_available_issue_snapshot_store", worker_store) + + sleep_calls = 0 + scan_calls = 0 + + async def release_failed_owner(_delay): + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls == 3: + owner_store.release_refresh(owner) + + async def available(): + nonlocal scan_calls + scan_calls += 1 + return [{"repository": "stackchain/web", "number": 9}] + + monkeypatch.setattr(main.asyncio, "sleep", release_failed_owner) + monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available) + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get("/api/v1/available-issues?page=1") + + assert response.status_code == 200 + assert response.json()["items"] == [ + {"repository": "stackchain/web", "number": 9} + ] + assert scan_calls == 1 + + @pytest.mark.anyio async def test_available_issue_endpoint_coalesces_cold_scan_and_reuses_it_for_pages(monkeypatch): calls = 0