Checkpointing and barriers¶

How clink takes a globally consistent snapshot of a running job by flowing in-band barriers through the dataflow, snapshotting state on barrier receipt, and committing two-phase-commit sinks once every subtask has acked.

Overview¶

A checkpoint is a globally consistent cut across a running job: every operator's keyed state, every source's read position, and every sink's pre-committed output, all corresponding to the same logical point in the stream. clink takes that cut with markers called barriers that travel in-band with the data. A coordinator periodically injects a barrier into the sources; each operator snapshots its state when the barrier passes through and acknowledges; once every subtask has acked, the checkpoint is complete and can be used as a recovery point. This is the substrate for fault tolerance and rescale (see fault-tolerance-and-rescale.md); this page documents how the cut itself is produced and made durable.

Where it lives¶

Path	What it is
`include/clink/checkpoint/checkpoint_barrier.hpp`	`CheckpointBarrier`: the in-band marker (id, terminal flag, alignment `Mode`)
`include/clink/checkpoint/checkpoint_coordinator.hpp`	`CheckpointCoordinator`: barrier creation, per-operator ack tracking, periodic trigger (local / single-process path)
`src/checkpoint/checkpoint_coordinator.cpp`	Coordinator implementation
`include/clink/runtime/multi_input_alignment.hpp`	`MultiInputAlignment`: the Chandy-Lamport alignment state machine for N-input operators
`include/clink/runtime/snapshot_worker.hpp`	`SnapshotWorker`: off-thread durable-write + ack
`include/clink/state/durable_file_write.hpp`	`write_fsync_rename` and `CLINK_STATE_FSYNC`
`include/clink/runtime/dag.hpp`	Barrier flow through source / single-input / multi-input / sink runners; the snapshot-on-barrier and unaligned-capture logic
`include/clink/operators/operator_base.hpp`	Source `snapshot_offset`/`restore_offset`/`notify_checkpoint_complete`/`notify_checkpoint_aborted`/`inject_pending_barrier`; sink `on_barrier`/`on_commit`/`on_abort`/`commit_group` hooks
`include/clink/connectors/file_2pc_sink.hpp`	`FileSink2PC<T>`: the canonical two-phase-commit sink
`include/clink/connectors/parquet_2pc_sink.hpp`	`ParquetSink2PC<T>`: a fsync-durable Parquet 2PC sink
`src/cluster/coordinator.cpp`	Cluster-side ack tracking, `COMPLETED-N` marker, `CommitCheckpoint`/`AbortCheckpoint` broadcast, commit-group gating

The coordinator in checkpoint/ drives the in-process (single-LocalExecutor) path. In a cluster the Coordinator owns the equivalent ack-tracking and commit-broadcast logic; the per-operator runner mechanics in dag.hpp are identical on both paths.

How it works¶

The barrier¶

CheckpointBarrier is a value type carrying three fields: a CheckpointId, a terminal flag, and an alignment Mode (Aligned or Unaligned). It travels as a StreamElement on the same channels as data and watermarks, so it observes the same FIFO ordering as records. The barrier defines an epoch boundary: every record that precedes the barrier on a channel belongs to checkpoint N, and everything after belongs to N+1.

A barrier is terminal when a source emits it after produce() returns false. Terminal barriers flow downstream like any other barrier, but sinks treat them as both pre-commit and commit in one step (there is no recovery scenario past end-of-stream), so they finalise locally with no coordinator round-trip.

The coordinator: barrier creation and ack tracking¶

CheckpointCoordinator is the single source of truth for the in-process checkpoint lifecycle. Operators register as participants via register_operator(OperatorId). trigger() allocates the next CheckpointId under a lock, records the full set of registered operators as the checkpoint's pending-ack set in in_flight_, stamps the configured default_mode (or a per-trigger override, or a ModeResolver decision) onto a fresh barrier, and returns it.

acknowledge(id, op) removes one operator from that checkpoint's pending set. When the set empties, the coordinator calls backend_->snapshot(id), records last_completed_, fires the OnComplete callback, and drops the in-flight entry. abort(id, reason) marks an in-flight checkpoint aborted (idempotent) and fires OnAbort. An ack for an unknown or already-aborted id is ignored.

start_periodic_trigger() spins a background thread that fires trigger() every interval and pushes the resulting barrier into each bound source injector (set_source_injectors). An interval of 0 disables periodic checkpointing.

Injecting barriers into sources¶

Barriers enter the dataflow only at sources, and only between produce() calls, so there is no race between record emission and offset capture. The mechanism (add_source in dag.hpp, and the hooks on Source in operator_base.hpp):

