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 logic and the restore of in-flight slots that checkpoints from earlier releases carry
`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. A sink whose task is already stopping aborts the terminal transaction instead: no checkpoint covers it, so a cancelled job's rerun would publish the tail again. A multi-input operator passes a terminal barrier on as terminal once every input has delivered it or finished, so a sink behind a fan-in commits the tail as one behind a single input does; an input that closed by cancellation or for a rescale handoff turns it into an ordinary barrier, because teardown is not end of input. (An aligned fan-in used to pass every terminal barrier on as an ordinary one, so a 2PC sink behind it waited 30 s for a commit nothing would send and then failed; an unaligned one passed on the first delivery as terminal, so the sink committed before the other inputs' tails had arrived.)

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.

In a cluster the injector reaches a source through its worker, which hands each TriggerCheckpoint to the injectors its source subtasks have registered. A source registers them only once its runner is built, after the PeerUpdate, while the coordinator triggers periodic checkpoints from the moment that PeerUpdate goes out and a savepoint from the moment the job is deployed, so the worker holds every trigger for each source subtask that has not registered yet (Worker::held_source_triggers_, created when the worker accepts the Deploy, before the subtask's thread has parsed its chain) and injects them, oldest first and with their barrier mode, when it registers. Every source of the job then emits the same barriers, however late it comes up. The worker used to queue only the newest trigger, and only while no source of the job had registered, replaying it into whichever source registered first: a sibling that registered later never emitted that checkpoint's barrier, and every operator that merged the two waited for it for good (see A barrier an input skipped). The broker-outage recovery gate stalled that way after a redeploy. tests/test_source_trigger_handoff.cpp holds one source's registration with the worker.before_source_injectors fault point and requires the checkpoints triggered meanwhile to complete, and holds every subtask before it parses its chain with worker.before_chain_parse while a savepoint is taken, which must complete too.

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 when a source registers and the triggers held for it are injected. 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 (a union, a join, a co-process or broadcast operator, a keyed exchange merging several upstream subtasks, a sink fed by several subtasks) 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 }. Every runner aligns every barrier, whatever Mode it is stamped with, so every checkpoint is a consistent cut across all inputs.

Alignment (Chandy-Lamport)¶

flowchart LR
  F["input 0: rows, then barrier N"] --> P["input 0 paused at barrier N"]
  S["input 1: rows still ahead of barrier N"] --> R["processed normally, inside checkpoint N"]
  R --> B["input 1 delivers barrier N"]
  P --> A["every alive input delivered barrier N"]
  B --> A
  A --> O["snapshot, forward barrier N, unpause all inputs"]

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 (marked Aligned unless it was stamped otherwise and the operator captures in-flight rows, which none does), 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. A runner watches for a close even on an input it has paused: an input that delivered barrier N and then closed would otherwise never be seen to close, and a later barrier that the other inputs deliver could not align without it. When the close is itself what completes a pending barrier, on_input_closed returns the release, and every multi-input runner acts on it exactly as on a barrier delivery: it forwards the barrier (and the co-operator snapshots and acknowledges it, a fan-in sink and an N-input operator hand it to on_barrier) before the watermark the close frees. While another input is still open, every runner forwards that watermark straight away through refresh_watermark(), so the survivors' time is not held at the closed input's last value while they sit idle. In-flight records are not persisted, because by the time the barrier forwards every input has reached it.

A barrier an input skipped¶

Barriers reach an input in id order, so an input that delivers barrier N+1 without having delivered N has gone past N for good, and N can never align at this operator. The aligner drops N, unpauses the inputs paused on it, and drops any later delivery of N, as it does any barrier at or below one its own input already delivered; the runner goes on to N+1, which completes, and N cannot. Before this, the inputs that had delivered N stayed paused for it and the input that went past paused on N+1: no input was read again, and the operator and everything downstream of it stopped for good while the job reported itself running. A terminal barrier is never dropped this way, because the close that follows it completes it, and a closed input has gone past nothing. Each dropped barrier counts in clink_op_barrier_skipped_total{op_id="N"} and each runner logs the first; a nonzero count means a source upstream did not emit that checkpoint's barrier. tests/test_skipped_barrier_alignment.cpp drives the aligner and the union, fan-in sink and N-input operator runners through a skipped barrier.

Unaligned and adaptive barriers align at every fan-in¶

A barrier stamped Unaligned, whether by CheckpointAlignment::Unaligned, by adaptive alignment once its policy turns unaligned, or by a per-operator override, is aligned by every multi-input runner exactly as an Aligned one: the input that delivered it pauses until every alive input has delivered that id, and the barrier goes on marked Aligned. No runner captures the rows still in flight on its other inputs, so aligning is what keeps each checkpoint a consistent cut; the cost is the latency of an aligned checkpoint under backpressure, which is what unaligned checkpoints set out to avoid. Each such alignment counts in clink_op_barrier_forced_alignments_total{op_id="N"}, and each runner logs the first one it makes. The count is of the first fan-in on a path: it forwards the barrier marked Aligned, so a fan-in further downstream sees an ordinary aligned barrier and counts nothing.

Forwarding at once is only correct for an operator that captures those rows into the checkpoint, and none of the earlier attempts did. Up to v0.10.0 a union forwarded the first unaligned barrier and the slower inputs' earlier rows behind it, and whatever consumed the union (a sink, a keyed operator, a bridge to the next worker) filed those rows under the next checkpoint while their sources' offsets in this one were already past them, so a restart from it lost them. The co-operator, interval join and broadcast runners drained the waiting input's queued rows into an in-flight state slot and never processed them in the live run, so a join under unaligned checkpoints dropped rows even without a restart. The forward-on-first-delivery path stays in the aligner behind a constructor flag, MultiInputAlignment(n, /*captures_in_flight=*/true), for an operator that does capture: its first delivery forwards at once with unaligned_first = true, and pending_inputs_for names the inputs whose rows it must capture. No runner sets the flag.

Restore still honours the slots those releases wrote. The co-operator (__co_op_left_inflight__ / __co_op_right_inflight__), interval join (__interval_join_left_inflight__ / __interval_join_right_inflight__) and broadcast (__broadcast_main_inflight__ / __broadcast_brod_inflight__) runners read each slot at startup, erase it, and replay its rows through a local pending queue ahead of the live input, so a checkpoint an earlier release took in unaligned mode restores with what it captured, exactly once. What it captured is as the defect left it: a slot can hold rows the slow input sent after its own barrier, which its source replays again, and lacks rows that were still in transit when it was written.

Per-checkpoint mode is decided by the first delivery and pinned for that id. apply_barrier_mode_override lets a per-operator JobConfig override re-stamp a barrier; an Unaligned override aligns like any other unaligned barrier. Async multi-input operators also drain their popped, in-flight async work to quiescence at the barrier (drain_for_barrier), so the captured state reflects every record admitted before it.

Adaptive mode (measured pressure decides per checkpoint)¶

CheckpointAlignment::Adaptive puts the aligned-versus-unaligned stamp on measured pressure instead of a static flag, one checkpoint at a time. The stamp is all it decides: every fan-in aligns an unaligned barrier as an aligned one (see above), so an adaptive job takes aligned checkpoints whichever way the policy goes. 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 does not depend on the policy: whichever stamp it chooses, every multi-input operator aligns the barrier, so each checkpoint is a consistent cut across all inputs.

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 (_jobs/<job id>/COMPLETED-<id>, recording the checkpoint's generation, the subtasks that acked it, and state=, the job's state root, which is where a restore reads it from; see state and backends), 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 write is a temp file (COMPLETED-N.tmp.<pid>.<n>), fsynced and then renamed into place, and the rename is the decision: a coordinator killed between the fsync and the rename leaves the temp file and no marker, that checkpoint never completed, and a reader of the marker directory counts only whole marker names. 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. The write runs outside the coordinator's lock, so the advance takes two gates: a marker whose job redeployed while the write was out stays on disk and moves nothing (the new run restored from a point the checkpoint is not part of), and no later restore chooses it: the redeploy put the job's RESTORED record (_jobs/<job id>/RESTORED: the checkpoint it restored from and the first id its run numbers) before its frames left, and a takeover or a resume passes over a marker between the two, because the sinks that reopened have rolled that checkpoint's transactions back (see distributed runtime). One landing while the restart is held for in-doubt resolution leaves the walk's range where the hold fixed it. 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 (advancing the in-memory latest_confirmed_checkpoint_id only after that write returns, as the completed restore point does, so a failed write leaves it where it was; a restart's redeploy discards the confirmation sets still open, as the specification's Redeploy forgets every broadcast checkpoint, so a CommitConfirmed for the run before it drains nothing, and one whose marker write was already out when the job redeployed leaves its marker but does not move the new run's restore point), 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. Receipts and markers live under the job id, which restarts at 1 in every coordinator process, so a run in a reused directory can find an earlier run's: every Deploy carries the run's checkpoint-id floor (DeployMsg::run_checkpoint_floor), fixed when the run is first submitted and kept by its restarts and an HA takeover (the job manifest records it). A fresh start's floor sits above every id an earlier run of the job left on record or as a marker; a submit that restores from checkpoint R continues the run that took R, and its floor is R, so that run's receipts above R suppress the replay of output already published. A sink settles markers and arms from receipts above the larger of the restore point and the floor, so a restart that restores from nothing (no checkpoint of the run confirmed yet) still reads this run's, and a fresh start reads none of an earlier run's. A Deploy without the floor, from an older coordinator or a takeover of a job persisted before it, keeps the older rule: at restore point 0 the sink neither settles nor arms. 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 settles the marker at its next open() BEFORE opening its producer, while the never-fenced identity still lets the broker name its fate: a commit DescribeTransactions names gets its receipt written there, and an undecided orphan is ended with EndTxn(abort) under the identity the describe reported, so the replay is its interval's only copy (an abort refused because the old incarnation's commit got there first is followed by a describe that proves the commit). It 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 (Coordinator::Config::restart_capacity_timeout) rather than failing - worker registration re-fires held restarts - with the no-slot failure reserved for capacity that never returns. The walk runs once per restart: a restart waiting for capacity after its walk redeploys from the restore point the walk left, and neither walks again nor moves its completed restore point for a COMPLETED marker landing meanwhile. 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 commit-dispatch thread and is serialised against the sink's teardown by a per-task CommitDispatchGate: each wrapped callback enters the gate for the dispatch's duration, and the sink's runner retires the gate - blocking until in-flight dispatch drains - as the teardown begins, after the end-of-stream wait for staged commits and before close(), and on an exit by exception (Sink::fence_commit_dispatch, installed by the code that registers the sink's callbacks). The runner retires it again before its LocalExecutor (owner of the RuntimeContext and StateBackend the sink's finalise path touches) is destroyed. So a commit never runs under close(): one already inside its external call is waited for, one that has not started it declines (Sink::teardown_begun), so a store that is down cannot hold a restart drain through a commit it was never asked for, and a dispatch arriving after the teardown began is refused, logged and not confirmed; the persisted handle is then re-committed idempotently at the next restore. Retiring the gate only at the executor's destruction left a window in which a commit reached a sink already closed and detached, finalised nothing, and was confirmed as if it had committed; a committing sink reached after its teardown now throws as well. 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. A job already restarting, completing or cancelling is left to that path, and the log line reads checkpoint failure -> whole-job restart. A job with no restart budget left cannot rewind, and it fails instead, naming the checkpoint and the cause, as the subtask-error path does when out of budget; the failed id still becomes its rewind floor, so nothing completes above it and its recovery point stays below the gap, and an operator restore from that point is the rewind it could not make. It used to carry on: the checkpoint above the failed one completed and committed, every later one built on it, and the interval's output was lost with every gate green. Pinned by CheckpointCompletion.AFailedCheckpointWithNoRestartBudgetFailsTheJobInsteadOfSailingOn and modelled by MC_RecoverableNoBudget, with the sail_on_without_budget mutant refuted. 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, under the job's own state root and at the top of the checkpoint directory, where state lived before it was namespaced by job - 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.

Nor do the snapshot files show every capture. A checkpoint whose barrier has reached a worker leaves no file until its capture lands, and a worker that outlives the dead coordinator can still land it after the new leader has read the directory, at the path the new run's capture of a reused id writes: a later restore then reads one checkpoint from two vintages. CI's trace validation caught a takeover reusing such an id. So every checkpoint id is claimed in _jobs/<job_id>/TRIGGERED (the id in decimal, then the claimant) before it is allocated, and a takeover numbers above that record too. A claim lands only over a record holding less, through the coordination store's compare-and-set, so the record only rises and each id is claimed once across every coordinator of the job. That matters because a superseded coordinator whose trigger loop runs on goes on claiming and triggering under its old epoch. A worker that has re-registered with the new leader refuses its barriers, but one that has not yet re-registered still accepts them, and the new leader recovers its jobs once registrations settle, not once every worker has bound its epoch. A coordinator whose claim is refused therefore numbers above whatever the record holds and allocates nothing that round, so the leader never uses an id the superseded coordinator may have delivered; the renumber is logged, and recorded in the protocol trace as Renumber. A job taken over before it completed a checkpoint of its own numbers above the record too; it used to number from 1 again. A fresh job in a directory an earlier run of its id used still numbers from 1, through ids no coordinator can claim again once the record has passed them.

Each job has one claimer thread, started by its first claim and parked between claims until the job ends, which puts the next id on record as soon as the previous one is allocated. The claim has normally landed by the time the id falls due, so the trigger loop never waits on the store, and a slow or failing record holds only its own job's checkpoints. A claim that keeps failing is retried after a delay that doubles from 100 ms up to 2 s, and is reported once until the record can be written again; meanwhile the job's periodic checkpoints wait, since nothing is allocated before it is on record. The store write logs nothing itself (fenced_put_quietly), and a claim the record already holds is an answer, not an error. The record names the claimant whose claim put the id there (claimant= on its second line), unique to one coordinator's deploy of the job, so a claim whose write landed and whose answer was lost (on S3 an ordinary retry of the conditional write is refused, and a network filesystem can report a failure for a rename that landed) is taken as claimed when its retry finds it, where it used to read as another coordinator's and cost the job a renumber. A savepoint and a hot cutover normally find their id on record already, and claim it themselves before allocating it when they do not. A savepoint whose id cannot be claimed fails with nothing sent. A hot cutover whose id cannot be claimed is not begun, and the replan takes the request; the cutover's barrier is checked against the record again as it is sent, because the arm acks that release it do not name the checkpoint. A source's end-of-input request for its final id arrives on its worker's control connection and is handled on that connection's dispatch thread, which handles the worker's frames one at a time, so it is never claimed for there: a claim would hold every later frame from that worker, its checkpoint acks included, behind the store write. A request whose id is not on record yet is held instead, the job's claimer is asked for the id, and the trigger loop answers the request once the claim lands. While the id cannot be recorded the request goes unanswered, and the source fails its subtask when its bounded wait runs out, so the restart replays the tail under a checkpoint. The COMPLETED- and CONFIRMED- markers are written on the dispatch thread of the worker whose ack or confirmation completes the step, so a slow store delays that worker's later frames; it no longer holds the worker's heartbeats, which the connection's reader answers, and a write slower than the heartbeat timeout no longer gets a live worker declared lost (see distributed-runtime.md).

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. Only a job that takes no checkpoints may fall back to the local terminal commit: the coordinator's reply says why it declined (FinalCheckpointDecline), and a decline because the job is cancelling or complete stops the source's task on the spot, while a request that cannot be sent or gets no answer within 30 s fails the subtask so the restart replays the tail under a checkpoint. Every decline used to fall back to the local commit. A cancel marks the job cancelling before it broadcasts CancelJob, and a source that reached the end of its input between the two committed its tail with no checkpoint behind it; the cancelled run left no FINISHED marker, so a rerun resumed from the last completed checkpoint and published the tail again (EmbeddedEngine.ACancelThatRacesEndOfInputDoesNotPublishATailARerunRepublishes, which holds that window with the coordinator.before_final_checkpoint_request and coordinator.before_cancel_broadcast fault points).

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, or the emitted count a GeneratorSource regenerates up to), 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. The PostgreSQL CDC source does too, for a different reason: its position is replayable, but confirming a position to the replication slot lets the server drop the WAL before it, and PostgreSQL starts a stream asked for an earlier position at the slot's own. It confirms the slot only to the position of a completed checkpoint, from the produce() thread, so every completed checkpoint stays restorable (PostgreSQL connector). 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 of every barrier; returns `forward`, and `unaligned_first` only on an aligner built with `captures_in_flight`
`MultiInputAlignment::pending_inputs_for`	Inputs not yet delivering a barrier (which channels an in-flight capture would drain; no runner captures)
`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, the PostgreSQL CDC slot confirmation)
`Sink::on_barrier` / `on_commit` / `on_abort`	2PC pre-commit / commit / rollback hooks
`Sink::set_commit_group` / `commit_group`	Commit-group membership: a failed ack aborts the group at once; no cross-sink atomicity (see above)

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 barrier stamp; sources stamp barriers from it. Every fan-in aligns an `Unaligned` barrier, so checkpoints are aligned cuts either way
`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 stamp decision on the coordinator's trigger sweep. All three are accepted, and unaligned and adaptive align at every fan-in
`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).
A sink that must own its chain's checkpoint stands alone on it. Two kinds of sink hold their guarantee only while the checkpoint's acknowledgement follows their own barrier hook: an at-least-once sink that writes out at the barrier everything it received before it (Sink::gates_checkpoint_ack()), and a two-phase-commit sink that prepares there and stages its handle into the snapshot (Sink::stages_state_at_barrier(), which every CommittingSink declares). A sink owns the checkpoint only while it is the only sink on its chain. A second sink hands ownership back to the upstream operator, which acks at its own barrier without waiting for either sink, so a checkpoint could complete with the interval neither written nor prepared, and a restore from it would never replay those records. Dag::add_sink refuses the combination in either order, with or without an operator in front, naming the sink and the ways to give the other sink a subtask of its own (tests/test_sink_chain_ownership.cpp, tests/test_cancel_teardown.cpp). Every connector sink that writes outside the job declares one of the two; the in-process sinks (collect, print, blackhole, the network bridge, whose downstream subtask acknowledges after the records ahead of the barrier) declare neither. scripts/check-sink-ownership.py holds every sink class in the tree to a listing in scripts/sink-ownership.txt, owning or free with the reason, in CI and the pre-commit hook. A job graph (SQL, a cluster, the embedded engine) plans every sink as a subtask of its own, and the planner's opt-in par-1 fusion leaves a chain's sink unfused when the chain also sends a side output to another subtask, so only the Dag API meets the refusal. A Dag that will run where no checkpoint completes, under a LocalExecutor with no on_checkpoint_ack, can declare so first with Dag::run_without_checkpoint_completion(): its only barrier is the end-of-input one, which every sink commits itself, so such sinks may then share a chain, and LocalExecutor refuses to run the Dag with an acknowledgement callback. LocalSubmitter, which builds a fluent pipeline as one Dag, declares it for a run without a callback. impls/clickhouse/tests/test_clickhouse_sink_chain.cpp holds the ClickHouse text sink's barrier INSERT and shows the acknowledgement waiting for it.
A two-phase-commit sink needs a state backend. A CommittingSink persists each prepared handle in operator state and publishes only what a completed checkpoint covers. A runtime without a state backend has neither, and its sources emit no end-of-input checkpoint, so CommittingSink::open() refuses it, naming the sink, rather than letting the job finish cleanly with every record staged and none published. A cluster and the embedded engine always give a task a backend (in memory when no checkpoint directory is set); a LocalExecutor run needs JobConfig::state_backend. A sink driven directly, with no runtime attached, is unaffected.
Parallel stages take no part in a checkpoint. The Dag's parallel stages (add_parallel_operator, add_parallel_sink and fork_parallel, which LocalSubmitter builds for an operator whose parallelism is above 1) align each barrier across a subtask's inputs, but their operator and sink subtasks snapshot nothing and acknowledge nothing, and a parallel source records its offsets but leaves the success acknowledgement to the end of its chain. No checkpoint can complete over them, so LocalExecutor refuses a Dag with parallel stages an on_checkpoint_ack callback (Dag::has_parallel_stages). It used to run one, and its checkpoints either never completed or completed under a rule that counted the other runners' acknowledgements, over state nothing had captured. Without the callback, the barrier that matters is the end-of-input one: each sink subtask commits its final interval locally, as a single sink does, or aborts it when the task is being torn down. Before, a two-phase-commit sink in a parallel stage prepared that interval and never published it. Checkpoints complete on a cluster and in the embedded engine, which plan each parallel subtask as a task of its own; nothing in a LocalExecutor run collects acknowledgements or dispatches commits, so no sink there commits an intermediate checkpoint, parallel or not.
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 and adaptive checkpoints align at every fan-in. No multi-input operator captures in-flight rows, so each one aligns a barrier stamped Unaligned exactly as an aligned one, and every checkpoint is a consistent cut across all inputs, with the latency of aligned checkpoints under backpressure. Both modes stay accepted so existing deployments and saved configurations start unchanged. 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 chooses a stamp, not a protocol. The policy's decision is stamped on each triggered barrier and counted in clink_ckpt_mode_total, but since every fan-in aligns an unaligned barrier, an adaptive job takes aligned checkpoints whichever way the policy decides.
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.