diff --git a/alembic/versions/031_job_run_state.py b/alembic/versions/031_job_run_state.py new file mode 100644 index 0000000..6f15e0b --- /dev/null +++ b/alembic/versions/031_job_run_state.py @@ -0,0 +1,51 @@ +"""Durable last-run state per scheduled job + +Revision ID: 031 +Revises: 030 +Create Date: 2026-08-08 00:00:00.000000 + +Job run state lived only in an in-memory dict in ``app.scheduler``, so every +process restart wiped it. Admin → Jobs could then only report "Active" with no +indication of whether a job had ever run, or how it ended — which is exactly +the information an operator opens that page for. + +One row per job, upserted on ``job_name``. Not history: ``system_events`` +already grows unbounded with no retention job, and a second append-only +operational table would repeat that debt. + +The table starts empty; each job populates its row the next time it finishes. +No backfill from ``system_events`` — that table only records warning/error +outcomes and uses a different status vocabulary, so seeding from it would +invent successful runs that never happened. +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "031" +down_revision: Union[str, None] = "030" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "job_run_state", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("job_name", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("processed", sa.Integer(), nullable=True), + sa.Column("total", sa.Integer(), nullable=True), + sa.Column("message", sa.Text(), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("job_name", name="uq_job_run_state_job_name"), + ) + + +def downgrade() -> None: + op.drop_table("job_run_state") diff --git a/app/main.py b/app/main.py index c8064ad..e945849 100644 --- a/app/main.py +++ b/app/main.py @@ -21,7 +21,12 @@ from app.config import settings from app.database import async_session_factory, engine from app.middleware import register_exception_handlers from app.models.user import User -from app.scheduler import configure_scheduler, load_schedule_config, scheduler +from app.scheduler import ( + configure_scheduler, + flush_job_run_persists, + load_schedule_config, + scheduler, +) from app.routers.admin import router as admin_router from app.routers.auth import router as auth_router from app.routers.health import router as health_router @@ -91,6 +96,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: scheduler.shutdown(wait=False) logger.info("Scheduler stopped") + # Drain detached last-run writes before the engine goes away, or a job that + # finished during shutdown loses the row it just wrote. + await flush_job_run_persists() await engine.dispose() logger.info("Shutting down") diff --git a/app/models/__init__.py b/app/models/__init__.py index 198dfb0..bd003f7 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -18,6 +18,7 @@ from app.models.benchmark_price import BenchmarkPrice from app.models.signal_context_snapshot import SignalContextSnapshot from app.models.system_event import SystemEvent from app.models.sec_filing_gap import SecFilingGap +from app.models.job_run_state import JobRunState __all__ = [ "Ticker", @@ -42,4 +43,5 @@ __all__ = [ "SignalContextSnapshot", "SystemEvent", "SecFilingGap", + "JobRunState", ] diff --git a/app/models/job_run_state.py b/app/models/job_run_state.py new file mode 100644 index 0000000..f749bf5 --- /dev/null +++ b/app/models/job_run_state.py @@ -0,0 +1,37 @@ +from datetime import datetime + +from sqlalchemy import DateTime, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class JobRunState(Base): + """How each scheduled job last finished. One row per job, overwritten. + + The scheduler's ``_job_runtime`` dict is the live view and is deliberately + in-memory, but it is also wiped by every process restart -- so after a deploy + Admin → Jobs could only say "Active" with no indication of whether a job had + ever run. This is the durable half. + + Deliberately not history: ``system_events`` already grows without a reaper, + and a second append-only operational table would repeat that. Rows are + upserted on ``job_name``; adding history later is purely additive. + """ + + __tablename__ = "job_run_state" + + id: Mapped[int] = mapped_column(primary_key=True) + job_name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + # Scheduler vocabulary: completed | skipped | error | rate_limited | deferred. + # Distinct from data_import_runs' statuses, which is one reason this is its + # own table rather than a widened column there. + status: Mapped[str] = mapped_column(String(32), nullable=False) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + processed: Mapped[int | None] = mapped_column(Integer, nullable=True) + total: Mapped[int | None] = mapped_column(Integer, nullable=True) + message: Mapped[str | None] = mapped_column(Text, nullable=True) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False + ) diff --git a/app/scheduler.py b/app/scheduler.py index a470d3c..5868985 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -33,6 +33,7 @@ from app.models.ticker import Ticker from app.exceptions import ProviderError from app.providers.alpaca import AlpacaOHLCVProvider from app.providers.protocol import SentimentData +from app.services import job_run_store from app.services import ( ingestion_service, pipeline_run, @@ -87,6 +88,19 @@ scheduler = AsyncIOScheduler( ) +def _on_job_finished(event: object) -> None: + """Persist the run, then re-pause the job if it only runs on demand. + + Covers every job APScheduler fires itself, including manual triggers. + Pipeline *steps* are invoked as plain coroutines and emit no events, so + ``_run_pipeline`` persists those directly. + """ + job_id = getattr(event, "job_id", None) + if job_id: + _schedule_persist(job_id) + _repause_after_manual_run(event) + + def _repause_after_manual_run(event: object) -> None: """Re-pause a job that only ever runs on demand, once its run finishes. @@ -112,7 +126,7 @@ def _repause_after_manual_run(event: object) -> None: logger.debug("Could not re-pause %s after its run", job_id, exc_info=True) -scheduler.add_listener(_repause_after_manual_run, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) +scheduler.add_listener(_on_job_finished, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) # Track last successful ticker per job for rate-limit resume _last_successful: dict[str, str | None] = { @@ -308,6 +322,44 @@ def _runtime_finish( pass +async def _persist_job_run(job_name: str) -> None: + """Write a job's finished runtime row to the durable last-run table. + + Never raises: a persistence failure must not break the pipeline that was + otherwise successful. The in-memory row stays authoritative for live state. + """ + runtime = _job_runtime.get(job_name) + if not runtime or runtime.get("running") or not runtime.get("finished_at"): + return + try: + async with async_session_factory() as db: + await job_run_store.record_finish(db, job_name, runtime) + await db.commit() + except Exception: + logger.exception("Could not persist last-run state for %s", job_name) + + +# Detached persists are kept referenced: a bare create_task result can be +# garbage-collected mid-flight, and the shutdown drain needs something to await. +_persist_tasks: set[asyncio.Task] = set() + + +def _schedule_persist(job_name: str) -> None: + try: + task = asyncio.get_running_loop().create_task(_persist_job_run(job_name)) + except RuntimeError: # no loop (sync context / tests) — nothing to persist + return + _persist_tasks.add(task) + task.add_done_callback(_persist_tasks.discard) + + +async def flush_job_run_persists(timeout: float = 5.0) -> None: + """Await in-flight last-run writes. Called from the app's shutdown path.""" + if not _persist_tasks: + return + await asyncio.wait(set(_persist_tasks), timeout=timeout) + + def get_job_runtime_snapshot(job_name: str | None = None) -> dict[str, dict[str, object]] | dict[str, object]: if job_name is not None: return dict(_job_runtime.get(job_name, {})) @@ -1346,6 +1398,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None: if not await _is_job_enabled(db, job_name): _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled") _runtime_finish(job_name, "skipped", processed=0, total=0, message="Disabled") + await _persist_job_run(job_name) return total = len(steps) @@ -1361,6 +1414,11 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None: await funcs[func_name]() except Exception: logger.exception("%s step %s failed", job_name, step_name) + # Outside the except on purpose: the step's own _runtime_finish has + # already recorded its outcome, so persisting here captures failures + # too. Steps are plain coroutine calls and fire no scheduler events, + # so the listener cannot see them -- this is their only write path. + await _persist_job_run(step_name) done += 1 _runtime_finish(job_name, "completed", processed=done, total=total, message="Pipeline complete") _log_event(logging.INFO, "job_complete", job=job_name) @@ -1369,6 +1427,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None: _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc)) finally: pipeline_run.release(token) + await _persist_job_run(job_name) async def run_daily_pipeline() -> None: diff --git a/app/services/admin_service.py b/app/services/admin_service.py index f86ee9b..9458f2e 100644 --- a/app/services/admin_service.py +++ b/app/services/admin_service.py @@ -18,7 +18,7 @@ from app.models.settings import SystemSetting from app.models.ticker import Ticker from app.models.trade_setup import TradeSetup from app.models.user import User -from app.services import settings_store +from app.services import job_run_store, settings_store logger = logging.getLogger(__name__) @@ -689,11 +689,13 @@ async def list_jobs(db: AsyncSession) -> list[dict]: name: flags.get(f"job_{name}_enabled", "true") == "true" for name in VALID_JOB_NAMES } + last_runs = await job_run_store.get_map(db, visible) jobs_out = [] for name in visible: job = scheduler.get_job(name) runtime = get_job_runtime_snapshot(name) + last = last_runs.get(name) jobs_out.append({ "name": name, @@ -717,6 +719,14 @@ async def list_jobs(db: AsyncSession) -> list[dict]: "runtime_started_at": runtime.get("started_at"), "runtime_finished_at": runtime.get("finished_at"), "runtime_message": runtime.get("message"), + # Survives restarts, unlike runtime_*. Reported separately so the + # status chip keeps meaning "state now" rather than "last outcome, + # forever" -- an error a week ago must not read as Inactive today. + "last_run_at": last.finished_at.isoformat() if last else None, + "last_run_status": last.status if last else None, + "last_run_message": last.message if last else None, + "last_run_processed": last.processed if last else None, + "last_run_total": last.total if last else None, **_next_run_fields(scheduler, name, enabled_map), }) diff --git a/app/services/job_run_store.py b/app/services/job_run_store.py new file mode 100644 index 0000000..5a59534 --- /dev/null +++ b/app/services/job_run_store.py @@ -0,0 +1,70 @@ +"""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.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.""" + result = await db.execute( + select(JobRunState).where(JobRunState.job_name.in_(list(job_names))) + ) + return {row.job_name: row for row in result.scalars().all()} + + +async def record_finish(db: AsyncSession, job_name: str, runtime: dict) -> JobRunState: + """Upsert the last-run row from a scheduler runtime snapshot. + + Select-then-update-or-insert rather than a dialect-specific upsert, matching + ``settings_store.upsert_setting`` — tests run on SQLite, prod on Postgres. + """ + existing = await db.execute( + select(JobRunState).where(JobRunState.job_name == job_name) + ) + row = existing.scalar_one_or_none() + + finished_at = _as_datetime(runtime.get("finished_at")) or datetime.now(timezone.utc) + status = str(runtime.get("status") or "completed") + message = runtime.get("message") + values = { + "status": status, + "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, + } + + if row is None: + row = JobRunState(job_name=job_name, **values) + db.add(row) + else: + for field, value in values.items(): + setattr(row, field, value) + return row diff --git a/frontend/src/api/admin.ts b/frontend/src/api/admin.ts index 20993a4..651972c 100644 --- a/frontend/src/api/admin.ts +++ b/frontend/src/api/admin.ts @@ -207,14 +207,28 @@ export function backfillTickerNames() { } // Jobs +export type JobCategory = 'pipeline' | 'pipeline_step' | 'scheduled' | 'manual'; +export type NextRunSource = 'own_schedule' | 'via_pipeline' | 'manual_only'; + export interface JobStatus { name: string; label: string; enabled: boolean; - next_run_at: string | null; - via_pipeline?: boolean; registered: boolean; + category?: JobCategory; + /** Server-assigned ordering; the payload already arrives grouped by it. */ + sort_order?: [number, number]; + /** Parent pipelines for a step. Many-to-many: data_collector runs in all four. */ + pipelines?: string[]; + /** Step names, for a pipeline row. */ + steps?: string[]; + next_run_at: string | null; + next_run_source?: NextRunSource; + /** For a step: the soonest enabled parent's next run, and which parent. */ + via_next_run_at?: string | null; + via_next_run_job?: string | null; running?: boolean; + /** runtime_* is live, in-memory state only — it resets when the app restarts. */ runtime_status?: string | null; runtime_processed?: number | null; runtime_total?: number | null; @@ -223,6 +237,13 @@ export interface JobStatus { runtime_started_at?: string | null; runtime_finished_at?: string | null; runtime_message?: string | null; + /** last_run_* is persisted and survives restarts. Kept separate from + * runtime_* so a stale error cannot pin the status chip or the banner. */ + last_run_at?: string | null; + last_run_status?: string | null; + last_run_message?: string | null; + last_run_processed?: number | null; + last_run_total?: number | null; } export interface TriggerJobResponse { diff --git a/frontend/src/components/admin/JobControls.tsx b/frontend/src/components/admin/JobControls.tsx index b063f2e..3bbe4b8 100644 --- a/frontend/src/components/admin/JobControls.tsx +++ b/frontend/src/components/admin/JobControls.tsx @@ -1,4 +1,5 @@ import { useJobs, useToggleJob, useTriggerJob } from '../../hooks/useAdmin'; +import type { JobStatus } from '../../api/admin'; import { SkeletonTable } from '../ui/Skeleton'; function formatNextRun(iso: string | null): string { @@ -29,10 +30,34 @@ function lastRunColor(status: string | null | undefined): string { return 'text-gray-500'; } +/** One consistent answer per job: its own timer, its parent's, or "manual only". + * A step has no schedule of its own, so reporting one was the original bug. */ +function NextRun({ job, labels }: { job: JobStatus; labels: Record }) { + const muted = 'text-[11px] text-gray-500'; + if (job.next_run_source === 'manual_only') { + return manual only; + } + if (job.next_run_source === 'via_pipeline') { + if (!job.via_next_run_at || !job.via_next_run_job) { + return runs via pipeline; + } + return ( + + Next via {labels[job.via_next_run_job] ?? job.via_next_run_job}{' '} + {formatNextRun(job.via_next_run_at)} + + ); + } + if (!job.next_run_at) return null; + return Next run {formatNextRun(job.next_run_at)}; +} + export function JobControls() { const { data: jobs, isLoading } = useJobs(); const toggleJob = useToggleJob(); const triggerJob = useTriggerJob(); + // Job id -> display label, so a step can name its parent pipeline. + const labels = Object.fromEntries((jobs ?? []).map((job) => [job.name, job.label])); const anyJobRunning = (jobs ?? []).some((job) => job.running); const runningJob = jobs?.find((job) => job.running); const pausedJob = jobs?.find((job) => !job.running && job.runtime_status === 'rate_limited'); @@ -148,24 +173,18 @@ export function JobControls() { ? 'Active' : 'Inactive'} - {job.via_pipeline ? ( - runs via pipeline - ) : ( - job.enabled && job.next_run_at && ( - - Next run {formatNextRun(job.next_run_at)} - - ) - )} + {job.enabled && } {!job.registered && ( Not registered )} - {!job.running && job.runtime_finished_at && ( -
- Last run {formatAgo(job.runtime_finished_at)} - {job.runtime_status ? ` · ${job.runtime_status}` : ''} - {job.runtime_message ? ` — ${job.runtime_message}` : ''} + {/* Persisted, so this survives a deploy — unlike runtime_*, + which the status chip above still reads for live state. */} + {!job.running && job.last_run_at && ( +
+ Last run {formatAgo(job.last_run_at)} + {job.last_run_status ? ` · ${job.last_run_status}` : ''} + {job.last_run_message ? ` — ${job.last_run_message}` : ''}
)} {job.running && ( diff --git a/tests/unit/test_job_run_persistence.py b/tests/unit/test_job_run_persistence.py new file mode 100644 index 0000000..4af6f2b --- /dev/null +++ b/tests/unit/test_job_run_persistence.py @@ -0,0 +1,166 @@ +"""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()