Skip to main content

prefect_bootstrap

Idempotent Prefect bootstrapping for the metadata runtimes.

Every process that serves and triggers the file_metadata/scan_metadata deployments against a Prefect server must first ensure three things: a tag-based concurrency limit serialising cache writes, the zombie-flow-run reaper automation (self-hosted Prefect ships neither by default), and a heartbeat cadence fine enough for that automation to be accurate.

This is shared by the orchestrator (its embedded desktop-app server) and bitfount.scripts.run_pod (the sidecar server used in Docker), so the policy — and any future tuning of it — has one source of truth. The heartbeat frequency and the reaper's detection window are coupled (a window shorter than the cadence reaps healthy runs), which is exactly why they live in the same module.

What this module cannot ensure is that the reaper ever fires. Its pending detection window lives in Prefect's automation_bucket table and is swept on server restart, and it cannot witness a crash that took the server down in the same instant as the run — which is the normal shape of a laptop reboot or power cut here, since the server is a child of the same application as the pod. The client-side complement is bitfount.runtimes.recovery, which each process runs once at start-up: it judges liveness from the heartbeat stream rather than waiting to witness a transition, ends the runs that are provably gone, and resubmits them. The two are not redundant — the reaper handles the common case promptly while the process is up; recovery handles the case the reaper structurally cannot see.

Module​

Functions​

ensure_background_dag_rerun_concurrency_limit​

async def ensure_background_dag_rerun_concurrency_limit() ‑> None:

Idempotently create or update the background_dag_rerun limit.

An upsert rather than a create, so an operator who changes background_rerun_concurrency gets the new value on the next process start instead of being silently held at whatever the limit was first created with.

Must run before any rerun does: the flow body acquires its slot with strict=True, which raises when the limit is absent rather than running unbounded.

ensure_heartbeat_frequency​

def ensure_heartbeat_frequency() ‑> None:

Pin PREFECT_FLOWS_HEARTBEAT_FREQUENCY before any serve() call.

Liveness detection is only as good as the heartbeat cadence underneath it. Prefect's default is 180 s (prefect.settings.models.flows), so a process that leaves it unset emits heartbeats less often than the zombie window below — and every long-running flow gets reaped mid-flight as a false positive.

Must be called before the deployments are served, because serve() runs each flow in a subprocess that inherits this environment. Uses setdefault, so an explicit operator override still wins.

ensure_scan_chain_automation​

async def ensure_scan_chain_automation() ‑> None:

Idempotently register — and reconcile — the scan-chain automation.

Replaces the completion-callback chaining this once had, which rebuilt the scan trigger from an in-memory dict and was therefore silently lost when a restart mid-run emptied that dict. A server-side automation reacts to the file_metadata deployment's flow-run Completed event instead, so the chain survives the restart of whatever submitted the run.

Shared by both processes that serve these deployments, and not optional for either: scan_metadata_runtime declines to run while a file_metadata run is in flight (bitfount.runtimes.dedup.is_indexing_in_flight) precisely because this automation will run it once indexing completes. A server without it would turn every such skip into scan metadata that never gets collected.

Must be called after the scan deployment is served — its id is resolved here.

Reconciles rather than skips for the same reason ensure_zombie_automation does: the automation outlives any deploy, so an install upgraded from an older build holds a spec whose parameters predate only_paths, and skipping would leave every scoped file_metadata run chaining into an unscoped scan run.

Raises

  • Exception: Whatever the Prefect client raises. Callers treat a failure here as non-fatal, but they do the logging — the two callers word it differently.

ensure_zombie_automation​

async def ensure_zombie_automation() ‑> None:

Idempotently register — and reconcile — the zombie-reaper automation.

A killed flow run (e.g. a Windows Update reboot mid-walk, or a Docker container recycle) would otherwise sit at Running forever — nothing flips it. This registers a Proactive EventTrigger that fires when a flow run has gone silent for _zombie_window() and moves it to Crashed. See _desired_zombie_automation for why the trigger is shaped the way it is.

Unlike create_concurrency_limit, create_automation is not itself idempotent — calling it twice would create duplicate automations — so this is guarded by a name lookup (read_automations_by_name).

That guard cannot merely skip when a match is found, though: the automation lives in the Prefect server's DB and outlives any deploy, so a definition registered by an older version would otherwise persist forever and no fix to the spec above would ever reach an existing install. Instead we compare against the desired spec and update in place when it differs, which makes every startup self-healing.

liveness_silence_window​

def liveness_silence_window() ‑> datetime.timedelta:

How long a flow run may go silent before anything may presume it dead.

The public name for the window the reaper automation uses, exported so that client-side liveness checks derive "dead" from the same value rather than a second guess at it. A client that used a shorter window would call a run dead that the reaper still considers healthy; a longer one would refuse to act on a run Prefect had already marked Crashed.

This matters because the reaper is not sufficient on its own: its pending detection window is swept on server restart, so after a machine-wide outage a dead run is never marked Crashed and a client must reach the same verdict from the heartbeat stream itself

scan_chain_automation_registered​

async def scan_chain_automation_registered(client: PrefectClient) ‑> bool:

Whether the scan-chain automation is registered and enabled.

Two decisions elsewhere are only safe because the chain exists — recovery replaying file_metadata alone and expecting scan to follow, and scan_metadata_runtime skipping itself while indexing is in flight — and registration is best-effort at start-up. Both therefore ask rather than assume: an assumption that silently fails here drops scan metadata until someone restarts the process.

A disabled automation counts as absent: it will not fire, whatever the row says.

Arguments

  • client: An open PrefectClient.

Returns Whether an enabled automation is registered under SCAN_CHAIN_AUTOMATION_NAME. False if the question cannot be answered — the callers' fallbacks cost work, and being wrong the other way costs coverage.