Skip to main content

progress

How much work a background run has banked, and when to stop replaying it.

A recovered background run replays its whole DAG and relies on each step skipping the input it has already cached. That makes rows in the cache the only meaningful measure of what an attempt achieved: a run killed after processing 900 of 1,000 files leaves 900 rows behind, and its successor starts from there.

Bounding replay on progress rather than on raw attempts follows directly. A lineage that banks rows on every attempt is making headway however many times the laptop reboots, and stopping it would strand work no one else will finish. A lineage that banks nothing is either poisoned or looping, and each further attempt costs the same as the last. So the bound is on consecutive attempts that banked no progress.

Everything here is pure: progress_for_dag reads the cache and should_recover reads nothing at all. Neither polls, decides when to run, nor writes anything — the recovery poller owns that and consults these.

Module​

Functions​

progress_by_partition​

def progress_by_partition(    cache: CacheProtocol,    dag: BackgroundDAG,    task_hash: str | None = None,    project_id: str | None = None,) ‑> dict[tuple[str, int, str], int]:

Return the row count of every cache partition dag's steps write.

Keyed by (cache_table, cache_table_version, task_hash) so two steps that happen to share a partition are counted once, and so a caller can log which partition moved. Steps with no cache table contribute nothing: they hold their output in memory and hand it down a DAG edge, so a replay recomputes it for free.

A partition whose table cannot be read is counted as empty rather than raising. The table may legitimately not exist yet — nothing has written to it — and this measurement only ever feeds a decision that is safe to make conservatively: under-counting progress can end a lineage early, but it cannot cause a double run.

Arguments

  • cache: The cache instance holding the run's partitions.
  • dag: The DAG whose steps are being measured, already stamped by stamp_dag_partitions (unstamped steps fall back to task_hash, as the executor does).
  • task_hash: The run's datasource-level hash, used for any step left unstamped. Steps with neither are skipped.
  • project_id: The lineage's project, used to scope project-keyed partitions to this project's rows (see _partition_filters). None counts such a partition whole, which over-counts when another project shares it.

Returns Mapping of partition key -> row count.

progress_for_dag​

def progress_for_dag(    cache: CacheProtocol,    dag: BackgroundDAG,    task_hash: str | None = None,    project_id: str | None = None,) ‑> int:

Return how many rows dag's steps have banked in total.

A single number, compared only against the same number from an earlier attempt of the same lineage, which is what makes its absolute value unimportant. Rows are never deleted by a step, so the sequence is monotonic and a difference of zero genuinely means nothing was achieved.

What the difference cannot promise is that this lineage is what achieved it. An idle sibling sharing a partition offsets both readings equally and cancels out; one writing between two attempts does not, and its rows read as this lineage's progress. Passing project_id closes that for project-keyed partitions, which is the case where the borrowed rows are pure noise — this lineage can never read another project's eligibility rows.

It stays open for entity-keyed partitions, deliberately: a same-type datasource on the pod shares those, and its rows for a file this lineage also lists are real progress, since the replay now skips that file. Only its rows for entities this lineage does not hold inflate the reading, and telling the two apart would cost a file-list intersection per measurement. The residue is that a lineage stuck on its own can look productive while a busy sibling writes beside it, so should_recover is not on its own a bound on total replay cost — a caller that needs one wants an absolute attempt or wall-clock cap alongside.

Arguments

  • cache: The cache instance holding the run's partitions.
  • dag: The DAG whose steps are being measured.
  • task_hash: The run's datasource-level hash, for unstamped steps.
  • project_id: The lineage's project, scoping project-keyed partitions; pass it whenever it is known, since without it a sibling project's writes read as this lineage's progress.

Returns Total rows across the DAG's cache partitions.

should_recover​

def should_recover(    attempt: int, progress_history: Sequence[int], limit: int = 3,) ‑> RecoveryDecision:

Decide whether a lineage may be replayed once more.

Bounds consecutive attempts that banked no progress, not raw attempts: a lineage that keeps banking rows keeps its right to another attempt however many times the machine has gone down under it, and one that banks nothing stops after limit tries because each further attempt costs the same as the last and achieves the same.

Arguments

  • attempt: The attempt number that just ended — the recovered run would be attempt + 1. Used only in the reason text.
  • progress_history: Row counts, one per completed attempt, oldest first; see unproductive_streak. Empty means nothing is known about the lineage, which is not evidence of failure, so recovery is permitted.
  • limit: How many consecutive unproductive attempts to tolerate. A limit of zero or less disables recovery outright.

Returns The decision, carrying a reason fit to show whoever asks why a background task went quiet.

unproductive_streak​

def unproductive_streak(progress_history: Sequence[int]) ‑> int:

Return how many of the most recent attempts banked no progress.

progress_history is the total row count observed at the end of each attempt, oldest first. The count before the first attempt is taken as zero, so a lineage whose first and only attempt banked nothing has a streak of one.

A reading that goes down counts as unproductive rather than as an error: a step's rows cannot be deleted by a replay, so a fall means the measurement changed underneath us (a partition re-keyed by a config change, say), and reading that as progress would be the unsafe direction.

Arguments

  • progress_history: Row counts, one per completed attempt, oldest first.

Returns The length of the trailing run of attempts that banked nothing.

Global variables​

  • DEFAULT_UNPRODUCTIVE_ATTEMPT_LIMIT - How many consecutive attempts may bank no progress before a lineage is abandoned. Three, because the failure this feature exists for — a reboot mid run — banks progress, so an attempt that banks nothing is already the unusual case; two in a row is a coincidence and three is a pattern. The recovery poller takes the effective value from config.

Classes​

RecoveryDecision​

class RecoveryDecision(recover: bool, reason: str):

Whether to replay a lineage once more, and why not when not.

Arguments

  • recover: Whether another attempt is permitted.
  • reason: Why the decision went the way it did. Support-facing when recover is False — it becomes the give-up message on the flow_specs row and the final flow run — so it names the bound in plain terms rather than describing internals.

Variables​

  • static reason : str - Why the decision went the way it did.
  • static recover : bool - Whether another attempt is permitted.