From 5d107d2adce0a323927d05bcd2301cb0fd62c460 Mon Sep 17 00:00:00 2001 From: Guillaume Onfroy Date: Thu, 17 Sep 2026 22:55:47 +0200 Subject: [PATCH] feat(db)!: retry a failed run, and report the stack's verdict A failed run queues its own next attempt when its target job declares a policy. That happens in `RunStore.complete`, the single terminal path every failure takes (executor verdict, executor exception, launch failure, reaper timeout), inside the transaction that marks the run failed. Doing it there rather than in a sweeping controller is what makes "has a successor" true the instant a run fails, so a doomed attempt never looks final to anything reading the table. A run is one attempt; `root_run_id` groups the attempts of one unit of work into a stack, and `scheduled_for` holds the backoff, honoured by the queue's claim so a retry waits without needing a status of its own and without holding the head of the line. The root cannot be a column default, since it references the row's own id, and a `table=True` model skips validation, so a `before_insert` listener owns the invariant and no creation site has to know about it. Backfills finalize on stacks: the verdict reads each stack's latest attempt, so an attempt a later one healed no longer condemns the batch, while a queued successor still counts as in flight and keeps it open. The `executions` view ranks by attempt before severity for the same reason, and gains an `attempts` column; `operation_retried` counts toward it without ever winning the verdict. Hooks observe a verdict, never an attempt: the sweep skips a failed run whose successor already exists, which needs no knowledge of budgets or backoff. The context gains the stack's position so a message can say the work succeeded on the second attempt or failed after three. Nothing retries until a job declares a policy: there is no instance-wide default. By Digitl --- .../specs/2026-09-17-retry-design.md | 21 ++ .../migrations/versions/004_run_attempts.py | 125 ++++++++++++ .../src/interloper_db/models/runs.py | 47 ++++- .../src/interloper_db/store/runs.py | 99 +++++++++- .../versions/test_004_run_attempts.py | 176 +++++++++++++++++ .../interloper-db/tests/store/test_runs.py | 183 ++++++++++++++++++ .../src/interloper_scheduler/hooks.py | 19 +- .../src/interloper_scheduler/queue.py | 9 +- .../interloper-scheduler/tests/test_hooks.py | 101 ++++++++++ .../interloper-scheduler/tests/test_queue.py | 45 +++++ 10 files changed, 815 insertions(+), 10 deletions(-) create mode 100644 packages/interloper-db/src/interloper_db/migrations/versions/004_run_attempts.py create mode 100644 packages/interloper-db/tests/migrations/versions/test_004_run_attempts.py diff --git a/docs/superpowers/specs/2026-09-17-retry-design.md b/docs/superpowers/specs/2026-09-17-retry-design.md index e22c21b8..68aea94e 100644 --- a/docs/superpowers/specs/2026-09-17-retry-design.md +++ b/docs/superpowers/specs/2026-09-17-retry-design.md @@ -350,6 +350,20 @@ UPDATE runs SET root_run_id = chain.root FROM chain WHERE runs.id = chain.id; Any run row whose `retry_of` points at a deleted run (the FK is `ON DELETE SET NULL`) becomes its own root, which is correct: its lineage is gone. +Three constraints the implementation found, all load-bearing: + +- **The DDL must be idempotent.** `create_all` creates tables from the models and *then* runs the + chain, so on a fresh database the columns already exist by the time 004 runs. Without + `IF NOT EXISTS`, the migration fails on every fresh database, which includes `make dev-reset` and + any new deployment. +- **`attempts` is appended as the view's last column.** `CREATE OR REPLACE VIEW` may only add + columns at the end; inserting one mid-list reads to Postgres as renaming the column that was + there, and it refuses. +- **`root_run_id` cannot be a column default**, because it references the row's own id, and a + `table=True` model skips pydantic validation so a validator never fires. A `before_insert` + listener on `Run` owns the invariant instead, which keeps every creation site, and every future + one, free of it. + --- ## 9. Surfaces @@ -435,6 +449,13 @@ changed, after phase 2 runs retry and hooks report verdicts, after phase 3 the s ## 14. Follow-ups, recorded +- **Retried attempts are indistinguishable in a trace.** Each attempt opens its own + `interloper.operation.execute` span, which is the conventional shape, but the attributes are built + once from the operation's own metadata and carry no attempt number, so N attempts produce N + identically named, identically attributed sibling spans. Adding `interloper.attempt` is small. + Giving the node a parent span with the attempts as children is the fuller fix: it matches what the + UI timeline shows, and it changes the span tree existing dashboards read. + - **An instance-wide default.** Dropped from this design on purpose. Whoever picks it up owns two questions: whether an instance default is a floor, a ceiling or a plain fallback under a declared policy, and whether the operation-level default belongs in a settings block next to the run one or diff --git a/packages/interloper-db/src/interloper_db/migrations/versions/004_run_attempts.py b/packages/interloper-db/src/interloper_db/migrations/versions/004_run_attempts.py new file mode 100644 index 00000000..2d177a75 --- /dev/null +++ b/packages/interloper-db/src/interloper_db/migrations/versions/004_run_attempts.py @@ -0,0 +1,125 @@ +"""Add the run stack and its schedule. + +A run is one attempt. ``root_run_id`` groups the attempts of one unit of work +so a stack is a single indexed predicate rather than a recursive walk, and is +the run's own id for a first attempt. ``scheduled_for`` holds a retry's +backoff: the queue claims a run only once it has passed. + +The DDL is idempotent, because ``create_all`` creates the tables from the +models and *then* runs the chain: on a fresh database the columns already +exist by the time this runs, and on an existing one they do not. + +Existing rows are folded into stacks by walking the ``retry_of`` chains manual +retries already created. A run whose predecessor was deleted becomes its own +root, which is correct: its lineage is gone. + +Revision ID: 004 +Revises: 003 +""" + +from __future__ import annotations + +from alembic import op + +_EXECUTIONS_VIEW = """CREATE OR REPLACE VIEW executions AS +WITH ranked AS ( + SELECT + e.run_id, + e.org_id, + e.component_id, + e.component_key, + e.event_type, + e.timestamp, + row_number() OVER ( + PARTITION BY e.run_id, e.component_id + ORDER BY + COALESCE((e.data->>'attempt')::int, 1) DESC, + CASE e.event_type + WHEN 'operation_failed' THEN 1 + WHEN 'operation_canceled' THEN 2 + WHEN 'operation_completed' THEN 3 + WHEN 'operation_started' THEN 4 + WHEN 'operation_skipped' THEN 5 + WHEN 'operation_retried' THEN 6 + WHEN 'operation_queued' THEN 7 + END, + e.timestamp DESC + ) AS rn, + max(COALESCE((e.data->>'attempt')::int, 1)) OVER ( + PARTITION BY e.run_id, e.component_id + ) AS attempts, + min(CASE WHEN e.event_type = 'operation_queued' THEN e.timestamp END) OVER ( + PARTITION BY e.run_id, e.component_id + ) AS queued_at, + min(CASE WHEN e.event_type = 'operation_started' THEN e.timestamp END) OVER ( + PARTITION BY e.run_id, e.component_id + ) AS started_at, + max(CASE WHEN e.event_type IN ('operation_completed', 'operation_failed', 'operation_canceled') + THEN e.timestamp END) OVER ( + PARTITION BY e.run_id, e.component_id + ) AS completed_at + FROM events e + WHERE e.component_id IS NOT NULL + AND e.event_type IN ( + 'operation_queued', 'operation_skipped', 'operation_started', + 'operation_completed', 'operation_failed', 'operation_canceled', + 'operation_retried' + ) +) +SELECT + r.run_id, + r.org_id, + r.component_id, + r.component_key, + CASE r.event_type + WHEN 'operation_failed' THEN 'failed' + WHEN 'operation_canceled' THEN 'canceled' + WHEN 'operation_completed' THEN 'success' + WHEN 'operation_started' THEN 'running' + WHEN 'operation_skipped' THEN 'skipped' + WHEN 'operation_retried' THEN 'running' + WHEN 'operation_queued' THEN 'queued' + END AS status, + r.started_at, + r.completed_at, + r.queued_at AS created_at, + r.attempts +FROM ranked r +WHERE r.rn = 1 +""" + +# revision identifiers, used by Alembic. +revision: str = "004" +down_revision: str | None = "003" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.execute("ALTER TABLE runs ADD COLUMN IF NOT EXISTS root_run_id uuid REFERENCES runs(id) ON DELETE SET NULL") + op.execute("ALTER TABLE runs ADD COLUMN IF NOT EXISTS scheduled_for timestamptz") + op.execute( + """ + WITH RECURSIVE chain AS ( + SELECT id, id AS root FROM runs WHERE retry_of IS NULL + UNION ALL + SELECT r.id, c.root FROM runs r JOIN chain c ON r.retry_of = c.id + ) + UPDATE runs SET root_run_id = chain.root FROM chain WHERE runs.id = chain.id + """ + ) + op.execute("UPDATE runs SET root_run_id = id WHERE root_run_id IS NULL") + op.execute("ALTER TABLE runs ALTER COLUMN root_run_id SET NOT NULL") + op.execute("CREATE INDEX IF NOT EXISTS ix_runs_root_run_id ON runs (root_run_id)") + op.execute("CREATE INDEX IF NOT EXISTS ix_runs_claim ON runs (status, scheduled_for, created_at)") + op.execute(_EXECUTIONS_VIEW) + + +def downgrade() -> None: + # Leaves the attempt-aware view in place: it is a superset of the old one + # for any data the old chain could have produced, and re-creating the + # previous definition here would duplicate migration 002 verbatim. + op.execute("DROP INDEX IF EXISTS ix_runs_claim") + op.execute("DROP INDEX IF EXISTS ix_runs_root_run_id") + op.execute("ALTER TABLE runs DROP COLUMN scheduled_for") + op.execute("ALTER TABLE runs DROP COLUMN root_run_id") diff --git a/packages/interloper-db/src/interloper_db/models/runs.py b/packages/interloper-db/src/interloper_db/models/runs.py index a80348e6..8d253daa 100644 --- a/packages/interloper-db/src/interloper_db/models/runs.py +++ b/packages/interloper-db/src/interloper_db/models/runs.py @@ -2,9 +2,9 @@ from datetime import datetime from typing import Any, ClassVar, Optional -from uuid import UUID +from uuid import UUID, uuid4 -from sqlalchemy import ForeignKey, Index +from sqlalchemy import ForeignKey, Index, event from sqlmodel import Column, Relationship, SQLModel, text from sqlmodel import Field as SQLField @@ -52,6 +52,12 @@ class Backfill(SQLModel, table=True): class Run(SQLModel, table=True): """A single execution of a component's operation. + A run is one *attempt*. ``root_run_id`` groups the attempts of one unit of + work into a stack and is the run's own id for a first attempt, so stack + membership is one indexed predicate rather than a recursive walk. + ``scheduled_for`` is the earliest instant the queue may claim the run, + which is how a retry's backoff is served without a second status. + ``quota_reserved_at`` is set when a dispatch-time quota reservation was taken; its month tells settlement which usage period to release. ``billable`` records the operation's declaration at creation time, so @@ -86,6 +92,11 @@ class Run(SQLModel, table=True): ) attempt: int = 1 retry_scope: str | None = None + root_run_id: UUID = SQLField( + default=None, + sa_column=Column(ForeignKey("runs.id", ondelete="SET NULL"), index=True, nullable=False), + ) + scheduled_for: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime)) billable: bool = True quota_reserved_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime)) started_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime)) @@ -126,6 +137,29 @@ def event_metadata(self, target: Component | None) -> dict[str, Any]: return metadata +@event.listens_for(Run, "before_insert") +def _stamp_stack_root(_mapper: Any, _connection: Any, target: Run) -> None: + """Default a run's stack root to itself, and its id to a fresh one. + + A first attempt roots its own stack, which cannot be expressed as a column + default because it references the row's own id. Doing it here rather than + at each creation site keeps the invariant in one place: a ``table=True`` + model skips pydantic validation, so a validator would never fire, and + every caller remembering would be a trap for the next one. The id is + generated too, because the root cannot be set before it exists; the + column's server default stays for rows inserted outside the ORM. + + Args: + _mapper: The mapper being flushed, unused. + _connection: The connection the flush runs on, unused. + target: The run row about to be inserted, stamped in place. + """ + if target.id is None: + target.id = uuid4() + if target.root_run_id is None: + target.root_run_id = target.id + + class Event(SQLModel, table=True): """An execution event persisted for observability. @@ -166,9 +200,11 @@ class Event(SQLModel, table=True): class Execution(SQLModel, table=True): """Read model over the ``executions`` view — never written. - One row per ``(run, operation)``: the current status derived from - lifecycle events (severity then recency) plus the queued/started/completed - timestamps. The view itself is created by migration 002; ``create_all`` + One row per ``(run, operation)``: the operation's verdict, derived from + its lifecycle events (latest attempt first, then severity, then recency) + plus the queued/started/completed timestamps and how many attempts it + took. The timestamps span every attempt, so a retried operation reads as + one execution from its first start to its final outcome. The view itself is created by migration 002; ``create_all`` skips view-backed models (see the ``is_view`` marker). """ @@ -180,6 +216,7 @@ class Execution(SQLModel, table=True): org_id: UUID component_key: str | None = None status: str + attempts: int = 1 started_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime)) completed_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime)) created_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime)) diff --git a/packages/interloper-db/src/interloper_db/store/runs.py b/packages/interloper-db/src/interloper_db/store/runs.py index 4ed9551a..e1d0a203 100644 --- a/packages/interloper-db/src/interloper_db/store/runs.py +++ b/packages/interloper-db/src/interloper_db/store/runs.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Any from uuid import UUID @@ -250,6 +250,10 @@ def complete(self, run_id: UUID, *, success: bool) -> Run: every run takes (scheduled, manual, retried), so the component's "last run" reflects all of them. + A failure also queues its own next attempt when the target's policy + allows one, before the backfill advances so the batch's in-flight + count sees the successor and does not finalize early. + Args: run_id: The run UUID. success: Whether the run succeeded. @@ -277,6 +281,9 @@ def complete(self, run_id: UUID, *, success: bool) -> Run: db_component.stamp_state(last_run_at=db_run.completed_at, last_run_status=db_run.status) session.add(db_component) + if not success: + self._plan_retry(session, db_run) + if db_run.backfill_id: self._advance_backfill(session, db_run.backfill_id, failed=not success) @@ -315,6 +322,7 @@ def retry(self, run_id: UUID, *, scope: str = "all") -> Run: if src.billable: self._quotas.check(src.org_id, QUOTA_MAX_SUCCESSFUL_RUNS_PER_MONTH, subject="retry") db_run = Run( + root_run_id=src.root_run_id, org_id=src.org_id, component_id=src.component_id, partition_key=src.partition_key, @@ -330,6 +338,75 @@ def retry(self, run_id: UUID, *, scope: str = "all") -> Run: _ = db_run.target # load before the session closes; readers reach it detached return db_run + @staticmethod + def _retry_policy(session: Session, db_run: Run) -> il.RetryPolicy | None: + """The run-level policy in force for a run. + + A job's declared policy governs its runs, and nothing else does: a + source's or an asset's own ``retry`` is an operation budget, and + reading it here would spend an operation's attempts on whole runs. + There is no instance-wide default either, so a run whose target + declares nothing is attempted once. ``config`` is a plain JSON + column, so this is one row read and no hydration. + + Args: + session: Open session the target row is read through. + db_run: The run whose policy is resolved. + + Returns: + The policy, or ``None`` when the target declares none. + """ + if db_run.component_id is None: + return None + db_component = session.get(Component, db_run.component_id) + if db_component is None or db_component.kind != "job": + return None + declared = (db_component.config or {}).get("retry") + return il.RetryPolicy.model_validate(declared) if declared else None + + def _plan_retry(self, session: Session, db_run: Run) -> Run | None: + """Queue the next attempt of a failed run, when its budget allows one. + + Called from the single terminal path, in the transaction that marks + the run failed, so a doomed attempt never looks final to anything + reading the table — which is what lets the hook evaluator gate on the + successor's existence without knowing anything about budgets. The + successor stays in its predecessor's backfill and stack, and re-runs + only what failed. + + The quota is deliberately not checked here: dispatch is the + authoritative gate and cancels an over-quota run at claim time, like + any other run. + + Args: + session: Open session the successor is written through. + db_run: The run that just failed. + + Returns: + The queued successor, or ``None`` when nothing is retried. + """ + policy = self._retry_policy(session, db_run) + if policy is None or not policy.allows(db_run.attempt + 1): + return None + + successor = Run( + org_id=db_run.org_id, + component_id=db_run.component_id, + backfill_id=db_run.backfill_id, + partition_key=db_run.partition_key, + status="queued", + scheduled_for=datetime.now(timezone.utc) + timedelta(seconds=policy.delay_before(db_run.attempt + 1)), + retry_of=db_run.id, + root_run_id=db_run.root_run_id, + attempt=db_run.attempt + 1, + retry_scope="failed", + billable=db_run.billable, + ) + session.add(successor) + session.flush() + logger.info("Queued attempt %d of run stack %s", successor.attempt, successor.root_run_id) + return successor + # -- Backfills ------------------------------------------------------------- def create_backfill( @@ -586,7 +663,11 @@ def _advance_backfill(session: Session, backfill_id: UUID, *, failed: bool) -> N """Advance a backfill after a run completes. 1. **Fail-fast**: if enabled and the run failed, cancel pending runs. - 2. **Finalize**: if nothing in-flight or pending, mark complete. + 2. **Finalize**: if nothing in-flight or pending, mark complete. The + verdict reads each stack's latest attempt, so an attempt a later one + healed no longer condemns the batch. A queued successor still counts + as in flight, which is what keeps the batch open while a retry waits + out its backoff. 3. **Advance**: promote next pending runs up to concurrency limit. Args: @@ -628,8 +709,20 @@ def _advance_backfill(session: Session, backfill_id: UUID, *, failed: bool) -> N ).all() if in_flight_count == 0 and len(pending_runs) == 0: + latest = ( + select(col(Run.root_run_id), func.max(col(Run.attempt)).label("attempt")) + .where(Run.backfill_id == backfill_id) + .group_by(col(Run.root_run_id)) + .subquery() + ) any_failed = session.exec( - select(Run).where(Run.backfill_id == backfill_id, Run.status == "failed") + select(Run) + .join( + latest, + onclause=(col(Run.root_run_id) == latest.c.root_run_id) + & (col(Run.attempt) == latest.c.attempt), + ) + .where(Run.backfill_id == backfill_id, Run.status == "failed") ).first() db_backfill.status = "failed" if any_failed else "success" db_backfill.completed_at = datetime.now(timezone.utc) diff --git a/packages/interloper-db/tests/migrations/versions/test_004_run_attempts.py b/packages/interloper-db/tests/migrations/versions/test_004_run_attempts.py new file mode 100644 index 00000000..dd214ca1 --- /dev/null +++ b/packages/interloper-db/tests/migrations/versions/test_004_run_attempts.py @@ -0,0 +1,176 @@ +"""Tests for migration ``004_run_attempts``: the attempt-aware ``executions`` view. + +The view is Postgres SQL, so nothing but a live server exercises its ranking: +the SQLite suites stand the read model up as a table and test the mapping, not +the query. This module reads a server DSN from ``INTERLOPER_TEST_POSTGRES_DSN``, +provisions a throwaway database migrated to head, and drops it afterwards; +without the variable it skips. +""" + +from __future__ import annotations + +import datetime as dt +import os +from collections.abc import Iterator +from urllib.parse import urlparse, urlunparse +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy import Engine +from sqlmodel import Session, select + +from interloper_db import engine as engine_module +from interloper_db import provision +from interloper_db.models import Event, Execution, Run + +pytestmark = pytest.mark.integration + +_ORG_ID = uuid4() + + +@pytest.fixture(scope="module") +def postgres_db() -> Iterator[Engine]: + """A throwaway Postgres database migrated to head. + + Yields: + The global engine bound to that database, dropped once the module finishes. + """ + server_dsn = os.getenv("INTERLOPER_TEST_POSTGRES_DSN") + if not server_dsn: + pytest.skip("INTERLOPER_TEST_POSTGRES_DSN not set") + dsn = urlunparse(urlparse(server_dsn)._replace(path=f"/interloper_test_{uuid4().hex[:8]}")) + provision.ensure_database(dsn) + engine = engine_module.init_engine(dsn) + try: + provision.create_all(engine) + yield engine + finally: + engine.dispose() + engine_module._engine = None + provision.drop_database(dsn) + + +def _emit(session: Session, run_id: UUID, component_id: UUID, event_type: str, attempt: int, second: int) -> None: + """Append one operation-lifecycle event for a given attempt.""" + session.add( + Event( + id=uuid4(), + org_id=_ORG_ID, + run_id=run_id, + event_type=event_type, + component_id=component_id, + component_key="orders", + data={"attempt": attempt}, + timestamp=dt.datetime(2026, 1, 1, 12, 0, second, tzinfo=dt.timezone.utc), + ) + ) + + +def _execution(engine: Engine, run_id: UUID) -> Execution: + with Session(engine) as session: + return session.exec(select(Execution).where(Execution.run_id == run_id)).one() + + +def _run(session: Session) -> UUID: + run = Run(org_id=_ORG_ID, status="running") + session.add(run) + session.flush() + return run.id + + +class TestVerdictReadsTheLatestAttempt: + def test_an_operation_that_healed_reads_as_a_success(self, postgres_db: Engine) -> None: + component_id = uuid4() + with Session(postgres_db) as session: + run_id = _run(session) + _emit(session, run_id, component_id, "operation_started", 1, 0) + _emit(session, run_id, component_id, "operation_retried", 1, 1) + _emit(session, run_id, component_id, "operation_started", 2, 2) + _emit(session, run_id, component_id, "operation_completed", 2, 3) + session.commit() + + execution = _execution(postgres_db, run_id) + + assert execution.status == "success" + assert execution.attempts == 2 + + def test_an_operation_that_exhausted_its_budget_reads_as_a_failure(self, postgres_db: Engine) -> None: + component_id = uuid4() + with Session(postgres_db) as session: + run_id = _run(session) + _emit(session, run_id, component_id, "operation_started", 1, 0) + _emit(session, run_id, component_id, "operation_retried", 1, 1) + _emit(session, run_id, component_id, "operation_started", 2, 2) + _emit(session, run_id, component_id, "operation_failed", 2, 3) + session.commit() + + execution = _execution(postgres_db, run_id) + + assert execution.status == "failed" + assert execution.attempts == 2 + + def test_a_single_attempt_is_unchanged(self, postgres_db: Engine) -> None: + component_id = uuid4() + with Session(postgres_db) as session: + run_id = _run(session) + _emit(session, run_id, component_id, "operation_queued", 1, 0) + _emit(session, run_id, component_id, "operation_started", 1, 1) + _emit(session, run_id, component_id, "operation_completed", 1, 2) + session.commit() + + execution = _execution(postgres_db, run_id) + + assert (execution.status, execution.attempts) == ("success", 1) + + def test_events_written_before_attempts_existed_count_as_one(self, postgres_db: Engine) -> None: + # Historical rows carry no `attempt` in their data; COALESCE reads them + # as the first attempt rather than dropping them from the ranking. + component_id = uuid4() + with Session(postgres_db) as session: + run_id = _run(session) + session.add( + Event( + id=uuid4(), + org_id=_ORG_ID, + run_id=run_id, + event_type="operation_completed", + component_id=component_id, + component_key="orders", + timestamp=dt.datetime(2026, 1, 1, 12, 0, 0, tzinfo=dt.timezone.utc), + ) + ) + session.commit() + + execution = _execution(postgres_db, run_id) + + assert (execution.status, execution.attempts) == ("success", 1) + + def test_a_retried_attempt_never_wins_the_verdict(self, postgres_db: Engine) -> None: + # The newest event of the latest attempt is the retry itself, which + # must read as still running rather than as the operation's outcome. + component_id = uuid4() + with Session(postgres_db) as session: + run_id = _run(session) + _emit(session, run_id, component_id, "operation_started", 1, 0) + _emit(session, run_id, component_id, "operation_retried", 1, 1) + session.commit() + + execution = _execution(postgres_db, run_id) + + assert execution.status == "running" + assert execution.attempts == 1 + + def test_the_timestamps_span_every_attempt(self, postgres_db: Engine) -> None: + component_id = uuid4() + with Session(postgres_db) as session: + run_id = _run(session) + _emit(session, run_id, component_id, "operation_started", 1, 0) + _emit(session, run_id, component_id, "operation_retried", 1, 1) + _emit(session, run_id, component_id, "operation_started", 2, 5) + _emit(session, run_id, component_id, "operation_completed", 2, 9) + session.commit() + + execution = _execution(postgres_db, run_id) + + assert execution.started_at == dt.datetime(2026, 1, 1, 12, 0, 0, tzinfo=dt.timezone.utc) + assert execution.completed_at == dt.datetime(2026, 1, 1, 12, 0, 9, tzinfo=dt.timezone.utc) diff --git a/packages/interloper-db/tests/store/test_runs.py b/packages/interloper-db/tests/store/test_runs.py index 8793be79..b6a2afe9 100644 --- a/packages/interloper-db/tests/store/test_runs.py +++ b/packages/interloper-db/tests/store/test_runs.py @@ -117,6 +117,27 @@ def _component(store: Store, kind: str, key: str | None = None, name: str | None return row.id +def _job_with_retry(store: Store, **policy: Any) -> UUID: + """A job component whose config declares a retry policy. + + Returns: + The component id. + """ + with Session(store.engine) as session: + row = Component( + id=uuid4(), + org_id=_ORG_ID, + kind="job", + key="job", + name="job", + config={"retry": policy} if policy else {}, + ) + session.add(row) + session.commit() + assert row.id is not None + return row.id + + class TestRunTargetOperations: """Run creation validates the target's operation and records billability.""" @@ -200,6 +221,168 @@ def test_backfill_rejects_a_kind_with_no_workload(self, store: Store): store.runs.create_backfill(_ORG_ID, component_id=target, start_key="2026-01-01", end_key="2026-01-02") +class TestStackIdentity: + """Every run belongs to a stack; a first attempt is its own root.""" + + def test_a_new_run_is_its_own_stack_root(self, store: Store) -> None: + run = store.runs.create(_ORG_ID) + + assert run.root_run_id == run.id + assert run.scheduled_for is None + assert run.attempt == 1 + + def test_backfill_runs_are_each_their_own_root(self, store: Store) -> None: + backfill = _backfill(store) + + with Session(store.engine) as session: + runs = session.exec(select(Run).where(Run.backfill_id == backfill.id)).all() + assert {run.root_run_id for run in runs} == {run.id for run in runs} + + def test_a_manual_retry_joins_its_predecessors_stack(self, store: Store) -> None: + run = store.runs.create(_ORG_ID) + store.runs.complete(run.id, success=False) + + retry = store.runs.retry(run.id) + + assert retry.root_run_id == run.root_run_id + assert retry.id != run.id + assert retry.attempt == 2 + + +class TestAutomaticRetry: + """A failed run queues its own next attempt when its target allows one.""" + + def test_a_failed_run_queues_its_next_attempt(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=2, delay=60) + run = store.runs.create(_ORG_ID, component_id=target) + + store.runs.complete(run.id, success=False) + + with Session(store.engine) as session: + successor = session.exec(select(Run).where(Run.retry_of == run.id)).one() + assert successor.root_run_id == run.root_run_id + assert successor.attempt == 2 + assert successor.retry_scope == "failed" + assert successor.status == "queued" + assert successor.scheduled_for is not None + assert successor.billable == run.billable + + def test_an_exhausted_budget_queues_nothing(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=1) + run = store.runs.create(_ORG_ID, component_id=target) + + store.runs.complete(run.id, success=False) + + with Session(store.engine) as session: + assert session.exec(select(Run).where(Run.retry_of == run.id)).all() == [] + + def test_a_successful_run_queues_nothing(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=3) + run = store.runs.create(_ORG_ID, component_id=target) + + store.runs.complete(run.id, success=True) + + with Session(store.engine) as session: + assert session.exec(select(Run).where(Run.retry_of == run.id)).all() == [] + + def test_a_target_declaring_no_policy_queues_nothing(self, store: Store) -> None: + target = _component(store, kind="job") + run = store.runs.create(_ORG_ID, component_id=target) + + store.runs.complete(run.id, success=False) + + with Session(store.engine) as session: + assert session.exec(select(Run).where(Run.retry_of == run.id)).all() == [] + + def test_a_source_policy_is_not_read_at_the_run_level(self, store: Store) -> None: + # A source's `retry` is an operation budget; reading it here would + # apply an operation's attempts to whole runs. + with Session(store.engine) as session: + row = Component( + id=uuid4(), + org_id=_ORG_ID, + kind="source", + key="shop", + name="shop", + config={"retry": {"max_attempts": 5}}, + ) + session.add(row) + session.commit() + target = row.id + run = store.runs.create(_ORG_ID, component_id=target) + + store.runs.complete(run.id, success=False) + + with Session(store.engine) as session: + assert session.exec(select(Run).where(Run.retry_of == run.id)).all() == [] + + def test_a_run_whose_target_is_gone_queues_nothing(self, store: Store) -> None: + run = store.runs.create(_ORG_ID) + + store.runs.complete(run.id, success=False) + + with Session(store.engine) as session: + assert session.exec(select(Run).where(Run.retry_of == run.id)).all() == [] + + def test_the_successor_stays_in_its_backfill(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=2, delay=0) + backfill = store.runs.create_backfill( + _ORG_ID, component_id=target, start_key="2026-01-01", end_key="2026-01-01" + ) + with Session(store.engine) as session: + run = session.exec(select(Run).where(Run.backfill_id == backfill.id)).one() + + store.runs.complete(run.id, success=False) + + with Session(store.engine) as session: + successor = session.exec(select(Run).where(Run.retry_of == run.id)).one() + assert successor.backfill_id == backfill.id + + +class TestBackfillStacks: + """A batch's verdict reads each stack's latest attempt, not every attempt.""" + + def _single_partition_backfill(self, store: Store, target: UUID) -> tuple[UUID, Run]: + backfill = store.runs.create_backfill( + _ORG_ID, component_id=target, start_key="2026-01-01", end_key="2026-01-01" + ) + with Session(store.engine) as session: + run = session.exec(select(Run).where(Run.backfill_id == backfill.id)).one() + assert backfill.id is not None + return backfill.id, run + + def _successor(self, store: Store, run_id: UUID) -> Run: + with Session(store.engine) as session: + return session.exec(select(Run).where(Run.retry_of == run_id)).one() + + def test_a_backfill_healed_by_a_retry_succeeds(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=2, delay=0) + backfill_id, first = self._single_partition_backfill(store, target) + + store.runs.complete(first.id, success=False) + store.runs.complete(self._successor(store, first.id).id, success=True) + + assert store.runs.get_backfill(backfill_id).status == "success" + + def test_a_backfill_whose_stack_exhausts_its_budget_fails(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=2, delay=0) + backfill_id, first = self._single_partition_backfill(store, target) + + store.runs.complete(first.id, success=False) + store.runs.complete(self._successor(store, first.id).id, success=False) + + assert store.runs.get_backfill(backfill_id).status == "failed" + + def test_a_pending_retry_keeps_the_backfill_open(self, store: Store) -> None: + target = _job_with_retry(store, max_attempts=2, delay=0) + backfill_id, first = self._single_partition_backfill(store, target) + + store.runs.complete(first.id, success=False) + + # The successor is queued, so the batch still has work in flight. + assert store.runs.get_backfill(backfill_id).status == "running" + + class TestCreateBackfill: """Dispatch order: newest partition first (ITLPR-120).""" diff --git a/packages/interloper-scheduler/src/interloper_scheduler/hooks.py b/packages/interloper-scheduler/src/interloper_scheduler/hooks.py index 0e252fb3..78b3c3c8 100644 --- a/packages/interloper-scheduler/src/interloper_scheduler/hooks.py +++ b/packages/interloper-scheduler/src/interloper_scheduler/hooks.py @@ -5,6 +5,12 @@ window, matches them against hooks watching the run's target component (or its parent source), and calls each matching hook's ``fire()``. +A hook observes a **verdict**, never an attempt: a failed run whose next +attempt is already queued is not an outcome, so the sweep skips it. Because +the successor is created in the same transaction that marks the run failed, +there is no window in which a doomed attempt looks final, and the rule needs +no knowledge of budgets or backoff. + Delivery is **at-least-evaluated, at-most-fired-once**: every firing is claimed by an ``events`` row whose id is deterministic (uuid5 of hook + run), so the overlap window and restarts re-evaluate runs without re-firing hooks. @@ -25,6 +31,8 @@ from interloper_db import Store from interloper_db.models import Component, ComponentRelation, Run from interloper_db.models import Event as EventRow +from sqlalchemy import func +from sqlalchemy.orm import aliased from sqlmodel import Session, col, select from interloper_scheduler.controller import Controller @@ -92,10 +100,13 @@ def _tick(self) -> None: since = self._watermark - _OVERLAP with Session(self._store.engine) as session: + successor = aliased(Run) + has_successor = select(successor.id).where(col(successor.retry_of) == Run.id).exists() runs = session.exec( select(Run) .where(col(Run.status).in_(_TERMINAL_STATUSES)) .where(col(Run.completed_at) > since) + .where(~((col(Run.status) == "failed") & has_successor)) .order_by(col(Run.completed_at)) ).all() @@ -146,7 +157,9 @@ def _event_metadata(self, session: Session, run: Run, target: Component, event_t The ids in the context are the machine-readable half; this is the half a hook addressing humans (a Slack message) renders, so it carries the - component's display name and — for a failure — the error the run + component's display name, the stack's position (this attempt's number + and how many the stack holds, so a message can say it succeeded on the + second or failed after three) and — for a failure — the error the run recorded, which lives on the run's event rows rather than the run. Returns: @@ -156,6 +169,10 @@ def _event_metadata(self, session: Session, run: Run, target: Component, event_t "status": run.status, "component_name": target.name or target.key, "component_key": target.key, + "attempt": run.attempt, + "attempts": session.exec( + select(func.count()).select_from(Run).where(Run.root_run_id == run.root_run_id) + ).one(), } if event_type == "run_failed": error = session.exec( diff --git a/packages/interloper-scheduler/src/interloper_scheduler/queue.py b/packages/interloper-scheduler/src/interloper_scheduler/queue.py index d6a1964b..1ba1d94f 100644 --- a/packages/interloper-scheduler/src/interloper_scheduler/queue.py +++ b/packages/interloper-scheduler/src/interloper_scheduler/queue.py @@ -11,6 +11,7 @@ from interloper_db import Store from interloper_db.models import Backfill, Event, Run from interloper_db.store.runs import cancel_backfill_runs +from sqlalchemy import func from sqlmodel import Session, col, select from interloper_scheduler.controller import Controller @@ -77,7 +78,12 @@ def _tick(self) -> None: self._store.runs.complete(run_id, success=False) def _claim_next(self) -> UUID | None: - """Claim the oldest queued run, reserve its quota slot, and mark it dispatched. + """Claim the oldest claimable queued run, reserve its quota, and dispatch it. + + A run carrying a schedule is not claimable until it has passed, which + is how a retry serves its backoff without needing a status of its own. + It is skipped rather than waited on, so a run backing off never holds + the head of the queue. This is the authoritative run-quota gate: dispatch requires an atomic reservation, so an exhausted organisation can never execute past its @@ -93,6 +99,7 @@ def _claim_next(self) -> UUID | None: statement = ( select(Run) .where(Run.status == "queued") + .where(col(Run.scheduled_for).is_(None) | (col(Run.scheduled_for) <= func.now())) .order_by(col(Run.created_at).asc()) .limit(1) .with_for_update(skip_locked=True) diff --git a/packages/interloper-scheduler/tests/test_hooks.py b/packages/interloper-scheduler/tests/test_hooks.py index 79dbee06..8eec7fbf 100644 --- a/packages/interloper-scheduler/tests/test_hooks.py +++ b/packages/interloper-scheduler/tests/test_hooks.py @@ -269,6 +269,8 @@ def test_metadata_carries_component_identity(self, store: Store, monkeypatch: py "status": "success", "component_name": "Demo", "component_key": "demo_source", + "attempt": 1, + "attempts": 1, } def test_metadata_falls_back_to_key_when_unnamed(self, store: Store, monkeypatch: pytest.MonkeyPatch): @@ -347,6 +349,105 @@ def test_chain_to_unwatched_target_is_allowed(self, store: Store): assert [q.component_id for q in queued] == [root.id] +class TestVerdictGating: + """A hook observes a stack's verdict, never one of its attempts.""" + + def _job(self, **retry: Any) -> UUID: + """A job row declaring a retry policy, inserted directly. + + ``store.components.create`` would build the component; this only needs + the row, which is all the run level and the hook sweep read. + + Returns: + The component id. + """ + with Session(engine_module.get_engine()) as session: + row = Component( + id=uuid4(), + org_id=_ORG, + kind="job", + key="cron_job", + name="Nightly", + config={"cron": "0 6 * * *", **({"retry": retry} if retry else {})}, + ) + session.add(row) + session.commit() + assert row.id is not None + return row.id + + def _watching_hook(self, store: Store, component_id: UUID) -> UUID: + hook = store.components.create( + _ORG, kind="hook", key="webhook_hook", name="Notify", + config={"events": ["run_completed", "run_failed"], "url": "https://example.invalid/hook"}, + relations={"watches": [component_id]}, + ) + return hook.id + + def _fired(self, run_id: UUID) -> list[EventRow]: + with Session(engine_module.get_engine()) as session: + return list( + session.exec(select(EventRow).where(EventRow.run_id == run_id, EventRow.event_type == "hook_fired")) + ) + + def _successor(self, run_id: UUID) -> Run: + with Session(engine_module.get_engine()) as session: + return session.exec(select(Run).where(Run.retry_of == run_id)).one() + + def test_a_failed_run_that_will_be_retried_does_not_fire( + self, store: Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + _capture_posts(monkeypatch) + job = self._job(max_attempts=2, delay=0) + self._watching_hook(store, job) + run = store.runs.create(_ORG, component_id=job) + store.runs.complete(run.id, success=False) + + _sweep(store) + + assert self._fired(run.id) == [] + + def test_an_exhausted_stack_fires_once(self, store: Store, monkeypatch: pytest.MonkeyPatch) -> None: + _capture_posts(monkeypatch) + job = self._job(max_attempts=1) + self._watching_hook(store, job) + run = store.runs.create(_ORG, component_id=job) + store.runs.complete(run.id, success=False) + + _sweep(store) + + assert len(self._fired(run.id)) == 1 + + def test_a_healed_stack_fires_completed_on_the_successful_attempt( + self, store: Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + _capture_posts(monkeypatch) + job = self._job(max_attempts=2, delay=0) + self._watching_hook(store, job) + first = store.runs.create(_ORG, component_id=job) + store.runs.complete(first.id, success=False) + successor = self._successor(first.id) + store.runs.complete(successor.id, success=True) + + _sweep(store) + + assert self._fired(first.id) == [] + assert len(self._fired(successor.id)) == 1 + + def test_the_context_carries_the_stacks_position( + self, store: Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + payloads = _capture_posts(monkeypatch) + job = self._job(max_attempts=1) + self._watching_hook(store, job) + run = store.runs.create(_ORG, component_id=job) + store.runs.complete(run.id, success=False) + + _sweep(store) + + assert payloads[0]["metadata"]["attempt"] == 1 + assert payloads[0]["metadata"]["attempts"] == 1 + + class TestFirstTickWatermark: """The first sweep sets its own watermark rather than replaying history.""" diff --git a/packages/interloper-scheduler/tests/test_queue.py b/packages/interloper-scheduler/tests/test_queue.py index 2275f003..00117943 100644 --- a/packages/interloper-scheduler/tests/test_queue.py +++ b/packages/interloper-scheduler/tests/test_queue.py @@ -2,6 +2,7 @@ from __future__ import annotations +import datetime as dt from collections.abc import Iterator from typing import Any from uuid import UUID, uuid4 @@ -74,6 +75,50 @@ def test_tick_drains_the_queue(store: Store) -> None: assert set(_statuses(store).values()) == {"dispatched"} +def _schedule(store: Store, run_id: UUID, *, seconds: float) -> None: + """Move a queued run's earliest claim time by *seconds* from now.""" + with Session(store.engine) as session: + db_run = session.get(Run, run_id) + assert db_run is not None + db_run.scheduled_for = dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=seconds) + session.add(db_run) + session.commit() + + +def test_a_scheduled_run_is_not_claimed_before_its_time(store: Store) -> None: + run = store.runs.create(_ORG) + _schedule(store, run.id, seconds=3600) + launcher = _FakeLauncher() + + QueueController(launcher=launcher, store=store)._tick() + + assert launcher.launched == [] + assert _statuses(store)[run.id] == "queued" + + +def test_a_run_whose_schedule_has_passed_is_claimed(store: Store) -> None: + run = store.runs.create(_ORG) + _schedule(store, run.id, seconds=-1) + launcher = _FakeLauncher() + + QueueController(launcher=launcher, store=store)._tick() + + assert launcher.launched == [run.id] + + +def test_a_scheduled_run_does_not_block_the_queue_behind_it(store: Store) -> None: + # The claim orders by creation, so a run waiting out its backoff must be + # skipped rather than held at the head of the line. + waiting = store.runs.create(_ORG) + _schedule(store, waiting.id, seconds=3600) + ready = store.runs.create(_ORG) + launcher = _FakeLauncher() + + QueueController(launcher=launcher, store=store)._tick() + + assert launcher.launched == [ready.id] + + def test_empty_queue_is_a_noop(store: Store) -> None: launcher = _FakeLauncher() QueueController(launcher=launcher, store=store)._tick()