Skip to main content

dedup

Prefect-native liveness and dedup for background work.

Two kinds of run rely on this module, and they differ in how they are addressable.

Deployment runs (file_metadata, scan_metadata) are triggered as fire-and-forget Prefect deployment runs, tagged at submission with task_hash:<task_hash> (see task_hash_tag) and run_type:<file_metadata|scan_metadata> (see run_type_tag). Before submitting a new run, callers query Prefect for a live flow-run carrying the same task_hash tag on the same deployment and skip submission if one is found.

Background DAG runs execute in-process and so have no deployment_id to filter on. They are addressed purely by their lineage tags (see background_dag_tags), because a flow name is dag.name — shared by every project and datasource using the same template, and therefore useless as a key.

This makes Prefect's own flow-run state the source of truth, replacing the Bitfount runs table's stale-running gate. But state alone is never enough: the zombie-reaper automation (bitfount.runtimes.prefect_bootstrap.ensure_zombie_automation) cannot observe a crash that took the Prefect server down with it, and its pending detection window does not survive a server restart, so a machine-wide outage leaves a dead run sitting at RUNNING forever. Both liveness queries are therefore state plus heartbeat recency over RUNNING runs — see is_flow_run_lost, is_flow_run_active and is_background_flow_run_active. There is exactly one definition of "dead" in the codebase, sized from prefect_bootstrap.liveness_silence_window(), so poller, reaper and dedup cannot drift into disagreeing about it.

Judging liveness is all this module does. Acting on a dead run — ending it in Prefect, failing its bookkeeping row, replaying it — is bitfount.runtimes.recovery for the metadata runtimes and bitfount.flows.dag.recovery for background DAG runs.

Tagging​

Tags are the only identity a flow run carries, so every run this system creates gets run_type: and, where one exists, task_hash:. There are two ways they arrive, and the rule is:

Tag at creation if you control creation; the in-flow self-tag (ensure_flow_run_tags) exists only because Prefect's RunDeployment automation action cannot.

Creation-time tags are what let the liveness queries above see a run before its body starts — a run tagged only from inside itself is invisible to dedup for as long as it sits SCHEDULED/PENDING, which under a concurrency limit can be a while. Prefect unions deployment-level tags with per-run tags server-side, so the two routes compose without duplicates.

Module​

Functions​

attempt_from_tags​

def attempt_from_tags(tags: list[str]) ‑> int | None:

Return the attempt number carried in tags, or None if absent.

Each recovery attempt is a new flow run carrying its own tags rather than a mutation of the previous one's — Prefect exposes no tag-mutation action — so reading the lineage back means parsing it off the run.

Returns None for an untagged or malformed run rather than raising: a run whose lineage cannot be read is one this code did not create, and the caller treats that as "not part of a lineage" rather than an error.

attempt_tag​

def attempt_tag(attempt: int) ‑> str:

Return the flow-run tag encoding this run's position in its lineage.

background_dag_tags​

def background_dag_tags(    task_hash: str, project_id: str, datasource_name: str, attempt: int = 1,) ‑> list[str]:

Return the full tag set identifying one background DAG flow run.

Arguments

  • task_hash: The (pod, datasource) task hash.
  • project_id: Project the datasource is linked to.
  • datasource_name: Datasource the run is processing.
  • attempt: Position in the recovery lineage; 1 for the run a dataset-project link started.

Returns The tags to attach to the Prefect flow run.

datasource_tag​

def datasource_tag(datasource_name: str) ‑> str:

Return the flow-run tag encoding datasource_name.

ensure_flow_run_tags​

async def ensure_flow_run_tags(*tags: str) ‑> None:

Add tags to the flow run this coroutine is executing inside.

The in-flow half of the tagging rule in this module's docstring: a run whose creator could not tag it labels itself once its body starts. The only creator in that position is Prefect's RunDeployment automation action, which has no tags field at all — so the scan_metadata runs spawned by the scan-chain automation arrive with an empty tag set and no way for the automation to fix it.

Best-effort by construction. Tagging is an observability affordance; a run that indexed a datasource successfully but failed to label itself has done the job, and raising here would turn a cosmetic failure into a lost run.

The client is opened here, inside the same try, rather than taken as an argument — because a caller opening one for us puts the riskiest part of the operation outside the guarantee above. A flow that did that had this tagging call at the top of its body, so an unreachable Prefect server failed the run before it did any work, three times over (retries=3), and the pod's datasources got no schemas at all. Nothing about labelling a run should be able to do that.

Arguments

  • *tags: Tags to ensure are present. Already-present tags are a no-op.

ensure_flow_run_tags_sync​

def ensure_flow_run_tags_sync(*tags: str) ‑> None:

