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)."""