dennisthiessenandClaude Opus 5
94baa89423
Two review findings on 22bee28, both reproduced before fixing.
[1] Concurrent writes could lose or rewind a row. record_finish did
select-then-insert-or-update, and pipelines are separate scheduler jobs that can
overlap while sharing step ids -- data_collector belongs to all four. Reproduced
both halves: two sessions that SELECT before either INSERTs make the second
commit raise IntegrityError, which _persist_job_run swallows, so the run
silently vanishes; and a later write carrying an OLDER finished_at rewound the
row from 12:00 back to 09:00, dragging the status with it, so the panel would
report a stale outcome as the latest.
Now a single atomic INSERT ... ON CONFLICT (job_name) DO UPDATE, guarded by
WHERE job_run_state.finished_at < excluded.finished_at so an older completion
can never overwrite a newer one. Dialect-specific because prod is Postgres and
tests are SQLite; both support it (SQLite >= 3.24, ours is 3.45). updated_at is
set explicitly, as the model's onupdate hook does not fire for a core upsert,
and get_map now uses populate_existing since core writes leave any
previously-loaded ORM instance stale in the identity map.
[2] Shutdown could dispose the engine underneath a pending write.
scheduler.shutdown(wait=False) returns before APScheduler dispatches its
completion events, and those events are what create persist tasks -- so a single
snapshot of the task set missed writes still to be queued. flush_job_run_persists
now settles briefly for pending callbacks, then drains in a loop until the set
stays empty, with the deadline still bounding total shutdown time. Left
shutdown(wait=False) alone deliberately: waiting would block a deploy restart
behind a long-running scan.
Six regression tests: interleaved first writes, older-never-rewinds,
newer-still-wins, a task queued mid-drain, prompt return when idle, and giving
up rather than hanging shutdown.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
272 lines
10 KiB
Python
272 lines
10 KiB
Python
"""Durable last-run state.
|
|
|
|
Job outcomes lived only in an in-memory dict, so every deploy wiped them and
|
|
Admin → Jobs could only report "Active" with no indication a job had ever run.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app import scheduler as sched
|
|
from app.models.job_run_state import JobRunState
|
|
from app.services import job_run_store
|
|
from tests.conftest import _test_session_factory
|
|
|
|
|
|
@pytest.fixture
|
|
def session_factory():
|
|
"""A real, independently committing session.
|
|
|
|
_persist_job_run opens its own session and commits, which is what production
|
|
does; the shared db_session fixture holds an outer transaction that a commit
|
|
would tear down.
|
|
"""
|
|
return _test_session_factory
|
|
|
|
|
|
async def _rows(session) -> dict[str, JobRunState]:
|
|
result = await session.execute(select(JobRunState))
|
|
return {row.job_name: row for row in result.scalars().all()}
|
|
|
|
|
|
async def _committed_rows() -> dict[str, JobRunState]:
|
|
async with _test_session_factory() as session:
|
|
return await _rows(session)
|
|
|
|
|
|
class TestRecordFinish:
|
|
async def test_inserts_then_updates_one_row_per_job(self, db_session):
|
|
await job_run_store.record_finish(
|
|
db_session,
|
|
"rr_scanner",
|
|
{"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00", "processed": 5, "total": 5},
|
|
)
|
|
await db_session.flush()
|
|
await job_run_store.record_finish(
|
|
db_session,
|
|
"rr_scanner",
|
|
{"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"},
|
|
)
|
|
await db_session.flush()
|
|
|
|
rows = await _rows(db_session)
|
|
assert list(rows) == ["rr_scanner"], "upsert, not append-only history"
|
|
assert rows["rr_scanner"].status == "error"
|
|
assert rows["rr_scanner"].message == "boom"
|
|
|
|
async def test_missing_finish_time_falls_back_to_now(self, db_session):
|
|
await job_run_store.record_finish(db_session, "alerts", {"status": "completed"})
|
|
await db_session.flush()
|
|
assert (await _rows(db_session))["alerts"].finished_at is not None
|
|
|
|
|
|
class TestPipelinePersistence:
|
|
"""_run_pipeline is the only write path for steps: they are plain coroutine
|
|
calls, so they emit no scheduler events for the listener to catch."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _enabled(self, monkeypatch, session_factory):
|
|
async def enabled(db, job_name):
|
|
return True
|
|
|
|
monkeypatch.setattr("app.scheduler.async_session_factory", session_factory)
|
|
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
|
|
|
|
async def test_persists_both_the_step_and_the_orchestrator(self, monkeypatch):
|
|
async def ok_step():
|
|
sched._runtime_finish("rr_scanner", "completed", processed=3, total=3)
|
|
|
|
monkeypatch.setattr(sched, "ok_step", ok_step, raising=False)
|
|
await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "ok_step")])
|
|
|
|
rows = await _committed_rows()
|
|
assert rows["rr_scanner"].status == "completed"
|
|
assert rows["near_close_pipeline"].status == "completed"
|
|
|
|
async def test_a_failing_step_still_records_its_error(self, monkeypatch):
|
|
"""The persist sits after the except that swallows step errors — inside
|
|
it, exactly the runs worth seeing would be skipped."""
|
|
|
|
async def boom():
|
|
sched._runtime_finish("rr_scanner", "error", processed=0, total=1, message="kaboom")
|
|
raise RuntimeError("kaboom")
|
|
|
|
monkeypatch.setattr(sched, "boom", boom, raising=False)
|
|
await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "boom")])
|
|
|
|
rows = await _committed_rows()
|
|
assert rows["rr_scanner"].status == "error"
|
|
assert rows["rr_scanner"].message == "kaboom"
|
|
# The pipeline itself survives a failing step.
|
|
assert rows["near_close_pipeline"].status == "completed"
|
|
|
|
async def test_disabled_pipeline_records_skipped(self, monkeypatch):
|
|
async def disabled(db, job_name):
|
|
return False
|
|
|
|
monkeypatch.setattr("app.scheduler._is_job_enabled", disabled)
|
|
await sched._run_pipeline("daily_pipeline", [])
|
|
|
|
assert (await _committed_rows())["daily_pipeline"].status == "skipped"
|
|
|
|
async def test_persistence_failure_never_breaks_the_pipeline(self, monkeypatch):
|
|
calls: list[str] = []
|
|
|
|
async def exploding_record(db, job_name, runtime):
|
|
calls.append(job_name)
|
|
raise RuntimeError("db down")
|
|
|
|
async def ok_step():
|
|
sched._runtime_finish("rr_scanner", "completed", processed=1, total=1)
|
|
|
|
monkeypatch.setattr(sched.job_run_store, "record_finish", exploding_record)
|
|
monkeypatch.setattr(sched, "ok_step", ok_step, raising=False)
|
|
|
|
await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "ok_step")])
|
|
|
|
assert calls, "persistence was attempted"
|
|
assert sched.get_job_runtime_snapshot("near_close_pipeline")["status"] == "completed"
|
|
|
|
async def test_a_job_that_never_finished_writes_nothing(self):
|
|
sched._runtime_start("event_study", total=1)
|
|
await sched._persist_job_run("event_study")
|
|
assert "event_study" not in await _committed_rows()
|
|
|
|
|
|
class TestListJobsSplitsLiveFromPersisted:
|
|
async def test_a_stale_error_does_not_pin_the_status_chip(self, db_session):
|
|
"""runtime_* must stay live-only: the chip and the rate-limit banner read
|
|
it, so a week-old error there would read as the current state forever."""
|
|
from app.scheduler import configure_scheduler, scheduler
|
|
from app.services.admin_service import list_jobs
|
|
|
|
scheduler.remove_all_jobs()
|
|
configure_scheduler()
|
|
# _job_runtime is module-global and survives across tests; pin the live
|
|
# row to idle so the assertion is about the split, not about ordering.
|
|
sched._job_runtime["rr_scanner"] = sched._idle_runtime()
|
|
await job_run_store.record_finish(
|
|
db_session,
|
|
"rr_scanner",
|
|
{
|
|
"status": "error",
|
|
"finished_at": datetime(2026, 8, 1, tzinfo=timezone.utc).isoformat(),
|
|
"message": "old failure",
|
|
},
|
|
)
|
|
await db_session.flush()
|
|
|
|
job = {j["name"]: j for j in await list_jobs(db_session)}["rr_scanner"]
|
|
assert job["last_run_status"] == "error"
|
|
assert job["last_run_message"] == "old failure"
|
|
assert job["runtime_status"] == "idle"
|
|
assert job["running"] is False
|
|
scheduler.remove_all_jobs()
|
|
|
|
|
|
class TestConcurrentWrites:
|
|
"""Pipelines are separate scheduler jobs that can overlap, and they share
|
|
step ids — data_collector belongs to all four."""
|
|
|
|
async def test_interleaved_first_writes_do_not_collide(self):
|
|
"""Both sessions SELECT before either INSERTs: select-then-insert lost
|
|
this race with an IntegrityError, and the caller swallows it."""
|
|
async with _test_session_factory() as a, _test_session_factory() as b:
|
|
await job_run_store.record_finish(
|
|
a, "data_collector",
|
|
{"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00"},
|
|
)
|
|
await job_run_store.record_finish(
|
|
b, "data_collector",
|
|
{"status": "completed", "finished_at": "2026-08-08T10:00:01+00:00"},
|
|
)
|
|
await a.commit()
|
|
await b.commit() # must not raise
|
|
|
|
rows = await _committed_rows()
|
|
assert rows["data_collector"].status == "completed"
|
|
|
|
async def test_an_older_finish_never_rewinds_the_row(self):
|
|
"""A slower pipeline finishing an older run last must not overwrite a
|
|
newer outcome with a stale one."""
|
|
async with _test_session_factory() as s:
|
|
await job_run_store.record_finish(
|
|
s, "alerts",
|
|
{"status": "completed", "finished_at": "2026-08-08T12:00:00+00:00"},
|
|
)
|
|
await s.commit()
|
|
await job_run_store.record_finish(
|
|
s, "alerts",
|
|
{"status": "error", "finished_at": "2026-08-08T09:00:00+00:00", "message": "stale"},
|
|
)
|
|
await s.commit()
|
|
|
|
row = (await _committed_rows())["alerts"]
|
|
assert row.finished_at.isoformat().startswith("2026-08-08T12:00")
|
|
assert row.status == "completed"
|
|
assert row.message is None
|
|
|
|
async def test_a_newer_finish_still_wins(self):
|
|
async with _test_session_factory() as s:
|
|
await job_run_store.record_finish(
|
|
s, "rr_scanner",
|
|
{"status": "completed", "finished_at": "2026-08-08T09:00:00+00:00"},
|
|
)
|
|
await s.commit()
|
|
await job_run_store.record_finish(
|
|
s, "rr_scanner",
|
|
{"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"},
|
|
)
|
|
await s.commit()
|
|
|
|
row = (await _committed_rows())["rr_scanner"]
|
|
assert row.status == "error"
|
|
assert row.message == "boom"
|
|
|
|
|
|
class TestShutdownDrain:
|
|
async def test_drains_a_task_queued_after_the_flush_starts(self):
|
|
"""scheduler.shutdown(wait=False) returns before APScheduler dispatches
|
|
its completion events, so writes can appear mid-drain. Snapshotting the
|
|
task set once would miss them and dispose the engine underneath."""
|
|
import asyncio
|
|
|
|
done: list[str] = []
|
|
|
|
async def slow_first():
|
|
await asyncio.sleep(0.02)
|
|
done.append("first")
|
|
# Queued only once the first write is already finishing.
|
|
sched._persist_tasks.add(asyncio.get_running_loop().create_task(late()))
|
|
|
|
async def late():
|
|
await asyncio.sleep(0.02)
|
|
done.append("late")
|
|
|
|
sched._persist_tasks.clear()
|
|
sched._persist_tasks.add(asyncio.get_running_loop().create_task(slow_first()))
|
|
|
|
await sched.flush_job_run_persists(timeout=2.0)
|
|
|
|
assert done == ["first", "late"]
|
|
sched._persist_tasks.clear()
|
|
|
|
async def test_returns_promptly_when_there_is_nothing_to_drain(self):
|
|
sched._persist_tasks.clear()
|
|
await sched.flush_job_run_persists(timeout=2.0)
|
|
|
|
async def test_gives_up_rather_than_hanging_shutdown(self):
|
|
import asyncio
|
|
|
|
async def never():
|
|
await asyncio.sleep(30)
|
|
|
|
sched._persist_tasks.clear()
|
|
task = asyncio.get_running_loop().create_task(never())
|
|
sched._persist_tasks.add(task)
|
|
await sched.flush_job_run_persists(timeout=0.15) # returns, does not hang
|
|
task.cancel()
|
|
sched._persist_tasks.clear()
|