Merge pull request 'Keep cold Find Work startup reliable across workers' (#696) from timmy/695-cold-find-work-workers into main

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

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:

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