fix: keep cold Find Work reliable across workers (Closes #695)
This commit is contained in:
1 parent
47c6412a12
commit
df8385b325
3 files changed
+87
-4
No files matched your search
@@ -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
|
||||
|
||||
Reference in new issue
Block a user