Synchronous ensure_flow_run_tags, for use from a sync flow body.

The metadata runtimes' flows are sync (def file_metadata_runtime), so they cannot await. Prefect's own sync client is used rather than driving the async one, so this never touches the ambient event loop the flow is running on.

Arguments

  • *tags: Tags to ensure are present. Already-present tags are a no-op.

is_background_flow_run_active​

async def is_background_flow_run_active(    client: PrefectClient,    task_hash: str,    project_id: str,    silence_window: timedelta | None = None,    exclude_flow_run_id: uuid.UUID | None = None,) ‑> bool:

Return True if a genuinely live background DAG run already exists.

The tags-only counterpart to is_flow_run_active: in-process runs have no deployment_id to filter on, so the lineage tags are the whole key.

Stale-aware, and that is load-bearing. A bare state check would treat a permanently-RUNNING zombie as live and reject every subsequent dataset-project link as a duplicate, wedging that (task_hash, project_id) indefinitely. Cache-level dedup avoided this with a staleness threshold; a Prefect-level guard has to earn it back explicitly

Arguments

  • client: An open PrefectClient.
  • task_hash: The (pod, datasource) task hash to check for.
  • project_id: Project the run belongs to. Two projects on one datasource share a task_hash and must not dedup against each other.
  • silence_window: Passed through to is_flow_run_lost.
  • exclude_flow_run_id: A run to leave out of the answer. Required when the caller is one of the runs being counted: a flow asking this from inside its own body carries the same lineage tags and a heartbeat seconds old, so without the exclusion it would find itself, call itself a duplicate, and no background run could ever execute.

Returns Whether a live background DAG run for this pair was found.

is_flow_run_active​

async def is_flow_run_active(    client: PrefectClient,    deployment_id: uuid.UUID,    task_hash: str,    silence_window: timedelta | None = None,    dead_before: datetime | None = None,) ‑> bool:

Return True if a genuinely live flow-run for task_hash already exists.

"Live" means a flow-run of deployment_id tagged task_hash:<task_hash> that is either pre-RUNNING (SCHEDULED/PENDING) or RUNNING with a heartbeat inside the liveness window.

Arguments

  • client: An open PrefectClient.
  • deployment_id: The id of the deployment to scope the check to.
  • task_hash: The (pod, datasource) task hash to check for.
  • silence_window: Passed through to is_flow_run_lost.
  • dead_before: Passed through to is_flow_run_lost. A start-up sweep must supply the same value it discovered the run with, or it re-judges its own orphan by the window alone and calls it live — see bitfount.runtimes.recovery.

Safe to leave unset, and safe for a long-lived caller to pass: the cutoff is max(window cutoff, dead_before), so once a process is older than the window its own start time is inert.

Returns Whether a live flow-run for this task_hash was found.

is_flow_run_lost​

async def is_flow_run_lost(    client: PrefectClient,    flow_run: FlowRun,    silence_window: timedelta | None = None,    dead_before: datetime | None = None,) ‑> bool:

Return True when flow_run's process is provably gone.

Two ways a lost process presents, and both count:

  • Prefect already says CRASHED — the reaper saw the silence and acted. This is what happens when the application stayed up and only the run died.
  • Prefect says RUNNING but the last heartbeat is older than silence_window. This is what happens after a reboot or power cut, where the Prefect server died in the same instant as the run, nothing was alive to notice the silence, and the reaper's pending window was swept on restart. Left to Prefect alone such a run stays RUNNING forever. A pre-RUNNING run is reported not lost, and callers must not ask about one: there is no heartbeat to judge and its age says nothing useful (see BACKGROUND_ACTIVE_STATE_TYPES). Callers query RUNNING/CRASHED only, so such a run never reaches here.

A run with no heartbeat events at all falls back to its start_time age. Because the heartbeat cadence is pinned and the first beat is emitted immediately, a genuinely live run with no events is necessarily seconds old, so the same window is safe. The fallback also covers a run whose heartbeats aged out of retention — a start_time that old is unambiguously stale.

This says nothing about ownership: a suspended laptop's own healthy run also looks lost, because its heartbeat thread suspended with it. Callers must additionally exclude runs they are themselves executing

Arguments

  • client: An open PrefectClient.
  • flow_run: The run to judge.
  • silence_window: How long silence may last before the run is presumed dead. Defaults to the same window the reaper automation uses, so the two cannot disagree about what dead means.
  • dead_before: An instant before which any last signal proves the run is gone, regardless of the window. Raises the cutoff to max(window cutoff, dead_before); None leaves the window alone.

