Skip to main content

recovery

Restarting background DAG runs whose process is gone.

The sibling of bitfount.runtimes.recovery, and separate from it because the execution model decides everything: a background DAG run is in-process, so it is found by lineage tags and replayable only by rebuilding it from a stored flow spec, and it needs an attempt bound because this module polls. The metadata runtimes are deployment runs, found by deployment_id and resubmitted verbatim from their own parameters, once per process start. What the two share is the query, which lives in bitfount.runtimes.discovery.

A worker.background DAG runs in-process in the pod, on a customer's laptop. A Windows Update reboot, a power cut or an OOM kill takes it down mid-run, and nothing else notices: it is not a deployment run, so no scheduler re-submits it, and the pod that would have closed out its bookkeeping died with it.

This module is what notices, the next time a pod runs.

Why a poller and not an automation​

Prefect's own zombie reaper is a proactive automation, and proactive trigger state lives in automation_bucket, swept on server restart. The Prefect server here is a child of the same application as the pod, so a machine-wide outage kills both and the missed-heartbeat window is gone by the time anything is alive to act on it — the crash this feature exists for is exactly the one an automation cannot see. Detection therefore has to compare stored state against wall-clock time after the fact, which is a poll wherever it is hosted; and it is hosted in the pod because no automation action can reach in-process code, and a background DAG can only execute in a pod anyway.

The five guards​

A lost run is replayed only when all of these hold, and each covers a case the others cannot:

  1. Its process is provably gone — CRASHED, or RUNNING with a stale heartbeat (is_flow_run_lost). Heartbeat staleness is what survives an outage that took the Prefect server with it.
  2. This pod is not executing it — a suspended laptop's own healthy run looks stale, because its heartbeat thread suspended too. Ownership is the only thing that distinguishes "asleep" from "dead", so a run in our in-flight set is never recovered no matter how silent it is.
  3. Nothing has replaced it yet — nothing here ends a dead run and Prefect keeps terminal runs indefinitely, so every crash a lineage ever had stays discoverable and "lost" forever. Two checks, because they see different things: _latest_per_lineage drops the older crashes discovery returned, and _is_superseded drops a crash whose replacement has since failed or completed — invisible to a CRASHED/RUNNING query, and the case that otherwise replays one crash on every tick forever.
  4. Indexing is not in flight for it — a file_metadata re-index moves the input under the DAG's feet, so recovery waits for the next tick.
  5. The lineage has banked progress recently — should_recover bounds consecutive attempts that achieved nothing, so a poisoned spec cannot loop unattended forever.

Guards 1 and 2 are complementary in the same way heartbeats and ownership are throughout: staleness catches a dead process whoever owned it, including a sibling pod's; ownership catches our own live process being suspended.

Module​

Functions​

find_lost_background_runs​

async def find_lost_background_runs(    client: PrefectClient,    *,    in_flight_task_hashes: Container[str],    silence_window: timedelta | None = None,) ‑> list[LostRun]:

Return the background runs this pod should consider replaying.

Applies guards 1 to 3 (see the module docstring): the run's process must be provably gone, it must not be one this pod is currently executing, and it must be the run its lineage currently ends at — _latest_per_lineage for the runs discovery can see, then _is_superseded for the ones it cannot.

Arguments

  • client: An open PrefectClient.
  • in_flight_task_hashes: The task hashes this pod is running right now. A suspended laptop's own run looks exactly like a dead one, so this is the only thing that keeps it from being double-started.
  • silence_window: Passed through to is_flow_run_lost.

Returns At most one lost run per lineage: the RUNNING ones first, then the CRASHED ones, each group newest first.

recover_lost_runs​

async def recover_lost_runs(    *,    client: PrefectClient,    cache: CacheProtocol,    in_flight_task_hashes: Container[str],    replay: ReplayFn,    progress_for: ProgressFn,    attempt_limit: int,    silence_window: timedelta | None = None,) ‑> list[LostRun]:

Find lost background runs and replay the ones that qualify.

One tick of the poller. Never raises for a single run's sake: a lineage that cannot be judged is skipped with its reason logged, so one bad spec cannot stop the others being recovered.

Arguments

  • client: An open PrefectClient.
  • cache: The pod's background cache, holding both the specs and the rows progress is measured from.
  • in_flight_task_hashes: Task hashes this pod is executing (guard 2).
  • replay: Schedules a fresh attempt from a stored spec.
  • progress_for: Measures what a lineage has banked so far.
  • attempt_limit: How many consecutive unproductive attempts to tolerate.
  • silence_window: Passed through to is_flow_run_lost.

Returns The runs a replacement attempt was started for.

run_recovery_poller​

async def run_recovery_poller(    *, tick: Callable[[], Coroutine[Any, Any, Any]], interval_seconds: float,) ‑> None:

Run tick every interval_seconds until cancelled.

A tick that raises is logged and the loop continues: recovery is a background enhancement, and a poller that died on one bad tick would take the feature down silently for the pod's whole lifetime. Cancellation propagates, so pod shutdown is immediate.

The first tick runs immediately rather than after a delay, because the case this feature exists for — a pod starting up after a crash — has work waiting the moment the pod is ready.

Arguments

  • tick: One pass of recovery.
  • interval_seconds: Delay between passes.

Global variables​

  • ProgressFn - Measures what a lineage has banked so far, given its stored spec. Injected for the same reason: it needs a rebuilt DAG, which only the pod can assemble.
  • ReplayFn - Rebuilds and schedules a fresh attempt from a stored spec. Injected rather than imported: replaying needs the pod's datasources, Hub session and cache, none of which this module should know about — and keeping it a callback is what lets the pod route a recovered attempt through the same construction path a dataset-project link uses.

Classes​

LostRun​

class LostRun(    flow_run_id: uuid.UUID,    task_hash: str,    project_id: str,    datasource_name: str,    attempt: int,    expected_start_time: datetime | None = None,):

A background flow run whose process is gone, identified by its tags.

Arguments

  • flow_run_id: The dead run, retained for logging and for excluding it from the liveness query when its replacement asks.
  • task_hash: The (pod, datasource) task hash from the run's tags.
  • project_id: Project from the run's tags.
  • datasource_name: Datasource from the run's tags.
  • attempt: The dead run's position in its lineage; its replacement is attempt + 1.

Variables​

  • static attempt : int - The dead run's position in its lineage.
  • static datasource_name : str - Datasource from the run's tags.
  • static expected_start_time : datetime.datetime | None - When Prefect expected this run to start — set on its first state transition, so every run has one. This, not attempt, is what orders a lineage: the tag is a label a buggy replay can repeat, while the timestamp is what the server itself sorts by. None only for a run whose record carries no such field at all, which is then ordered last.
  • static flow_run_id : uuid.UUID - The dead run.
  • static project_id : str - Project from the run's tags.
  • static task_hash : str - The (pod, datasource) task hash from the run's tags.