perf: share Find Work catalog across workers (#453)

tests/test_available_issue_snapshot_store.py

@@ -0,0 +1,90 @@
+import threading
+import os
+
+from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
+
+
+def test_independent_workers_allow_only_one_catalog_refresh(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    first = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    second = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    barrier = threading.Barrier(2)
+    results = []
+
+    def acquire(store):
+        barrier.wait()
+        results.append(store.try_acquire_refresh(lease_seconds=5))
+
+    threads = [threading.Thread(target=acquire, args=(store,)) for store in (first, second)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert sum(owner is not None for owner in results) == 1
+
+
+def test_published_catalog_and_metadata_are_visible_to_another_worker(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    owner = writer.try_acquire_refresh(lease_seconds=5)
+
+    writer.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
+
+    state = reader.load()
+    assert state.items == [{"repository": "stackchain/api", "number": 7}]
+    assert state.created_at == 100.0
+    assert state.retry_at is None
+    assert state.refreshing is False
+
+
+def test_failed_refresh_backoff_is_shared_and_lease_is_released(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    owner = writer.try_acquire_refresh(lease_seconds=5)
+
+    writer.fail_refresh(owner, retry_at=105.0)
+
+    state = reader.load()
+    assert state.retry_at == 105.0
+    assert state.refreshing is False
+    assert reader.try_acquire_refresh(lease_seconds=5) is not None
+
+
+def test_confirmed_claim_is_removed_for_every_worker(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    owner = writer.try_acquire_refresh(lease_seconds=5)
+    writer.publish(owner, items=[
+        {"repository": "stackchain/api", "number": 7},
+        {"repository": "stackchain/web", "number": 8},
+    ])
+
+    writer.remove_claimed("stackchain/api", 7)
+
+    assert reader.load().items == [{"repository": "stackchain/web", "number": 8}]
+
+
+def test_invalidation_clears_catalog_and_claim_filters(tmp_path):
+    store = AvailableIssueSnapshotStore(tmp_path / "available.sqlite3", clock=lambda: 100.0)
+    owner = store.try_acquire_refresh(lease_seconds=5)
+    store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
+    store.remove_claimed("stackchain/api", 7)
+
+    store.invalidate()
+    owner = store.try_acquire_refresh(lease_seconds=5)
+    store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
+
+    assert store.load().items == [{"repository": "stackchain/api", "number": 7}]
+
+
+def test_catalog_store_is_private(tmp_path):
+    path = tmp_path / "state" / "available.sqlite3"
+
+    AvailableIssueSnapshotStore(path)
+
+    assert os.stat(path.parent).st_mode & 0o777 == 0o700
+    assert os.stat(path).st_mode & 0o777 == 0o600

tests/test_gitea_work_search.py

@@ -6,11 +6,18 @@ import httpx
 
 from src import gitea_proxy
 from src import main
+from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
 
 
 @pytest.fixture(autouse=True)
-def reset_available_issue_snapshot():
+def reset_available_issue_snapshot(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        main,
+        "_available_issue_snapshot_store",
+        AvailableIssueSnapshotStore(tmp_path / "available-issues.sqlite3"),
+    )
     main._available_issue_snapshot_task = None
+    main._available_issue_snapshot_lock = asyncio.Lock()
     main._available_issue_snapshot_value = None
     main._available_issue_snapshot_created_at = None
     main._available_issue_snapshot_retry_at = None
@@ -19,6 +26,7 @@ def reset_available_issue_snapshot():
     if task is not None and not task.done():
         task.cancel()
     main._available_issue_snapshot_task = None
+    main._available_issue_snapshot_lock = asyncio.Lock()
     main._available_issue_snapshot_value = None
     main._available_issue_snapshot_created_at = None
     main._available_issue_snapshot_retry_at = None
@@ -207,6 +215,31 @@ async def test_available_issue_endpoint_is_bounded_retryable_and_no_store(monkey
     assert calls == [True]
 
 
+@pytest.mark.anyio
+async def test_available_issue_endpoint_reuses_catalog_published_by_another_worker(
+    monkeypatch, tmp_path
+):
+    store = AvailableIssueSnapshotStore(tmp_path / "available.sqlite3", clock=lambda: 100.0)
+    owner = store.try_acquire_refresh(lease_seconds=5)
+    store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
+    monkeypatch.setattr(main, "_available_issue_snapshot_store", store, raising=False)
+    monkeypatch.setattr(main.time, "time", lambda: 100.0)
+
+    async def must_not_scan():
+        raise AssertionError("fresh shared catalog should avoid an upstream scan")
+
+    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 response.status_code == 200
+    assert response.json() == {
+        "items": [{"repository": "stackchain/api", "number": 7}],
+        "page": 1, "total": 1, "has_more": False,
+    }
+
+
 @pytest.mark.anyio
 async def test_available_issue_endpoint_coalesces_cold_scan_and_reuses_it_for_pages(monkeypatch):
     calls = 0