For a caller that owns every run it is asking about, its own process start time is exactly such an instant: a run whose last heartbeat predates this process cannot be running in this process, so it is dead however recent that heartbeat looks. That is what lets a start-up sweep recover a run orphaned seconds ago, which the window alone cannot — a run killed 5s before a restart has a heartbeat well inside 90s and is indistinguishable from a healthy one by age.

It is not a substitute for the window and must not be used as one. The window is what identifies a dead run this process never owned — a sibling pod's, say — and it is also the only thing standing between a start-up sweep and a run this boot has just created: such a run's heartbeats postdate the process, so the raised cutoff spares it while a blanket "any RUNNING at boot is dead" rule would not.

Returns Whether this run's process should be presumed gone.

is_indexing_in_flight​

async def is_indexing_in_flight(client: PrefectClient, prefect_task_hash: str) ‑> bool:

Return whether file_metadata indexing is running for prefect_task_hash.

Anything that reads the file inventory has to ask this first, because a live indexer is building that inventory: a reader that starts anyway sees a partial one, banks whatever it found and finishes COMPLETED, which is under-coverage wearing the costume of success. Two callers, for the same reason: bitfount.flows.dag.recovery defers replaying a DAG (the next tick finds the run again), and scan_metadata_runtime skips its own body (the indexer's Completed event chains a fresh scan). Uses the same deployment-scoped check file_metadata_refresh uses to avoid double-indexing.

A missing deployment answers False: nothing can be indexing if nothing is registered to index, and the pod's own indexing path would have registered it.

Arguments

  • client: An open PrefectClient.
  • prefect_task_hash: The datasource-level task hash to check.

Returns Whether indexing is in flight for this datasource.

last_heartbeat_at​

async def last_heartbeat_at(    client: PrefectClient, flow_run_id: uuid.UUID,) ‑> datetime | None:

Return when flow_run_id last emitted a heartbeat, or None if never.

Prefect 3.x records no heartbeat timestamp on the flow-run row — _emit_flow_run_heartbeat only publishes an event — so flow_run.updated tracks the last state transition and is hours stale on a perfectly healthy long run. The event stream is the only real per-run liveness signal, and this query is backed by a purpose-built (event, resource_id, occurred) index.

None has two causes and the caller must not conflate them with "alive": the run died before its first heartbeat, or its heartbeats aged out of the events retention window (7 days by default). is_flow_run_lost handles both by falling back to the run's own start time.

Arguments

  • client: An open PrefectClient.
  • flow_run_id: The flow run to look up.

Returns The occurred timestamp of the most recent heartbeat, or None.

pod_tag​

def pod_tag(pod_name: str) ‑> str:

Return the flow-run tag encoding pod_name.

project_id_tag​

def project_id_tag(project_id: str) ‑> str:

Return the flow-run tag encoding project_id.

run_type_tag​

def run_type_tag(run_type: str) ‑> str:

Return the flow-run tag encoding run_type (e.g. "file_metadata").

tag_value​

def tag_value(tags: Sequence[str], prefix: str) ‑> str | None:

Return the value carried by the prefix tag in tags, or None.

Tags are the only lineage a background flow run carries — a flow name is dag.name, shared across templates — so reading a run's identity back means parsing them off it. None for a run that has no such tag, which the caller should read as "this run is not one of ours" rather than as an error.

Arguments

  • tags: The flow run's tags.
  • prefix: The tag prefix to look for, e.g. TASK_HASH_TAG_PREFIX.

Returns The remainder of the first matching tag, or None.

task_hash_tag​

def task_hash_tag(task_hash: str) ‑> str:

Return the flow-run tag encoding task_hash.

Global variables​

  • ACTIVE_STATE_TYPES : list[prefect.client.schemas.objects.StateType] - Flow-run state types that count as "still active" for dedup purposes. AwaitingConcurrencySlot is a state name whose type is PENDING, so it is covered without a separate entry.
  • BACKGROUND_ACTIVE_STATE_TYPES : list[prefect.client.schemas.objects.StateType] - The cost is the microseconds between a genuine sibling being created and reaching RUNNING: two triggers landing in that window could both proceed. That is the cheaper failure by a wide margin — a duplicate background run re-upserts the same partitions (see flows/dag/CONTEXT.md), whereas a wrongly-skipped run means the work silently never happens.
  • POD_TAG_PREFIX - Identifies the pod a run belongs to, for runs that have no task_hash because they are pod-wide rather than per-datasource (schema-manager).
  • PROJECT_ID_TAG_PREFIX - Tag prefixes carrying a background DAG run's lineage. A flow name is dag.name, shared across every project and datasource using the same template, so these tags are the only thing that identifies a run.
  • TASK_HASH_TAG_PREFIX - Tag prefixes applied to every triggered file_metadata/scan_metadata flow-run.