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>
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
"""Single source for JobRunState reads/writes.
|
|
|
|
Mirrors ``settings_store``: ``record_finish`` never commits — the caller owns
|
|
the transaction — and reads are batched so the admin listing stays one query.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Iterable
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.job_run_state import JobRunState
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _as_datetime(value: object) -> datetime | None:
|
|
"""Runtime snapshots carry ISO strings; the column wants a datetime."""
|
|
if isinstance(value, datetime):
|
|
return value
|
|
if isinstance(value, str) and value:
|
|
try:
|
|
return datetime.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
async def get_map(db: AsyncSession, job_names: Iterable[str]) -> dict[str, JobRunState]:
|
|
"""Return {job_name: row} for the given jobs that have ever finished.
|
|
|
|
``populate_existing`` because rows are written by core upserts, which leave
|
|
any previously-loaded ORM instance in the identity map stale.
|
|
"""
|
|
result = await db.execute(
|
|
select(JobRunState)
|
|
.where(JobRunState.job_name.in_(list(job_names)))
|
|
.execution_options(populate_existing=True)
|
|
)
|
|
return {row.job_name: row for row in result.scalars().all()}
|
|
|
|
|
|
def _insert_for(db: AsyncSession):
|
|
"""ON CONFLICT is dialect-specific; prod is Postgres, tests are SQLite."""
|
|
dialect = db.get_bind().dialect.name
|
|
return pg_insert if dialect == "postgresql" else sqlite_insert
|
|
|
|
|
|
async def record_finish(db: AsyncSession, job_name: str, runtime: dict) -> None:
|
|
"""Upsert the last-run row from a scheduler runtime snapshot.
|
|
|
|
Atomic, and newer-wins. Select-then-insert loses races that really happen
|
|
here: pipelines are separate scheduler jobs that can overlap, and they share
|
|
step ids -- data_collector belongs to all four. Two of them finishing that
|
|
step together would both see no row and both insert, and the loser's
|
|
IntegrityError is swallowed by the caller, so the run silently vanishes.
|
|
|
|
The ``where`` guard is the other half: without it a slower pipeline
|
|
finishing an *older* run last would rewind finished_at and the status with
|
|
it, so the panel would report a stale outcome as the latest one.
|
|
"""
|
|
finished_at = _as_datetime(runtime.get("finished_at")) or datetime.now(timezone.utc)
|
|
message = runtime.get("message")
|
|
now = datetime.now(timezone.utc)
|
|
values = {
|
|
"job_name": job_name,
|
|
"status": str(runtime.get("status") or "completed"),
|
|
"started_at": _as_datetime(runtime.get("started_at")),
|
|
"finished_at": finished_at,
|
|
"processed": runtime.get("processed"),
|
|
"total": runtime.get("total"),
|
|
"message": str(message)[:4000] if message else None,
|
|
# Set explicitly: the model's onupdate hook does not fire for a core
|
|
# INSERT ... ON CONFLICT DO UPDATE.
|
|
"updated_at": now,
|
|
}
|
|
|
|
statement = _insert_for(db)(JobRunState).values(**values)
|
|
await db.execute(
|
|
statement.on_conflict_do_update(
|
|
index_elements=[JobRunState.job_name],
|
|
set_={key: statement.excluded[key] for key in values if key != "job_name"},
|
|
where=JobRunState.finished_at < statement.excluded.finished_at,
|
|
)
|
|
)
|