The coordinator (or, in a cluster, the Coordinator's trigger loop) calls a per-source injector, which calls source->inject_pending_barrier(b). This appends the barrier to a mutex-guarded pending_barriers_ queue on the source.
The source runner loop drains that queue between produce() calls via drain_pending_barriers(). For each pending barrier it: calls source->snapshot_offset(backend, op_id, ckpt_id) to record the source's read position into the live backend, then emitter.emit_barrier(b) to push the barrier downstream. The source runner acks only a delivery FAILURE (the diagnostic for a closed downstream channel, which would otherwise wedge the checkpoint silently). The success ack is deliberately not this runner's to send: snapshot_offset is an in-memory put, and durability belongs to the subtask's terminal runner - the network output sink attach_typed_output_groups adds (or the real sink of an in-process dag) - which captures the shared backend when the barrier reaches it and acks only after persist() returns. The source runner used to ack success right here, which let the coordinator publish COMPLETED-N while the subtask's snapshot was still a .part temp file; a SIGKILL in that window produced a completed checkpoint whose participant snapshot did not exist, and the restore refused it. The invariant - an ok-ack implies the acked checkpoint's snapshot is already on disk - is pinned by tests/test_checkpoint_ack_durability.cpp.

Because draining runs on the same thread as produce(), the persisted offset and the emitted barrier reach durability atomically with respect to the record stream: every record emitted before the barrier was emitted before snapshot_offset ran. One subtlety survives the same-thread drain: the DURABLE capture happens on the terminal runner's thread when the barrier reaches it, and by then this loop may have drained later barriers over the same offset slot - checkpoint N's file would record checkpoint N+1's offset, and a restore of N would silently skip the records in between (found live via clink state-cat: checkpoint 4 holding offset 12 while its committed output ended at record 8). The runner therefore calls StateBackend::stage_operator_rows(op, ckpt) right after snapshot_offset, pinning a barrier-consistent copy of the source's rows for that checkpoint id; snapshot(id) serialises the staged copy in place of the live rows and releases everything staged at or below id. The runner stamps each barrier with the job-global mode derived from JobConfig::unaligned_checkpoints (or a per-operator override) so sources stay mode-agnostic and downstream operators read the policy off the barrier itself.

Snapshot-on-barrier at a single-input operator¶

When a single-input operator runner pops a barrier and a state backend is present (add_operator in dag.hpp):

Only the chain's checkpoint owner (the most-downstream operator sharing the backend in a fused chain) snapshots and acks. Non-owners stage their timer slice into the shared backend and forward the barrier. This keeps a shared backend single-writer per barrier, which a delta-commit backend such as RemoteReadBackend requires.
Job resume across a coordinator restart is an HA-dir feature. recover_persisted_jobs() reads job manifests from <ha_dir>/jobs/ and is called only from the leadership callback; with no --ha-dir it returns immediately. A plain coordinator restart therefore abandons every running job: the COMPLETED-N markers under the checkpoint directory preserve a restore point, but nothing re-submits the job. Resubmitting with --restore-from-checkpoint-id=N resumes manually; a single coordinator started WITH --ha-dir self-acquires leadership and recovers its jobs automatically on restart. Established, not assumed: a compound-failure test that expected checkpoint_dir alone to suffice converged on nothing, and the respawned coordinator's log showed no deploy at all (FaultRecoveryTest.CoordinatorAndWorkerDyingTogetherStillCommitsExactlyOnce now runs on the real contract).
Triggers are generation-fenced across rescale transitions. A TriggerCheckpoint carries the state generation it was issued for; a worker drops a mismatched one at receipt AND at the queued-trigger replay that fires when a job's first sources register. Without that fence, a trigger straddling a rescale was queued against the vanished topology and replayed into the new one, which snapshotted old checkpoint ids into the new generation's directories - the follow-up 49 leftover, reproduced at 120+ per run with the transition window held open by a fault point (RescaleExactlyOnceTest.Holding*WindowOpen*). The coordinator also skips triggering for jobs mid-restart, which spares real multi-tick transitions the aborted checkpoints the straddlers caused. Zero generation means a pre-fence coordinator and is accepted, so mixed-version clusters keep the old behaviour rather than stalling.
Non-owners quiesce until the owner captures. Chain members are separate runner threads, so after forwarding barrier N a non-owner holds at a per-chain rendezvous (ChainBarrierEpoch) until the owner's capture of N completes. Without this, a non-owner's next records landed in the shared backend before the owner - a later thread - captured, and a completed checkpoint held more than its cut: a keyed counter one record ahead of the source's recorded offset, which a restore then replays and double-counts. The owner never waits (no cycle), marks for every outcome including terminal barriers, and every chain runner's cancel aborts the epoch so teardown releases any waiter. Deterministic regression: tests/test_chain_barrier_quiescence.cpp.
The owner snapshots its timers (snapshot_timers), then snapshots state, then forwards the barrier downstream by calling op->process() (so any user on_barrier hook runs and the barrier reaches the output channel), then acks.
If the operator is on the async-state path, the runner first calls aec->drain_for_barrier() to bring all in-flight async reads to quiescence so the captured cut reflects every record admitted before the barrier (no torn state). See async-state-execution.md.

The async snapshot worker¶

The expensive part of a checkpoint is the durable write, not building the in-memory state slice. SnapshotWorker (snapshot_worker.hpp) splits the two so record processing runs ahead of disk I/O while preserving the ack-after-durable invariant.

A worker is constructed per operator subtask only when the backend supports_async_persist() (FileBacked and disk-backed changelog; InMemory, RAM-only changelog and RocksDB stay on the synchronous path and never build one). The split:

sequenceDiagram
  participant Op as Operator thread
  participant SW as Snapshot-worker thread
  Note over SW: blocks on queue.pop
  Op->>Op: pop barrier
  Op->>Op: capture(ckpt_id) returns CaptureHandle
  Op->>Op: forward barrier downstream
  Op->>SW: enqueue(handle, backend, ack)
  Op->>Op: continue processing records
  SW->>SW: persist(handle) (slow durable write)
  SW->>Op: ack(ckpt_id, ok, err) (after persist returns)

StateBackend::capture(ckpt_id) produces a detached point-in-time blob cheaply on the operator thread; the barrier is forwarded immediately because the blob already reflects state at the barrier point. The worker calls StateBackend::persist (the slow write_fsync_rename) on its own thread and fires the ack only after persist returns, so an async checkpoint is never reported durable before its bytes are on stable storage.

The queue is FIFO, single-consumer, capacity 1: an operator may have at most one captured-but-not-persisted checkpoint queued behind the one being written. enqueue blocks once the worker falls a checkpoint behind, which bounds how far processing runs ahead of durability and provides backpressure for free. Teardown distinguishes a clean end-of-stream (drain_and_join persists and acks the backlog so a checkpoint the coordinator awaits still completes) from a cancel (cancel_and_join drops not-yet-started captures without acking; an un-acked checkpoint is simply never marked complete). If capture itself throws on the operator thread, the runner acks the failure inline.

Barrier alignment at multi-input operators¶

A multi-input operator (join, co-process, union) receives the same barrier on each input channel, but not at the same time. MultiInputAlignment is the per-operator state machine that decides when to forward the barrier and which inputs to pause. The operator drives it by calling on_barrier(input_index, barrier) and reads back a BarrierAdvance { forward, barrier, unaligned_first }. The barrier's stamped Mode, pinned on first delivery of a given id, selects behaviour per-checkpoint.

Aligned mode (default, Chandy-Lamport)¶

input 0:  -- r r r || ---------------------  (barrier arrives, input 0 PAUSED)
input 1:  -- r r r r r r || ----------------  (records before barrier still flow)
                         ^ barrier on every alive input -> ALIGNED
forward:  -- r r r r r r || ----------------  barrier forwarded, inputs unpaused

On the first input to deliver barrier N, that input is paused (input_paused(i) returns true and the runner stops polling it). Records still arriving on the other inputs are records that precede the barrier on those channels and so belong to checkpoint N; they are processed normally. When every alive input has delivered N, check_alignment_ forwards the barrier (preserving the stamped mode), unpauses all inputs, and records the align-wait time as a metric. Closed inputs implicitly satisfy any barrier and contribute Watermark::max() to the watermark min, so a finished input never wedges alignment. In-flight records are not persisted in aligned mode, because by the time the barrier forwards every input has reached it.

Unaligned mode (barriers overtake in-flight records)¶

input 0:  -- r r r || ---------------------  (barrier arrives FIRST)
                   |  forward immediately, no pause
input 1:  -- r r r [a b c] -----------------  (a b c are in-flight, NOT yet consumed)
forward:  -- r r r || ----------------------  barrier forwarded on first delivery
capture:           +- drain input 1's in-flight [a b c] into snapshot state

When the barrier is stamped Unaligned, the first input's delivery forwards the barrier immediately and never pauses, and the advance carries unaligned_first = true. The runner then drains the not-yet-delivered inputs' in-flight records and serialises them into the state backend under a per-operator key (for example the interval join writes __interval_join_left_inflight__ / __interval_join_right_inflight__ via serialize_records_). On restart those buffers are read back and replayed through local pending queues before the main poll loop resumes, so the records that the barrier overtook are not lost. This trades larger snapshots for faster checkpoint completion under backpressure. Subsequent deliveries of the same barrier id are absorbed silently and the alignment bookkeeping is GC'd once every alive input has been accounted for.

A union operator carries no state, so under unaligned mode it just lets the barrier overtake: the records still queued on the other inputs are forwarded on later iterations and, from the downstream operator's perspective, arrive after the barrier and so belong to the next epoch. Capture happens at the downstream stateful operator, not at union.

Per-checkpoint mode is decided by the first delivery and pinned for that id, so aligned and unaligned semantics never mix mid-flight for one checkpoint even if a later same-id delivery carried a different stamp. apply_barrier_mode_override lets a per-operator JobConfig override re-stamp a barrier (for example to force a stateful operator that does not implement in-flight capture to stay aligned while the rest of the job runs unaligned). Async multi-input operators force-align the popped/in-flight tail via drain_for_barrier while keeping the unaligned fast path for the unpopped other-channel records; an operator that cannot capture in-flight records is gated by can_unalign so an unaligned barrier degrades to aligned rather than losing data.

Adaptive mode (measured pressure decides per checkpoint)¶

CheckpointAlignment::Adaptive puts the aligned-versus-unaligned decision on measured pressure instead of a static flag, one checkpoint at a time. The decision discipline lives in checkpoint::AdaptiveModePolicy (include/clink/checkpoint/adaptive_mode_policy.hpp): an observation window is one checkpoint interval, a window is pressured when its normalised signal crosses pressure_threshold, switching to unaligned takes windows_to_unaligned consecutive pressured windows and switching back takes windows_to_aligned consecutive calm ones - so a one-window spike never flips the mode and the policy cannot oscillate faster than the configured runs. The recent observations are kept in a bounded history for diagnostics only.

The two runtimes feed the policy different signals through the same seam. In process, CheckpointCoordinator::enable_adaptive_mode(pressure_fn, cfg) installs the policy behind the existing set_mode_resolver hook; Dag::channel_pressure_fn() supplies the signal (maximum depth/capacity occupancy across the runner input channels, the same probes the LocalExecutor's metrics loop polls). In the cluster, the coordinator's trigger sweep observes the LAST completed checkpoint's duration relative to the configured interval - alignment stalls under backpressure are exactly what stretches it - and stamps the decision on the TriggerCheckpointMsg (barrier_mode_plus1, a trailing wire field; 0 from an older coordinator means "not stamped" and the worker keeps its deploy-static behaviour). An adaptive deploy sets DeployMsg::adaptive_barrier_mode, and sources then forward the injected barrier's stamp untouched instead of re-stamping the static mode; per-operator overrides still win. The EOS final checkpoint of a bounded source carries the static default - no trigger rides it to carry a stamp.

Every stamped decision increments clink_ckpt_mode_total{mode="aligned"|"unaligned"}, and each policy flip increments clink_ckpt_adaptive_switch_total - a healthy adaptive job switches rarely, and a counter climbing every few checkpoints means the thresholds sit on top of the workload's noise. Checkpoint correctness is mode-independent (each checkpoint runs entirely under the mode pinned by its first barrier delivery), so the adaptive policy chooses between two individually-proven protocols rather than creating a third.

Exactly-once: two-phase-commit sinks¶

Snapshot-on-barrier makes operator state recoverable; getting end-to-end exactly-once also requires that output is only published when the checkpoint that produced it is globally durable. That is the two-phase-commit (2PC) sink protocol. FileSink2PC<T> (file_2pc_sink.hpp) is the canonical implementation:

on_data(batch) appends records to an in-progress staging file (staging/sub<N>-pending.tmp).
on_barrier(b) (the pre-commit / phase 1) closes the in-progress file, atomically renames it to a checkpoint-tagged staging path (staging/sub<N>-<id>.dat), and stores that path in state under _2pc_pending_<sub>_<id>. The bytes are now pre-committed but not yet visible as output. The sink runner snapshots its state slice and acks.
on_commit(id) (phase 2) atomically renames the staging file into committed/, then erases the state key. This is the only step that makes output visible.
on_abort(id) deletes the staging file and erases the state key (idempotent).

ParquetSink2PC<T> (parquet_2pc_sink.hpp) follows the same staging-then-rename protocol but each pre-committed transaction is a complete, self-describing Parquet file written via write_fsync_rename, so a committed file is durable across an OS/power crash and readable by any standard Parquet consumer. The plain (non-2PC) ParquetSink and S3 sinks do not implement this contract; see ../connectors/README.md for which connectors are 2PC.

The commit phase and commit groups (cluster path)¶

Phase 2 is driven by the Coordinator (src/cluster/coordinator.cpp):

As subtasks ack (handle_subtask_checkpointed_), the coordinator erases each from the checkpoint's pending set.
When the pending set empties, the coordinator writes a durable COMPLETED-<id> marker to the checkpoint directory, advances latest_completed_checkpoint_id once that write has returned, then broadcasts CommitCheckpoint to every Worker hosting tasks for the job. The marker is written before the broadcast, so a crash mid-broadcast still lets recovery find COMPLETED-N and commit on restart. The in-memory restore point advances with the marker, not before it: the exactly-once specification found that a restart deciding its restore point between the two redeployed from a checkpoint the next coordinator could not see, and the recoverable sinks, which had already re-committed its handles at open, published its interval twice once the successor restored lower. One exception: a checkpoint that completes while the job is draining for a restart keeps its marker but is not broadcast - a broadcast into a half-torn-down job commits some sinks and not others, and a completed checkpoint with partial external commits is unrepairable at restore granularity (the restore must either replay committed slices as duplicates or skip uncommitted ones as loss). Left completed-but-unconfirmed, the restart's held in-doubt resolution finalises every prepared transaction as one decision; the Kafka 2PC sink's teardown cooperates by preserving barrier-sealed prepared transactions rather than aborting them.
Each Worker dispatches CommitCheckpoint to the commit callbacks its sinks registered, which calls Sink::on_commit(id). Non-2PC sinks ignore it. For a job containing a sink whose commit cannot be re-executed after a crash (ConnectorCapabilities::commit_recoverable == false - the Kafka transactional sink), the worker additionally sends CommitConfirmed per sink subtask whose callbacks ran without throwing; the coordinator seeds a per-checkpoint confirmation set from the tracked tasks at broadcast time, writes a CONFIRMED-<id> marker beside COMPLETED-<id> once it drains, and restores for such jobs select the newest confirmed checkpoint (worker-loss restarts and HA recovery both; the manifest carries the flag). Before that selection, both paths run in-doubt resolution (clink/cluster/in_doubt_resolution.hpp): the resume handles the sink staged inside each completed-but-unconfirmed checkpoint are dispatched to their registered resolver (Kafka: a wire-level EndTxn(commit) with the dead producer's identity), and a full success writes CONFIRMED-<id> and advances the restore point past the interval instead of replaying it. A handle is read ONLY from its own subtask's snapshot (union operator-state restore replicates stale copies of every sink's handle into every subtask, and a stale old-incarnation copy once aborted the whole walk on a phantom fence); a fenced EndTxn retry is further disambiguated with DescribeTransactions, because the staged epoch lags the broker's per-commit epoch bumps. A handle whose subtask left a commit receipt on disk (<checkpoint_dir>/_jobs/<job_id>/receipts/sub<K>-<N>, written by the sink the instant the broker acknowledged its commit) is taken as COMMITTED with no wire call at all - the wire can be fenced, timed out, or answering for a transaction the broker no longer remembers, none of which retracts a commit that executed; only the handles without receipts go to their resolver. The walk probes every handle of a checkpoint before deciding - a final refusal must not leave later handles unproven - and materialises the receipt for each commit it proves over the wire (or executes itself), writing sub<K>-<N> from the watermark horizon the handle stages; a proven-but-unreceipted commit is otherwise invisible to replay suppression, and a mixed verdict then replays it as duplicates (qual01-20260819f: one subtask's whole pane, twice). When the walk still falls back with some receipts on the books - written by the sinks or materialised by the walk - the restored sinks arm replay suppression from those receipts so the replayed interval is not published twice (the connector page's commit-receipts section has the full contract). Workers prune receipts in the same retention sweep that purges superseded checkpoints, never below the retain floor. In-incarnation restarts HOLD the redeploy while the resolver answers off-thread - nothing deploys in that window, so nothing can fence the orphan - and any refusal or failure releases the restart on the bounded contract unchanged. The hold is deadline-guarded in two stages: a walk that outruns the soft deadline is CANCELLED - it checks the token before every wire probe and every store effect (and between reading a probe's answer and acting on it, because an EndTxn probe executes a commit and the store writes steer every later recovery), returns its progress, and the restart proceeds on the bounded contract; only a walk that ignores its cancel through the hard grace is treated as hung and fails the job. A slow walk is not a hung walk: its own wire budget under an outage legitimately runs minutes, and the one-stage timeout it replaces failed the job while the abandoned walk kept committing transactions and writing CONFIRMED markers behind it. A walk that ends UNRESOLVED - transport retries exhausted, or cancelled - persists each unsettled handle as a sub<K>-<N>.unresolved marker beside the receipts (its mandated final act, the one store write a cancel does not suppress); the owning sink consumes the marker at its next open() BEFORE opening its producer, resolving the orphan with a read-only DescribeTransactions while the never-fenced identity still lets the broker name its fate - committed gets its receipt written there, undecided is left for the init's abort and the legitimate replay - and REFUSES to open at all while no broker can answer, because a blind fence erases the distinction permanently (the qual01 rig-night duplicate). Markers are retired by whichever side resolves them and never survive a final broker refusal. Every early stop of the walk - a refusal, exhausted transport, a cancel, an unreadable marker or snapshot - also leaves a marker for each unreceipted handle staged in every completed checkpoint above the stop: the walk used to return at the first refused checkpoint with the checkpoints above it never looked at, and a commit that had executed there without its receipt (an ack-window kill) was fenced blind by the redeploy and replayed as duplicates, with the refused checkpoint standing as a wall every later walk stopped at until a higher CONFIRMED marker landed. That one was found by the exactly-once specification, not by a rig. A restart whose deploy then finds too few slots (its lost worker has not re-registered yet) WAITS under a capacity deadline rather than failing - worker registration re-fires held restarts - with the no-slot failure reserved for capacity that never returns. CommitCheckpoint also carries a retention floor so no worker purges the confirmed restore target while newer checkpoints sit completed but unconfirmed. Jobs without such a sink never track, never confirm, and behave exactly as before. The dispatch runs on the worker's reader thread and is serialised against subtask teardown by a per-task CommitDispatchGate: each wrapped callback enters the gate for the dispatch's duration, and the runner retires the gate - blocking until in-flight dispatch drains - before its LocalExecutor (owner of the RuntimeContext and StateBackend the sink's finalise path touches) is destroyed. A dispatch arriving after retirement is refused; the persisted handle is then re-committed idempotently at the next restore. The worker records the job's committed high-water mark (what wait_final_committed polls for the EOS final checkpoint) after this dispatch, so a source runner that unblocks on it cannot exit and retire the sink's gate ahead of the commit actually running on that worker.

A FAILED checkpoint (any subtask could not snapshot) initiates the same whole-job restart a subtask error does. The failure's abort broadcast discards every sink's barrier-sealed staged transaction - one checkpoint interval of output - and the runner survives its own capture failure, so without the rewind the job would sail on minus that interval: silent loss from a transient snapshot error. The restart replays from the last completed checkpoint and re-produces the aborted interval. The restart is guarded exactly like the subtask-error path (budget, cancel, completion), and its log line reads checkpoint failure -> whole-job restart. The failed id also becomes the job's rewind floor until the restart redeploys: the trigger loop does not wait for one checkpoint's acks before issuing the next, so a checkpoint above the failed one is routinely still collecting acks, and one that finishes collecting them during the drain is discarded like the failed one (no marker, an abort for its staged transactions) rather than completed. Left to complete, it became a restore point past the aborted interval - in-doubt resolution committed its transactions and confirmed it, and the job restored from it minus one interval of output. Found by the exactly-once specification; pinned by CheckpointCompletion.ACheckpointAboveAFailedOneIsDiscardedDuringTheRewind.

A recovered or resumed job numbers its new checkpoints above every checkpoint id with a durable record in its directory - COMPLETED-/CONFIRMED- markers AND every snapshot file any incarnation left behind - not merely above its restore point. Markers alone are not enough: a seconds-lived incarnation (the middle attempts of a restart storm) dies holding snapshot files for checkpoints that never completed, no marker names them, and a successor numbering above markers alone would reuse their ids - its files then interleave with the dead incarnation's, and a later restore can assemble one checkpoint id from two vintages (one shakedown re-published ten windows exactly that way: window state of one vintage, source offsets of another, under one nominal id). The two can differ: a coordinator that dies between the marker write and the commit broadcast leaves its newest checkpoint completed-but-unconfirmed, and when in-doubt resolution reports that checkpoint's transaction as not externally committed, the job restores from the older confirmed one. Numbering from the restore point would then reuse the dead incarnation's id, overwriting a marker and snapshot files that recovery may still need to read consistently; the QUAL-01 shakedowns produced exactly that shape twice in one minute. Ids are never reused once anything durable names them, and the recovery log states the gap (numbers new checkpoints from ...) whenever it opens one.

Cross-sink agreement comes from that sequence, not from any per-sink option. Because commit is one job-wide broadcast issued only after every subtask acked, and because one failed ack aborts the checkpoint for the whole job, a job's transactional sinks are told to commit together or not at all. Being told to commit together is not the same as committing atomically: one sink can complete its commit and another's worker die before completing its own, and that split is repaired on restart rather than prevented - FileSink2PC::open() runs recovery and commits any leftover staging file whose tracked checkpoint_id corresponds to a COMPLETED-N marker. Idempotence of on_commit/on_abort is required for exactly that reason. A job that never restarts keeps the split.

A commit group (set_commit_group(name) on a sink, or commit_group in a sink's op params) is narrower than its name suggests, and does not add atomicity to the above. There is no group-scoped commit broadcast; CheckpointGroupState::pending is maintained and never read. Membership changes exactly one thing: a failed ack aborts the group immediately, whereas the checkpoint-level abort waits until every subtask has answered - and since nothing times a pending checkpoint out, a peer that never answers would otherwise leave staged sink transactions staged. Verified by running a two-sink job with and without a group and comparing what each sink published per checkpoint (tests/integration/test_commit_group_atomicity.cpp); the results are identical. The delivery-guarantee analyser therefore warns about multiple transactional sinks regardless of grouping, and says outright that setting a commit_group will not change it.

Fsync durability¶

A checkpoint is only honestly durable-before-ack if its bytes are on stable storage, not merely in the kernel page cache. write_fsync_rename (durable_file_write.hpp) provides that: it writes the temp file and fsyncs it through the same descriptor that wrote it (so a writeback error is not missed), atomically renames to the final path, then fsyncs the parent directory so the rename itself survives a crash. A successful return means both the bytes and the directory entry that names them are durable. The file fsync is strict (a failure throws, so the caller can ack the checkpoint as failed); the directory fsync is best-effort (its worst case is recovery falling back to the previous retained checkpoint, never corruption). This runs off the operator thread on the snapshot worker for async-capable backends, so the fsync cost is off the record-processing hot path.

Durability is on by default and disabled with CLINK_STATE_FSYNC=0 (or false), which falls back to flush-then-rename for fsync-hostile CI or pure-throughput benchmarks where the durability contract is not under test. The variable is read once per snapshot, not per record, so the toggle is dynamic.

Source-offset replay¶

The source side of exactly-once is snapshot_offset / restore_offset on Source (operator_base.hpp). snapshot_offset(backend, op_id, ckpt_id) persists whatever read position the source needs to resume from; it runs inside the barrier drain so the offset is captured atomically with the barrier. restore_offset(backend, op_id) is called by the source runner before open() on startup, so the source resumes from the recovered position. The defaults are no-ops: a source that does not override these replays from the start on restart (at-least-once at the source boundary). Sources that do override participate in pipeline-wide exactly-once. At end-of-stream a bounded source can request one final coordinator-coordinated checkpoint that durably commits the tail (records since the last completed periodic checkpoint) and blocks until it commits before the runner returns, so the job cannot be reported complete with an uncommitted tail; a crash in that window leaves the source unfinished and is recovered by restart-and-replay.

Source also exposes notify_checkpoint_complete(ckpt_id) / notify_checkpoint_aborted(ckpt_id) - the source-side mirror of a 2PC sink's on_commit / on_abort. Where snapshot_offset is a replayable offset (Kafka, Parquet, File), a crash simply resumes from it and these hooks are unused. But a source whose resume is an irreversible broker consume - an AMQP / JetStream / Pulsar ack, a cursor advance - cannot ack at the barrier: if the checkpoint later aborts, the broker will not redeliver an already-acked message, so that message is lost. Such a source records the position at snapshot_offset and defers the actual ack until notify_checkpoint_complete confirms the capturing checkpoint is globally durable; notify_checkpoint_aborted releases the pinned messages for redelivery. The cluster source runner (plugin_impl.hpp) drives both from the same CommitCheckpoint / AbortCheckpoint dispatch the 2PC sinks use (register_commit_callbacks / register_abort_callbacks), weak-capturing the source so a late notification during teardown is a safe no-op. The RabbitMQ, NATS and Pulsar sources adopt this; each keeps every broker call on the produce() thread (the notification only hands work across), so the non-thread-safe client connection is untouched from the dispatch thread. Both deployment shapes wire the notifications: the default (non-fused) subtask path via register_commit_callbacks in plugin_impl.hpp, and the par-1 chain-fusion path (CLINK_PLAN_FUSE_PAR1=1) via the TypeOps::fused_source_commit_hooks / fused_sink_commit_hooks seams - the fused-chain dispatch in worker.cpp recovers the typed source (and, symmetrically, a fused 2PC sink's on_commit/on_abort) and registers the callbacks into the same per-subtask committer/aborter buckets handle_commit_checkpoint_ / handle_abort_checkpoint_ dispatch, then drops them when the runner exits. So a fused chain gets per-checkpoint commit at both ends, not just the terminal on_commit the dag path fires (which still runs, and is safe because on_commit/on_abort are idempotent).

Key types and APIs¶

Type / function	Responsibility
`CheckpointBarrier`	In-band marker: `id()`, `is_terminal()`, `mode()`
`CheckpointCoordinator::trigger()` / `trigger(mode)`	Allocate id, record pending acks, return a stamped barrier
`CheckpointCoordinator::register_operator` / `acknowledge` / `abort`	Participant registration and ack/abort lifecycle
`CheckpointCoordinator::start_periodic_trigger` / `set_source_injectors`	Periodic in-process trigger (local path)
`CheckpointCoordinator::set_mode_resolver`	Per-checkpoint adaptive mode seam (e.g. backpressure-driven)
`CheckpointCoordinator::enable_adaptive_mode`	Installs an `AdaptiveModePolicy` over a caller-supplied pressure function, sampled once per trigger
`checkpoint::AdaptiveModePolicy`	The decision discipline: threshold, consecutive-window hysteresis (up and down), bounded history, switch counter
`Dag::channel_pressure_fn`	Normalised max channel occupancy across the DAG's runner input channels - the in-process pressure signal
`MultiInputAlignment::on_barrier`	Chandy-Lamport alignment; returns `forward` / `unaligned_first`
`MultiInputAlignment::pending_inputs_for`	Inputs not yet delivering a barrier (which channels to drain on unaligned capture)
`SnapshotWorker`	Off-thread `persist` + ack; FIFO capacity-1 queue with backpressure
`StateBackend::capture` / `persist` / `snapshot`	Detached blob capture; off-thread durable write; synchronous snapshot
`state::detail::write_fsync_rename`	Crash-safe durable write (file + dir fsync, atomic rename)
`Source::inject_pending_barrier` / `take_pending_barrier`	Barrier handoff into the source's drain loop
`Source::snapshot_offset` / `restore_offset`	Source-offset persistence and replay
`Source::notify_checkpoint_complete` / `notify_checkpoint_aborted`	Source-side commit/abort hooks for irreversible-consume sources (messaging acks)
`Sink::on_barrier` / `on_commit` / `on_abort`	2PC pre-commit / commit / rollback hooks
`Sink::set_commit_group` / `commit_group`	Atomic-commit group membership

Configuration and knobs¶

Knob	Where	Default	Effect
`CheckpointCoordinatorConfig::interval`	`checkpoint_coordinator.hpp`	`0` (disabled)	Periodic trigger cadence (in-process path)
`CheckpointCoordinatorConfig::timeout`	`checkpoint_coordinator.hpp`	`60000` ms	Checkpoint timeout field
`CheckpointCoordinatorConfig::default_mode`	`checkpoint_coordinator.hpp`	`Aligned`	Mode stamped on issued barriers absent an override
`JobConfig::unaligned_checkpoints`	`job_config.hpp`	`false`	Job-global alignment policy; sources stamp barriers from it
`JobConfig::adaptive_barrier_mode`	`job_config.hpp`	`false`	Sources forward the injected barrier's per-trigger stamp instead of re-stamping the static mode
`JobConfig::barrier_mode_overrides_by_operator`	`job_config.hpp`	empty	Per-operator mode override re-stamped on passing barriers
`CheckpointAlignment` (cluster job submit)	`protocol.hpp`	`Aligned`	`Aligned` / `Unaligned` / `Adaptive`; adaptive puts the mode decision on the coordinator's trigger sweep
`interval_ms` (cluster job submit)	`protocol.hpp` / `coordinator.cpp`	`0` (disabled)	Cluster periodic-checkpoint cadence; `> 0` with a checkpoint dir enables it
`CLINK_STATE_FSYNC` (env)	`durable_file_write.hpp`	on (unset)	Set to `0`/`false` to fall back to flush+rename

Guarantees and caveats¶

What is guaranteed. Operator state at a barrier is captured into a globally consistent cut; for backends that support it the durable write happens off-thread and the ack is never sent before persist returns (ack-after-durable). With fsync on, snapshot bytes and their directory entry are on stable storage before the ack. The coordinator/coordinator completes a checkpoint only when every registered subtask has acked.
Exactly-once is sink-contract-gated. End-to-end exactly-once applies only to sinks that implement the on_barrier / on_commit contract. FileSink2PC is the canonical example; ParquetSink2PC adds fsync durability. The plain ParquetSink and S3 sinks are not 2PC (see the capability catalogue and ../connectors/README.md).
Source replay is opt-in. Sources that do not override snapshot_offset / restore_offset replay from the start on restart (at-least-once at the source boundary, not exactly-once).
Unaligned checkpoints are partial. Unaligned mode is honoured at multi-input operators that implement in-flight capture; an operator that cannot capture (gated by can_unalign) degrades an unaligned barrier to aligned. Async single-input operators force-align (drain to quiescence) regardless of the barrier mode, which is lossless for a single input. Terminal barriers are always treated as aligned (no more records are coming).
Adaptive mode is a seam, not wired end-to-end. set_mode_resolver and the per-operator override map exist, but plumbing a live backpressure signal up to the resolver is left to the hosting runtime; only JobConfig-driven static selection is wired.
The async snapshot worker is backend-gated. It runs only for backends reporting supports_async_persist() (FileBacked, disk-backed changelog). InMemory, RAM-only changelog and RocksDB stay on the synchronous on-thread snapshot path.
fsync best-effort on the directory. The file fsync is strict; the directory fsync is best-effort and may be a no-op on filesystems that reject fsync on a directory fd. Setting CLINK_STATE_FSYNC=0 removes the durability guarantee entirely (process-crash survival only).

fault-tolerance-and-rescale.md: how completed checkpoints drive restart-from-checkpoint, rescale (key-group repartitioning) and schema evolution.
state-and-backends.md: the StateBackend capture / persist / snapshot / restore contract that checkpointing snapshots through.
async-state-execution.md: AsyncExecutionController::drain_for_barrier and force-alignment of async operators at a barrier.
time-and-windowing.md: watermarks, which share the in-band channel and the same alignment machinery as barriers.
operator-model.md and task-lifecycle.md: the operator runners in dag.hpp that drive barrier flow and snapshot-on-barrier.
distributed-runtime.md: the Coordinator ack tracking, COMPLETED-N markers and CommitCheckpoint/AbortCheckpoint broadcast on the cluster path.
exactly-once-specification.md: the TLA+ model of this protocol, the invariants TLC checks in CI, and the protocol trace the engine records at these decision points when CLINK_PROTOCOL_TRACE_DIR is set, validated against the model.
../connectors/README.md: which source/sink connectors implement the 2PC and source-offset contracts.