Skip to main content

discovery

Reading recoverable flow runs off a Prefect server.

Shared by the two recovery sweeps — bitfount.runtimes.recovery for the metadata deployment runs and bitfount.flows.dag.recovery for the in-process background DAG runs. Those two are deliberately separate, because the execution model decides everything about them: how a run is found (by deployment_id or by lineage tags), how it is replayed (verbatim from its own parameters, or rebuilt from a stored flow spec), what else has to be reconciled, and what bounds the whole thing. What they do share is the query, and this module is that shared part — extracted after both grew the same paging loop, the same two constants and the same subtle rule about which of two runs is later.

Two things here are easy to get wrong in a way nothing notices:

Paging. read_flow_runs with no limit does not mean "all of them": the server applies PREFECT_SERVER_API_DEFAULT_LIMIT, 200 by default. Both sweeps query the entire crash history of something, and nothing prunes terminal runs, so past 200 rows they silently stopped seeing whole lineages — and the supersession checks then read a stale row as the latest and declined to replay the newer crash they had never seen.

Ordering. "Which of these two runs is later" has to be answered by expected_start_time and nothing else. A run's attempt: tag looks like an ordering and is not: it is a label, and a replay that derived its number from an older crash repeats one. Where two guards ranked the same runs by different keys they each deferred to the other's loser and dropped the lineage between them.

Module​

Functions​

is_later​

def is_later(candidate: datetime | None, than: datetime | None) ‑> bool:

Whether candidate is the later of two runs' expected_start_times.

Unknown either side answers False — "not later", so not a successor. That asymmetry is deliberate: a replay that was not needed is a duplicate the submission-time dedup absorbs, while a replay wrongly withheld is a lineage that stops for good.

Arguments

  • candidate: The expected_start_time being tested.
  • than: The expected_start_time to beat.

Returns Whether candidate is strictly later.

read_all_pages​

async def read_all_pages(    client: PrefectClient, flow_run_filter: FlowRunFilter, description: str,) ‑> list[FlowRun]:

Return every run matching flow_run_filter, walking the pages.

Sorted newest-first, so a fixed offset means the same row between requests: the set being paged is dominated by terminal runs, which do not move.

Arguments

  • client: An open PrefectClient.
  • flow_run_filter: The filter to page through.
  • description: What is being read, for the truncation warning.

Returns The runs, newest first.

Global variables​

  • CRASHED_HORIZON - How far back a CRASHED run is worth looking at. Prefect keeps terminal runs indefinitely and nothing prunes these, so without a bound the discovery query grows for the lifetime of the installation. A crash older than this has been superseded many times over — the refresh cron alone re-indexes daily — so the supersession checks would discard it anyway, at the cost of a query per row. RUNNING is deliberately not bounded by this: a run wedged at RUNNING since long before the horizon is exactly what the sweeps exist to find.
  • MAX_PAGES - Most pages a single query will walk. A backstop against paging forever if the result set is being appended to as fast as it is read; hitting it is logged, because a truncated sweep must never read as full coverage.
  • PAGE_SIZE - Rows per page. Matches Prefect's own PREFECT_SERVER_API_DEFAULT_LIMIT default, which is what an unpaginated query silently got.