From 00d63b2eebed46d0f00150712ceba9b37ab5e24a Mon Sep 17 00:00:00 2001 From: Guillaume Onfroy Date: Fri, 18 Sep 2026 00:10:45 +0200 Subject: [PATCH] feat(api)!: list runs by stack A stack is one piece of work, so a listing shows its latest attempt and every filter reads that attempt: a stack whose first attempt failed and whose second succeeded is a success, which is what a reader means by "failed runs". Passing `root_run_id` asks for one stack's attempts instead, newest first. The narrowing is a grouped join rather than `DISTINCT ON`, so it runs on SQLite as well as Postgres, and it is scoped to the organisation alone: every attempt of a stack shares its target, its backfill and its org, so no other filter can change which attempt is the latest. Narrowing by the caller's filters would answer a different question, "the latest failed attempt" rather than "the stacks whose latest attempt failed". `RunResponse` gains `root_run_id` and `scheduled_for`. It gains no attempt count: a listing row is the stack's latest attempt, so `attempt` already is how many the stack took, and a second field would only duplicate it, or lie when an older attempt is fetched by id. In the app a row shows how many attempts its work took, the run page walks the stack's attempts, and the realtime path supersedes an earlier attempt's row instead of appending beside it, which would have broken the one-row-per-stack invariant the listing promises. By Digitl --- docs/guide/execution.md | 44 +++++++ docs/guide/hooks.md | 16 +++ .../src/interloper_api/routes/runs.py | 20 +++- .../interloper-api/tests/routes/test_runs.py | 45 +++++++ .../app/components/executions/RunsTable.vue | 13 +- .../app/app/pages/executions/runs/[run].vue | 31 +++++ .../interloper-app/app/app/stores/runs.ts | 28 ++++- packages/interloper-app/app/app/types/run.ts | 8 ++ .../src/interloper_db/store/runs.py | 111 +++++++++++++----- .../interloper-db/tests/store/test_runs.py | 61 ++++++++++ 10 files changed, 339 insertions(+), 38 deletions(-) diff --git a/docs/guide/execution.md b/docs/guide/execution.md index 5c400fd5..b8762ee2 100644 --- a/docs/guide/execution.md +++ b/docs/guide/execution.md @@ -118,6 +118,50 @@ Before anything executes, the runner validates the scope against every enabled operation: partitioned operations without a scope, windows against operations that forbid them, and time-partition mismatches fail the whole run up front. +## Retrying + +An operation that fails is attempted again when it carries a `RetryPolicy`, without the run seeing +the intermediate failure. Nothing retries by default: a policy is in force only where a component +declares one. + +```py +@il.asset(retry=il.RetryPolicy(max_attempts=3, delay=5)) +def orders(**kwargs): ... +``` + +The policy is declared on the component whose unit it governs, so one budget never means several +things: + +| declared on | governs | +|---|---| +| an asset, or any operation | that operation's execution | +| a source | the default its assets inherit | +| a job | that job's runs | + +`max_attempts` counts the first attempt, and the wait before each further one grows geometrically +from `delay` by `backoff`, capped at `max_delay` and spread by `jitter` so operations failing +together do not retry in lockstep. Whether a given error is worth another attempt is behaviour +rather than configuration: override `retryable()` to recognise one that never will be. + +```py +class Orders(il.Asset): + retry = il.RetryPolicy(max_attempts=3) + + def retryable(self, error: Exception) -> bool: + return not isinstance(error, PermissionError) +``` + +A retried attempt is not a verdict. The node keeps its status, nothing downstream is canceled, and +`fail_fast` is not tripped; only an exhausted or declined failure marks the node failed. Each +attempt is recorded as an `operation_retried` event, and `operation_failed` means the budget ran +out. + +A job's policy works one level up: when a run fails, the platform queues the next attempt after the +backoff. Those attempts form a **stack**, and the rest of the system reads the stack rather than any +one attempt. An automatic retry always re-runs only what failed, carrying forward every operation +that already succeeded anywhere in the stack; the manual retry endpoint additionally offers re-running +everything, for when the earlier success is the thing you distrust. + ## Running single assets `asset.run()` and `asset.materialize()` bypass the runner. Pass the DAG when the asset has diff --git a/docs/guide/hooks.md b/docs/guide/hooks.md index 3670df06..3d4061b5 100644 --- a/docs/guide/hooks.md +++ b/docs/guide/hooks.md @@ -83,6 +83,22 @@ any component. `HookState` (`last_fired_at`, `last_run_id`) is the hook's machine-owned state, stamped by the operator on every firing. +## Retries + +A hook observes a **verdict**, never an attempt. When a run fails and its job's retry policy allows +another attempt, the failure is not an outcome and no hook fires. Only the attempt that ends the +stack does: as `run_completed` if a retry healed the work, or as `run_failed` once the budget is +exhausted. + +The context carries the stack's position, so a message can say which attempt it is reporting: + +```py +def fire(self, context: il.HookContext) -> None: + attempt = context.metadata["attempt"] + attempts = context.metadata["attempts"] + post(f"{context.metadata['component_name']}: {context.metadata['status']} on attempt {attempt}/{attempts}") +``` + ## Scope Hooks fire on **persisted, terminal runs** evaluated by a scheduler. For observing execution diff --git a/packages/interloper-api/src/interloper_api/routes/runs.py b/packages/interloper-api/src/interloper_api/routes/runs.py index a680f1f7..e02296a4 100644 --- a/packages/interloper-api/src/interloper_api/routes/runs.py +++ b/packages/interloper-api/src/interloper_api/routes/runs.py @@ -40,6 +40,11 @@ class RunResponse(BaseModel): target kinds they do not know about. All three ``component_*`` identity fields are ``None`` exactly when the target was deleted (``component_id`` nulls on deletion). + + A response is one attempt. In a stack-native listing it is the stack's + latest, so ``attempt`` is also how many attempts the stack took; fetching + an older attempt by id gives that attempt's own number. ``root_run_id`` + is what groups them, and lists them through ``?root_run_id=``. """ id: UUID @@ -52,8 +57,10 @@ class RunResponse(BaseModel): partition_key: str | None status: str retry_of: UUID | None = None + root_run_id: UUID | None = None attempt: int = 1 retry_scope: str | None = None + scheduled_for: str | None = None started_at: str | None = None completed_at: str | None = None created_at: str | None = None @@ -79,8 +86,10 @@ def from_run(cls, run: Run) -> RunResponse: partition_key=run.partition_key, status=run.status, retry_of=run.retry_of, + root_run_id=run.root_run_id, attempt=run.attempt, retry_scope=run.retry_scope, + scheduled_for=str(run.scheduled_for) if run.scheduled_for else None, started_at=str(run.started_at) if run.started_at else None, completed_at=str(run.completed_at) if run.completed_at else None, created_at=str(run.created_at) if run.created_at else None, @@ -222,10 +231,16 @@ def list_runs( q: str | None = None, component_kind: str | None = None, component_key: str | None = None, + root_run_id: UUID | None = None, limit: int = 50, offset: int = 0, ) -> list[RunResponse]: - """List runs with optional filters. + """List one row per run stack, or one stack's attempts. + + A stack is one piece of work, so a listing carries its latest attempt and + every filter reads that attempt: a stack whose first attempt failed and + whose second succeeded is a success. Passing ``root_run_id`` asks for one + stack's attempts instead, newest first. ``after``/``before`` bound the runs to those whose execution overlaps that window — a run occupies ``started_at`` → ``completed_at`` (open-ended while @@ -247,6 +262,7 @@ def list_runs( component_kind: Keep only runs targeting a component of this kind; None applies no filter. component_key: Keep only runs targeting a component of this type (catalog key); None applies no filter. + root_run_id: List this stack's attempts rather than one row per stack. limit: Maximum number of runs on the page. offset: Number of matching runs to skip before the page starts. user: The authenticated user, required to hold at least the ``viewer`` role. @@ -266,6 +282,7 @@ def list_runs( q=q, component_kind=component_kind, component_key=component_key, + root_run_id=root_run_id, ) response.headers["X-Total-Count"] = str(total) runs = store.runs.list_all( @@ -278,6 +295,7 @@ def list_runs( q=q, component_kind=component_kind, component_key=component_key, + root_run_id=root_run_id, limit=limit, offset=offset, ) diff --git a/packages/interloper-api/tests/routes/test_runs.py b/packages/interloper-api/tests/routes/test_runs.py index 7c544634..581bd2e7 100644 --- a/packages/interloper-api/tests/routes/test_runs.py +++ b/packages/interloper-api/tests/routes/test_runs.py @@ -35,8 +35,10 @@ def _fake_run(run_id: UUID, org_id: UUID = _ORG_ID) -> SimpleNamespace: partition_key=None, status="failed", retry_of=None, + root_run_id=run_id, attempt=1, retry_scope=None, + scheduled_for=None, started_at=None, completed_at=None, created_at=None, @@ -252,6 +254,49 @@ def test_list_runs_forwards_the_time_window(store: FakeStore) -> None: assert (store.count_calls[0]["after"], store.count_calls[0]["before"]) == window +def _viewer_client(store: FakeStore) -> TestClient: + """A client authenticated as a viewer of the fixture organisation. + + Args: + store: The fake store the routes resolve against. + + Returns: + The client. + """ + app = _app(store) + app.dependency_overrides[require_viewer] = lambda: SimpleNamespace(id=uuid4()) + app.dependency_overrides[get_org_id] = lambda: _ORG_ID + return TestClient(app) + + +def test_list_runs_forwards_the_stack_filter(store: FakeStore) -> None: + """Asking for one stack narrows both the listing and its count.""" + root = uuid4() + resp = _viewer_client(store).get("/runs/", params={"root_run_id": str(root)}) + + assert resp.status_code == 200 + assert store.list_calls[0]["root_run_id"] == root + assert store.count_calls[0]["root_run_id"] == root + + +def test_list_runs_defaults_to_one_row_per_stack(store: FakeStore) -> None: + """Without the filter the store is asked for stacks, not attempts.""" + resp = _viewer_client(store).get("/runs/") + + assert resp.status_code == 200 + assert store.list_calls[0]["root_run_id"] is None + + +def test_a_run_response_carries_its_stack(store: FakeStore) -> None: + resp = _client(store).get(f"/runs/{_RUN_ID}") + + assert resp.status_code == 200 + body = resp.json() + assert body["root_run_id"] == str(_RUN_ID) + assert body["attempt"] == 1 + assert body["scheduled_for"] is None + + def test_list_runs_forwards_the_target_filters(store: FakeStore) -> None: """The target's kind, type and a name search narrow both the listing and its count.""" app = _app(store) diff --git a/packages/interloper-app/app/app/components/executions/RunsTable.vue b/packages/interloper-app/app/app/components/executions/RunsTable.vue index 276115ba..01439576 100644 --- a/packages/interloper-app/app/app/components/executions/RunsTable.vue +++ b/packages/interloper-app/app/app/components/executions/RunsTable.vue @@ -84,8 +84,19 @@ const columns: TableColumn[] = [ accessorKey: 'status', header: 'Status', cell: ({ row }) => { + const run = row.original as Run const status = row.getValue('status') - return h(UBadge, { color: statusColor(status) }, () => statusLabel(status)) + const badge = h(UBadge, { color: statusColor(status) }, () => statusLabel(status)) + // A row is a stack at its latest attempt, so `attempt` is how many + // it took: worth showing only when it took more than one. + if (run.attempt <= 1) return badge + return h('div', { class: 'flex items-center gap-1.5' }, [ + badge, + h('span', { + class: 'text-xs text-muted tabular-nums', + title: `This work took ${run.attempt} attempts`, + }, `${run.attempt} attempts`), + ]) }, }, { diff --git a/packages/interloper-app/app/app/pages/executions/runs/[run].vue b/packages/interloper-app/app/app/pages/executions/runs/[run].vue index 7ed90db9..94e41cba 100644 --- a/packages/interloper-app/app/app/pages/executions/runs/[run].vue +++ b/packages/interloper-app/app/app/pages/executions/runs/[run].vue @@ -100,6 +100,25 @@ async function onRetry(scope: 'all' | 'failed') { const fetchError = ref(null) +/** + * The attempts of this run's stack, newest first, fetched only when there is + * more than one. A listing carries a stack's latest attempt, so reaching the + * others is what this page adds. + */ +const stack = ref([]) + +async function loadStack(fetched: Run) { + const root = fetched.root_run_id + if (!root || (fetched.attempt === 1 && root === fetched.id)) return + try { + stack.value = await runsStore.fetchStack(root) + } + catch { + // The page stands on its own without the stack; losing it is not an error worth showing. + stack.value = [] + } +} + onMounted(async () => { try { const [fetchedRun] = await Promise.all([ @@ -118,6 +137,7 @@ onMounted(async () => { initialRun.value = fetchedRun // Seed the store so realtime updates can find and update it. runsStore._upsert(fetchedRun) + await loadStack(fetchedRun) } catch (e) { fetchError.value = e @@ -143,6 +163,17 @@ onUnmounted(() => { +
+ Attempts + {{ attempt.attempt }} +
{ /********************** * Internals **********************/ + /** + * A listing holds one row per stack, its latest attempt. So a run arriving + * over realtime either updates its own row, supersedes the earlier attempt + * of the stack it belongs to, or is new work. + */ function _upsert(run: Run) { const idx = runs.value.findIndex(r => r.id === run.id) - if (idx >= 0) runs.value[idx] = { ...runs.value[idx], ...run } - else { - runs.value.unshift(run) - total.value++ + if (idx >= 0) { + runs.value[idx] = { ...runs.value[idx], ...run } + return + } + const predecessor = run.root_run_id + ? runs.value.findIndex(r => (r.root_run_id ?? r.id) === run.root_run_id) + : -1 + if (predecessor >= 0) { + runs.value[predecessor] = run + return } + runs.value.unshift(run) + total.value++ } function _remove(id: string) { @@ -107,6 +120,12 @@ export const useRunsStore = defineStore('runs', () => { return apiFetch(`/runs/${id}`) } + /** Every attempt of one stack, newest first. A listing only ever carries the latest. */ + async function fetchStack(rootRunId: string): Promise { + const params = new URLSearchParams({ root_run_id: rootRunId }) + return apiFetch(`/runs?${params}`) + } + /** Queue a manual run for a runnable component (job, source, or asset). Returns the created run's id. */ async function createRun(componentId: string, partitionKey?: string): Promise { const run = await apiFetch('/runs', { @@ -176,6 +195,7 @@ export const useRunsStore = defineStore('runs', () => { fetch, fetchOne, createRun, + fetchStack, retryRun, goToPage, setFilters, diff --git a/packages/interloper-app/app/app/types/run.ts b/packages/interloper-app/app/app/types/run.ts index b97541f9..da2f0bd1 100644 --- a/packages/interloper-app/app/app/types/run.ts +++ b/packages/interloper-app/app/app/types/run.ts @@ -11,8 +11,16 @@ export interface Run { partition_key: string | null status: string retry_of: string | null + /** The stack this attempt belongs to; its own id for a first attempt. */ + root_run_id: string | null + /** + * This attempt's number. A listing carries each stack's latest attempt, so + * there it is also how many attempts the stack took. + */ attempt: number retry_scope: string | null + /** Earliest instant the queue may claim this run, set while a retry backs off. */ + scheduled_for: string | null started_at: string | null completed_at: string | null created_at: string | null diff --git a/packages/interloper-db/src/interloper_db/store/runs.py b/packages/interloper-db/src/interloper_db/store/runs.py index e1d0a203..84a245b5 100644 --- a/packages/interloper-db/src/interloper_db/store/runs.py +++ b/packages/interloper-db/src/interloper_db/store/runs.py @@ -149,10 +149,17 @@ def list_all( q: str | None = None, component_kind: str | None = None, component_key: str | None = None, + root_run_id: UUID | None = None, limit: int = 50, offset: int = 0, ) -> list[Run]: - """List runs with optional filters. + """List one row per stack, or one stack's attempts. + + A stack is one piece of work, so a listing shows its **latest + attempt** and every filter reads that attempt: a stack whose first + attempt failed and whose second succeeded is a success, which is what + a reader means by "failed runs". Passing *root_run_id* asks for one + stack instead, and returns its attempts newest first. Args: org_id: Organisation UUID. @@ -164,6 +171,7 @@ def list_all( q: Keep runs whose target's name or key contains this, case-insensitively. component_kind: Keep runs whose target is of this kind. component_key: Keep runs whose target is of this type (catalog key). + root_run_id: List this stack's attempts rather than one row per stack. limit: Max results (default 50). offset: Pagination offset. @@ -171,22 +179,25 @@ def list_all( List of Run rows. """ with session_scope(self._engine) as session: + filters = self._run_filters( + org_id, + component_id, + backfill_id, + status, + after, + before, + q=q, + component_kind=component_kind, + component_key=component_key, + root_run_id=root_run_id, + ) + if root_run_id is None: + filters.append(self._latest_attempt_only(org_id)) + order = col(Run.created_at).desc() if root_run_id is None else col(Run.attempt).desc() statement = ( select(Run) - .where( - *self._run_filters( - org_id, - component_id, - backfill_id, - status, - after, - before, - q=q, - component_kind=component_kind, - component_key=component_key, - ) - ) - .order_by(col(Run.created_at).desc()) + .where(*filters) + .order_by(order) .offset(offset) .limit(limit) .options(*RUN_LOAD_OPTIONS) @@ -205,6 +216,7 @@ def count( q: str | None = None, component_kind: str | None = None, component_key: str | None = None, + root_run_id: UUID | None = None, ) -> int: """Count runs matching the same filters as :meth:`list_all`. @@ -218,29 +230,27 @@ def count( q: Keep runs whose target's name or key contains this, case-insensitively. component_kind: Keep runs whose target is of this kind. component_key: Keep runs whose target is of this type (catalog key). + root_run_id: Count this stack's attempts rather than one per stack. Returns: Total number of matching runs (ignoring limit/offset). """ with session_scope(self._engine) as session: - statement = ( - select(func.count()) - .select_from(Run) - .where( - *self._run_filters( - org_id, - component_id, - backfill_id, - status, - after, - before, - q=q, - component_kind=component_kind, - component_key=component_key, - ) - ) + filters = self._run_filters( + org_id, + component_id, + backfill_id, + status, + after, + before, + q=q, + component_kind=component_kind, + component_key=component_key, + root_run_id=root_run_id, ) - return session.exec(statement).one() + if root_run_id is None: + filters.append(self._latest_attempt_only(org_id)) + return session.exec(select(func.count()).select_from(Run).where(*filters)).one() def complete(self, run_id: UUID, *, success: bool) -> Run: """Mark a run as completed and advance its backfill if applicable. @@ -585,6 +595,38 @@ def list_active_backfills(self, org_id: UUID) -> list[Backfill]: # -- Internals ------------------------------------------------------------- + @staticmethod + def _latest_attempt_only(org_id: UUID) -> Any: + """Keep only each stack's latest attempt. + + Scoped to the organisation alone on purpose: every attempt of a stack + shares its target, its backfill and its org, so no other filter can + change which attempt is the latest. Narrowing by the caller's filters + instead would answer a different question, such as "the latest *failed* + attempt" rather than "the stacks whose latest attempt failed". + + Expressed as a grouped join rather than ``DISTINCT ON`` so it runs on + SQLite as well as Postgres. + + Args: + org_id: Organisation whose stacks are reduced. + + Returns: + A filter expression selecting the latest attempt of each stack. + """ + latest = ( + select(col(Run.root_run_id), func.max(col(Run.attempt)).label("attempt")) + .where(Run.org_id == org_id) + .group_by(col(Run.root_run_id)) + .subquery() + ) + return col(Run.id).in_( + select(col(Run.id)).join( + latest, + onclause=(col(Run.root_run_id) == latest.c.root_run_id) & (col(Run.attempt) == latest.c.attempt), + ) + ) + @staticmethod def _run_filters( org_id: UUID, @@ -597,6 +639,7 @@ def _run_filters( q: str | None = None, component_kind: str | None = None, component_key: str | None = None, + root_run_id: UUID | None = None, ) -> list[Any]: """The shared where-clauses of :meth:`RunStore.list_all` / :meth:`RunStore.count`. @@ -625,6 +668,8 @@ def _run_filters( applies no kind filter. component_key: Keep runs whose target is of this type (catalog key); ``None`` applies no type filter. + root_run_id: Keep the attempts of this stack; ``None`` applies no + stack filter. Returns: Filter expressions for the given criteria. @@ -646,6 +691,8 @@ def _run_filters( filters.append(Run.component_id == component_id) if backfill_id: filters.append(Run.backfill_id == backfill_id) + if root_run_id: + filters.append(Run.root_run_id == root_run_id) if status: filters.append(Run.status == status) if after is not None: diff --git a/packages/interloper-db/tests/store/test_runs.py b/packages/interloper-db/tests/store/test_runs.py index b6a2afe9..25d4f98f 100644 --- a/packages/interloper-db/tests/store/test_runs.py +++ b/packages/interloper-db/tests/store/test_runs.py @@ -383,6 +383,67 @@ def test_a_pending_retry_keeps_the_backfill_open(self, store: Store) -> None: assert store.runs.get_backfill(backfill_id).status == "running" +class TestStackNativeListing: + """A listing shows one row per stack: its latest attempt.""" + + def _failed_then(self, store: Store, *, success: bool) -> tuple[Run, Run]: + """A two-attempt stack whose second attempt ends as asked. + + Returns: + The first attempt and its successor. + """ + target = _job_with_retry(store, max_attempts=2, delay=0) + first = store.runs.create(_ORG_ID, component_id=target) + store.runs.complete(first.id, success=False) + with Session(store.engine) as session: + successor = session.exec(select(Run).where(Run.retry_of == first.id)).one() + store.runs.complete(successor.id, success=success) + return first, successor + + def test_a_stack_is_one_row_at_its_latest_attempt(self, store: Store) -> None: + first, successor = self._failed_then(store, success=True) + + runs = store.runs.list_all(_ORG_ID) + + assert [run.id for run in runs] == [successor.id] + assert runs[0].attempt == 2 + assert first.id not in {run.id for run in runs} + + def test_count_matches_the_listing(self, store: Store) -> None: + self._failed_then(store, success=True) + + assert store.runs.count(_ORG_ID) == 1 + + def test_a_status_filter_reads_the_stacks_verdict(self, store: Store) -> None: + # The first attempt failed, so a run-level filter would surface it; the + # stack succeeded, and that is what a reader means by "failed runs". + self._failed_then(store, success=True) + + assert store.runs.list_all(_ORG_ID, status="failed") == [] + assert len(store.runs.list_all(_ORG_ID, status="success")) == 1 + + def test_an_exhausted_stack_still_reads_as_failed(self, store: Store) -> None: + self._failed_then(store, success=False) + + assert len(store.runs.list_all(_ORG_ID, status="failed")) == 1 + + def test_a_stack_lists_its_attempts_newest_first(self, store: Store) -> None: + first, successor = self._failed_then(store, success=True) + + attempts = store.runs.list_all(_ORG_ID, root_run_id=first.root_run_id) + + assert [run.id for run in attempts] == [successor.id, first.id] + assert store.runs.count(_ORG_ID, root_run_id=first.root_run_id) == 2 + + def test_unretried_runs_are_unaffected(self, store: Store) -> None: + first = store.runs.create(_ORG_ID) + second = store.runs.create(_ORG_ID) + + runs = store.runs.list_all(_ORG_ID) + + assert {run.id for run in runs} == {first.id, second.id} + + class TestCreateBackfill: """Dispatch order: newest partition first (ITLPR-120)."""