Durable Run State and Scheduler Reconciliation

The scheduler is itself a stateful service, and it can fail. A process restart, a machine lost, a deploy that replaces the pod: in each case the scheduler comes back with no memory of what it was doing, and it must reconstruct enough of the run’s state to continue without dispatching work that already happened or abandoning work that is still running. The design that makes this possible is a durable record of intent, written before the action, combined with a loop that repeatedly compares the record against reality.

Write intent before action

The invariant is that no task is dispatched unless the record of its dispatch is already durable. The record is written first, in a store that survives the scheduler, such as a relational database, a log, or a coordination service, and only after the write commits does the scheduler hand the task to a worker.

This ordering turns a crash into a recoverable event. If the scheduler dies after the write and before the dispatch, the record says the task was dispatched but the task never ran. That is a false positive: the record claims something that did not happen. If the ordering were reversed, dispatch first and record second, a crash in the same window would produce a false negative, a task running with no record of it. The two errors are not symmetric. A false positive is detectable by the reconcile loop, because the scheduler can ask the worker whether it knows about the attempt and find that it does not, and can then re-dispatch. A false negative is invisible: the scheduler has no reason to look, and the task’s output may exist while the graph believes it was never started, or may be overwritten by a later attempt that the scheduler believes is the first.

Writing first and dispatching second costs one durable write per task dispatch. On a relational store that write is a transaction commit, whose latency is dominated by the storage layer’s flush: a millisecond on local storage, more when the store is remote and replicated. For a graph of a thousand tasks the overhead is measured in seconds over the run, which is negligible against the tasks themselves but is not negligible for a scheduler that must dispatch hundreds of short tasks per second.

The practical mitigation is batching: write the dispatch records for a set of ready tasks in one transaction, then dispatch all of them. The window widens, since a crash now leaves a batch of records for tasks not yet dispatched, but the recovery mechanism handles a batch the same way it handles one, so the only cost is more false positives to reconcile.

What the record contains

A minimal record per task attempt holds: the run identifier, the task identifier, the attempt number, the state, the time the state was entered, and the identity of the worker or the dispatch token if one was issued.

The attempt number is the piece that makes retries distinguishable from separate tasks. Without it, a task that failed three times and succeeded on the fourth looks like four separate facts about one task, and the history of what was tried and when is lost. With it, each attempt is a row, the task’s state is the state of the latest attempt, and a retry is an insert rather than an update.

The dispatch token is what allows the reconcile loop to distinguish a task that is running from a task that was dispatched into a void. It should be unique per attempt and known to both sides: the scheduler records it when it writes the intent, and the worker presents it when it reports progress or completion. A completion carrying a token the scheduler does not recognise is either a duplicate from a previous attempt or a message from a worker that lost its lease; in both cases the scheduler can reject it rather than accept a result it cannot attribute.

The state field needs to be an explicit enumeration rather than a boolean, and it needs a terminal distinction. queued, running, success, failed, skipped, upstream_failed, the last being distinct from skipped because it records why. Storing only success or failure loses the information that the trigger rules depend on, and a scheduler that has to recompute a skip from the graph on every read will recompute it differently after a code change to the trigger rules, which means the run’s history is no longer stable.

The reconcile loop

The loop runs on a timer, and each pass applies three checks.

The first concerns tasks recorded as ready whose dispatch was never confirmed. For each such task, the loop checks whether the worker knows about the attempt. If not, re-dispatch with a new token and mark the old attempt abandoned. If the worker does know, mark the attempt running and continue.

The second concerns tasks recorded as running whose worker has stopped reporting. This requires a liveness signal from the worker, and the mechanism is a lease or heartbeat with a deadline, discussed in worker leases and zombie tasks. A task whose lease expired is not necessarily dead, since the worker may be partitioned from the scheduler while still executing, so marking it failed and retrying risks two concurrent attempts of the same task. The choice between waiting and retrying is a trade between a stalled run and a duplicated execution, and the tiebreaker is whether the task is idempotent.

The third concerns the run’s overall state against its tasks. A run whose tasks are all terminal should be terminal itself. A run marked running with no non-terminal tasks is a bug in the state machine, and the reconcile loop’s job is to find and repair it rather than to leave the run stuck. This check is cheap, an aggregate query over the attempt table, and it catches a whole class of failures that would otherwise require an operator to notice.

-- Runs that claim to be active but have no task left to do.
SELECT r.run_id, r.logical_date
FROM runs r
WHERE r.state = 'running'
  AND NOT EXISTS (
    SELECT 1 FROM task_attempts a
    WHERE a.run_id = r.run_id
      AND a.state IN ('queued', 'running')
  );

The query is a full scan of the runs table unless the state column is indexed, which is the usual reason schedulers keep active runs in a separate table or a separate partition from completed ones. A scheduler that stores every run it has ever executed in one table and scans it on every reconcile pass degrades as history accumulates, and the degradation appears as a scheduler that becomes slow over months rather than as an obvious fault.

The idempotence of the loop itself

The reconcile loop must be safe to run concurrently with itself, because a scheduler that restarts may briefly have two instances running, and because a loop pass that takes longer than the interval will overlap the next one. Every action the loop takes must be conditional on the state it observed: re-dispatch only if the attempt is still in the state that prompted the re-dispatch, using a compare-and-set on the attempt record. Without that condition, two loop instances both see an unconfirmed dispatch and both re-dispatch, producing the duplicate the loop exists to prevent.

Compare-and-set on the attempt record is a single-row conditional update, and its cost is the same as any other row update. The harder part is that the loop’s observation and its action are not atomic: the loop reads the worker’s state, then writes the attempt record, and the worker’s state may change in between. The write must therefore be conditioned on the attempt’s version or state, not on the worker’s, so that a worker completion that lands in the gap wins and the loop’s re-dispatch loses. This is the standard optimistic concurrency pattern, and it fails safely: the loser retries on the next pass with fresh state.

The ordering also means the loop can be slow without being wrong. A pass that takes ten seconds and finds nothing to do is wasted work but not a correctness problem, provided the pass’s writes are conditional. That property is what allows the loop’s interval to be tuned for cost rather than for latency, which matters because a large deployment’s reconcile pass may touch thousands of rows.

What durable state does not record

The record describes what the scheduler did, not what the task did. A task that wrote half its output before the scheduler lost contact leaves a record saying “running, lease expired” and an artefact that is neither complete nor absent. Recovery must decide what to do with that artefact, and the scheduler cannot help, because it never knew what the task’s output looked like. The only defence is the task’s own commit protocol: write to staging and publish atomically, so that an interrupted attempt leaves nothing visible at the target path. With that in place, the scheduler’s decision to re-dispatch is safe regardless of how the previous attempt died.

The second limit is that durable state records the schedule’s view of time, not the data’s. A run marked successful means its tasks reported success, not that the output matches the expected row count or that the source was complete. Verifying that requires reading the data, which is a different system with different failure modes. The scheduler’s state is a claim about execution, and the gap between execution and correctness is bridged by validation tasks rather than by more scheduler bookkeeping.

What durable state buys is the ability to answer, after any crash, which tasks have been attempted, which are in flight, and which have never started, answering it from a store that outlives the process. Everything the scheduler does afterwards is a function of that answer. Once the answer is reliable, the next question is where the run’s notion of “which day am I processing” comes from, which is the subject of logical dates and catchup.