Skip to main content

recovery

Recovery for metadata-runtime flow runs whose process is gone.

The sibling of bitfount.flows.dag.recovery, and separate from it because the execution model decides everything: these are deployment runs, so they are found by deployment_id, replayed verbatim from their own parameters, and need their Prefect state and bookkeeping rows reconciled — while a background DAG run is in-process, found by lineage tags, and replayable only by rebuilding it from a stored flow spec. What the two share is the query, which lives in bitfount.runtimes.discovery.

The metadata runtimes being deployment runs is also why ADR 0005's in-process recovery poller does not cover them — it discovers work by the run_type:background_dag tag and replays it from a persisted flow spec. These runs need neither: they are submitted by a deployment run whose parameters already carry everything needed to submit it again, so recovery is a verbatim resubmission.

What they lacked was anyone to notice. The submitters are one-shot — the orchestrator fires _trigger_file_metadata_runtime once per pod start, run_pod once at startup — and each gates on is_flow_run_active. Until that gate became heartbeat-aware it read the raw Prefect state, so a pod restarting inside the zombie-reaper's ~90s window found its own orphaned run still sitting at RUNNING, called it live, skipped, and never asked again. The datasource simply stopped being indexed, silently, and scan_metadata never ran either because the scan-chain automation fires on a Completed event that would now never arrive.

The gate fix alone would leave recovery dependent on a submitter running after the reaper had acted. This module removes that dependency: it runs once per process start, judges liveness itself, and resubmits.

Three things happen per lost run, and the order matters:

  1. Establish there is nothing live. Per (task_hash, run_type), ask is_flow_run_active. A live run means this pod is looking at its own healthy work and must not touch it.
  2. Reconcile. End the dead run at Crashed and fail its running bookkeeping row. The Prefect transition is not needed for correctness — the gate already ignores the run — but a permanently-RUNNING row is read by people as "still working", and the reaper cannot be relied on to fix it: it cannot observe a crash that took the Prefect server down with it, and its pending window is swept from automation_bucket on server restart. The bookkeeping row matters more: _is_run_active is the only thing that reaps stale running rows, it is called only from the refresh cron, and the refresh cron only visits datasources that already have rows — so a first run that crashed before banking anything leaves a row nothing will ever clean.
  3. Resubmit. A file_metadata resubmission normally supersedes a scan_metadata one, so at most one run per task_hash: the scan-chain automation runs scan when file_metadata completes, and running scan concurrently with the indexer would parse a partial inventory, bank rows and finish COMPLETED — under-coverage that looks like success and never trips any bound. That withholding is conditional on the chain existing, which this sweep ensures and then checks rather than assumes: with no chain, withholding is the drop, so both run types are replayed and scan_metadata_runtime's own deferral sequences them.

Recovery is not attempt-bounded, and that asymmetry with bitfount.flows.dag.recovery is deliberate. That module polls every 60s and so can burn a machine unattended; this one fires once per process start, and for any datasource still in the pod config _trigger_file_metadata_runtime already resubmits on every pod start regardless. A bound here would guard a loop that the restart cadence already bounds. Resubmissions do carry an incrementing attempt: tag, so a run that is genuinely looping is visible as such.

Module​

Functions​

collapse_scheduled_backlog​

async def collapse_scheduled_backlog(client: PrefectClient) ‑> int:

Cancel every superseded SCHEDULED run of the metadata deployments.

Runs once per process start, beside recover_lost_metadata_runs, which discovers RUNNING and CRASHED only — so nothing has ever collected a SCHEDULED run.

A backlog forms because a deployment's global_limit enqueues on collision: when the slots are taken the server returns the run to Scheduled(AwaitingConcurrencySlot), timed 30s out, and the runner logs "Server returned a non-pending state 'SCHEDULED'" and leaves it. That much is by design. What is not is that nothing drains it: ACTIVE_STATE_TYPES counts SCHEDULED as live and is_flow_run_lost will not judge anything pre-RUNNING, so a wedged run reads as live forever and permanently suppresses resubmission for its task_hash — one more orphan per day under a cron.

Keeps the newest of each group rather than ageing runs out. Every threshold available here is wrong: a run waiting its turn is indistinguishable by age from a wedged one, and liveness_silence_window() (90s) is orders of magnitude shorter than the daily cron it would judge, so sizing from it would cancel healthy queued work on every start.

Grouped per (deployment, task_hash, _work_key). Dropping any of the three loses work: per deployment alone would cancel datasource B's queued index because A had a newer one, and without the work key a scoped watcher run would cancel a queued full walk.

Cancelling needs no bound, unlike the resubmission in recover_lost_metadata_runs: it is terminal, so a collapsed group has one run left and a second pass finds nothing. force is left off, matching _end_run_as_crashed.

Arguments

  • client: An open PrefectClient.

Returns How many runs were cancelled.

recover_lost_metadata_runs​

async def recover_lost_metadata_runs(    client: PrefectClient,    *,    dead_before: datetime | None = None,    own_pod_name: str | None = None,) ‑> int:

Reconcile and replay every metadata run whose process is gone.

Runs once per process start, from the same place the other server-side guarantees are established (prefect_bootstrap's ensure_* calls). Callers must treat a failure here as non-fatal — indexing recovery is worth attempting, never worth blocking startup for.

The mutating half of this module carries a sleep hazard that is closed by where it is called from rather than by a threshold, because no threshold could close it: a laptop suspended for hours has its healthy run's heartbeat thread suspended too, so that run looks exactly as dead as a crashed one. Three things make it safe anyway. First, this runs only at process start, and a suspended machine is not starting processes — reap and sleep are mutually exclusive on one machine. Second, generate_prefect_task_hash is per-pod, so this can only ever touch runs belonging to this pod's datasources; a genuinely live run of the same task_hash would require a second process serving the same pod name, which the orchestrator's single-pod lock prevents and a container deployment has no desktop app beside it to create. Third, dead_before only ever raises the cutoff, and a resumed laptop's run has heartbeats from before this process started either way — so the sharper signal cannot make the sleep case any worse than the window already does.

Arguments

  • client: An open PrefectClient.
  • dead_before: This process's own start time. Without it the sweep cannot act on a fast restart: a run killed seconds before the restart has a heartbeat well inside the liveness window, so it reads as healthy, the sweep leaves it, the reaper Crashedes it moments later, and nothing re-indexes that datasource until the next start. A last heartbeat predating this process settles it — that run cannot be executing in a process that did not exist yet.
  • own_pod_name: The pod this process serves, when it serves exactly one. dead_before is applied only to that pod's runs. Omitting it asserts this process owns every metadata run on this Prefect server — true of the orchestrator, whose Prefect server is its own child on a private port, and untrue of several pods sharing one sidecar server, where one pod starting would otherwise judge another's live runs dead.

Returns How many replacement runs were submitted.