Keep cold Find Work startup reliable across workers #696
|
|
@ -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_AVAILABLE_ISSUE_SNAPSHOT_DB` overrides
|
||||||
`STACKCHAIN_STATE_DIR/available-issue-snapshot.sqlite3`. One expiring lease bounds each
|
`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
|
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
|
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
|
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
|
store generation, so a token from a different deployment or before replacement of the store
|
||||||
|
|
|
||||||
10
src/main.py
10
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:
|
if _available_issue_snapshot_value is not None:
|
||||||
return _available_issue_snapshot_value, True, True, False
|
return _available_issue_snapshot_value, True, True, False
|
||||||
if not local_refresh:
|
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)
|
await asyncio.sleep(0.02)
|
||||||
shared = await asyncio.to_thread(_available_issue_snapshot_store.load)
|
shared = await asyncio.to_thread(_available_issue_snapshot_store.load)
|
||||||
if shared.items is not None:
|
if shared.items is not None:
|
||||||
return shared.items, False, False, False
|
return shared.items, False, False, False
|
||||||
if not shared.refreshing:
|
if not shared.refreshing:
|
||||||
break
|
local_refresh = await _start_available_issue_refresh()
|
||||||
raise RuntimeError("available issue catalog refresh is owned by another worker")
|
if local_refresh:
|
||||||
|
break
|
||||||
|
if not local_refresh:
|
||||||
|
raise RuntimeError("available issue catalog refresh is owned by another worker")
|
||||||
try:
|
try:
|
||||||
return await asyncio.shield(_available_issue_snapshot_task), False, False, False
|
return await asyncio.shield(_available_issue_snapshot_task), False, False, False
|
||||||
except Exception:
|
except Exception:
|
||||||
|
|
|
||||||
|
|
@ -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
|
@pytest.mark.anyio
|
||||||
async def test_available_issue_endpoint_coalesces_cold_scan_and_reuses_it_for_pages(monkeypatch):
|
async def test_available_issue_endpoint_coalesces_cold_scan_and_reuses_it_for_pages(monkeypatch):
|
||||||
calls = 0
|
calls = 0
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user