peter
45c0ee7d58
What a job last did lived in the API process's memory and was gone at every restart; the audit log kept a run only when it was manual, failed or changed something. Now each run is a job_runs row written at its start and its end in its own session: who started it, its outcome as data, what it spent of the YouTube quota (quota_events.job_run_id) and the lines it logged (log_events.run_id, filterable in the log console). - a run a stopped process left open is marked interrupted at startup - "run now" is a one-shot job on the scheduler, which now always starts (interval jobs only with SCHEDULER_ENABLED); the manual actor is reset after the run, since pool threads are reused - the dashboard's last run comes from job_runs, so it survives a deploy - runs leave the audit log: migration 0109 moves the old job_run rows over (and back on downgrade); the audit page's run folding goes - job_run_retention_days (default 30) prunes old runs in audit_gc
105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
"""The durable history of background runs — the `job_runs` table.
|
|
|
|
A run is written twice: a `running` row when it starts, its outcome when it ends. Each write is its
|
|
OWN short session, committed at once, for the reason `quota.record_usage` gives: the run's own work
|
|
may roll back (its `except: db.rollback()`), and the history of what happened must not go with it —
|
|
nor may this bookkeeping commit the job's half-done session.
|
|
|
|
Only the API process runs jobs (the worker never starts the scheduler), so at its startup nothing can
|
|
still be running: `interrupt_leftovers` closes what a stopped process left open.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import delete, func, select, update
|
|
from sqlalchemy.engine import CursorResult
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import SessionLocal
|
|
from app.models import JobRun, QuotaEvent
|
|
from app.runstatus import RunStatus, RunTrigger
|
|
|
|
|
|
def start(job: str, trigger: RunTrigger, actor_id: int | None, target: str | None = None) -> int:
|
|
"""Open a run's row; returns its id."""
|
|
with SessionLocal() as s:
|
|
row = JobRun(
|
|
job=job,
|
|
trigger=trigger,
|
|
actor_id=actor_id,
|
|
target=target,
|
|
status=RunStatus.RUNNING,
|
|
started_at=datetime.now(timezone.utc),
|
|
)
|
|
s.add(row)
|
|
s.commit()
|
|
return row.id
|
|
|
|
|
|
def finish(
|
|
run_id: int,
|
|
status: RunStatus,
|
|
*,
|
|
summary: str | None = None,
|
|
result: dict | None = None,
|
|
error: str | None = None,
|
|
failure: dict | None = None,
|
|
) -> None:
|
|
"""Close a run with its outcome and what it spent. Conditional on the row still running: a row
|
|
the startup sweep already called interrupted is not rewritten by a thread that outlived it."""
|
|
with SessionLocal() as s:
|
|
units = s.scalar(
|
|
select(func.coalesce(func.sum(QuotaEvent.units), 0)).where(
|
|
QuotaEvent.job_run_id == run_id
|
|
)
|
|
)
|
|
s.execute(
|
|
update(JobRun)
|
|
.where(JobRun.id == run_id, JobRun.status == RunStatus.RUNNING)
|
|
.values(
|
|
status=status,
|
|
finished_at=datetime.now(timezone.utc),
|
|
summary=summary,
|
|
result=result,
|
|
error=error,
|
|
failure=failure,
|
|
quota_units=units or 0,
|
|
)
|
|
)
|
|
s.commit()
|
|
|
|
|
|
def interrupt_leftovers() -> int:
|
|
"""Every row a previous process left `running` → `interrupted`. Called once at startup, before
|
|
the scheduler can open a new run. Returns how many."""
|
|
with SessionLocal() as s:
|
|
res = s.execute(
|
|
update(JobRun)
|
|
.where(JobRun.status == RunStatus.RUNNING)
|
|
.values(status=RunStatus.INTERRUPTED, finished_at=datetime.now(timezone.utc))
|
|
)
|
|
s.commit()
|
|
return res.rowcount if isinstance(res, CursorResult) else 0
|
|
|
|
|
|
def last_runs(db: Session) -> dict[str, JobRun]:
|
|
"""Each job's most recent run (any trigger), by job id."""
|
|
rows = db.execute(
|
|
select(JobRun)
|
|
.distinct(JobRun.job)
|
|
.order_by(JobRun.job, JobRun.started_at.desc(), JobRun.id.desc())
|
|
).scalars()
|
|
return {r.job: r for r in rows}
|
|
|
|
|
|
def gc(db: Session, retention_days: int) -> int:
|
|
"""Delete finished runs older than `retention_days` (0 = keep forever). A run still going is
|
|
never reaped. The caller commits."""
|
|
if retention_days <= 0:
|
|
return 0
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
|
|
res = db.execute(
|
|
delete(JobRun).where(JobRun.started_at < cutoff, JobRun.status != RunStatus.RUNNING)
|
|
)
|
|
return res.rowcount if isinstance(res, CursorResult) else 0
|