The question this answers
How do you capture a consistent global state of a running distributed system without stopping it?
The recorded snapshot is a consistent cut: a global state that could have occurred, in the sense that no recorded event depends on an unrecorded one. It is *not* guaranteed to be a state the system actually passed through at any wall-clock instant, and it is not unique — different runs of the algorithm on the same execution legitimately produce different cuts.
Everything below is bought to hold this sentence. "Strongly consistent" with no scope attached is a slogan, not a guarantee — read what it actually covers, and what it explicitly does not.
A node knows its own local state at the moment it records, and it knows which messages it has received on each incoming channel since recording. It has no idea what any other node’s state is, whether they have recorded yet, or what is currently in flight toward it — and this is exactly why the algorithm makes the *marker* carry the information instead of asking anyone anything.
A node knows its own state and the messages that arrived. Everything else is inference from evidence that was already stale. "B has not replied in five seconds" is knowledge; "B is down" is a decision — and usually the bug.
Why "everyone records at 12:00" fails
The obvious approach is to pick a time and have every node write down its state then. It does not work, and the reason is the foundation of this domain: there is no shared 12:00. Clocks differ by milliseconds at best, and the differences are neither bounded nor knowable — that is There Is No Global Clock and Clock Skew: The Gap You Cannot Measure From Inside, stated as an operational problem instead of a fact.
Suppose the skew were somehow zero. It still fails, because of the messages in flight. Node A sends "transferred £100 to B" at its 11:59:59.9 and B receives it at 12:00:00.1. Both nodes record at exactly 12:00. A has already removed the £100; B has not yet added it. The snapshot shows £100 that has vanished from the system. Nothing is wrong with either node — the money is in the network, and a snapshot of nodes alone cannot see it.
So a correct global snapshot must record two things: the state of every node, and the messages that were in flight across every channel at the moment of the cut. The in-flight messages are the part people forget, and they are the part that makes the recorded state add up.
What a consistent cut actually means
Draw the execution as a spacetime diagram: nodes as horizontal lines, events as points, messages as arrows between lines. A cut is any line drawn across all the node lines, dividing every node’s events into "before" and "after". A cut is consistent if no message arrow crosses it backwards — that is, if there is no message whose *receive* is before the cut while its *send* is after it.
The intuition is causal, not temporal. A consistent cut is a state in which every effect you recorded has its cause recorded too. It never contains a received message that was never sent. It may perfectly well contain a sent message that has not been received — that message is simply in flight, and recording it as such is what makes the totals work.
This is Happens-Before: The Only Ordering You Actually Have used as a definition rather than a technique. The cut does not need to correspond to any real instant; it needs to be *causally closed*. And that is enough for the properties people actually want from a snapshot: it can be used to check a stable invariant (total money conserved, no deadlock, computation terminated), and it can be used as a restart point, because a system restarted from a consistent cut proceeds along a legal execution.
- A cut divides every node’s events into before and after.
- It is consistent when no message is received before the cut and sent after it.
- Effects recorded without their causes are the only thing forbidden; causes without effects are fine and become in-flight messages.
- A consistent cut need not be a state the system was ever in at one wall-clock instant — and does not need to be.
- Consistent cuts are not unique: many valid cuts exist for one execution, and the algorithm finds one, not the one.
Chandy-Lamport: let a marker do the coordinating
The algorithm is small, and its elegance is that no node ever asks another node anything. A special marker message flows along the same channels as ordinary messages, and the rules are:
To start, a node (any node) records its own local state, then immediately sends a marker on every outgoing channel — before sending any further ordinary message.
On receiving a marker on channel c: if this node has not recorded yet, it records its local state now, records channel c’s in-flight set as *empty*, and sends a marker on all of its own outgoing channels. If it *has* already recorded, it stops recording channel c and saves the messages it received on c between its own recording and this marker — that set is exactly what was in flight on c at the cut.
Finish when a node has received a marker on every incoming channel. The global snapshot is the union of all node states and all recorded channel sets, collected afterwards by whoever wants it.
Look at what the marker is doing. Because a node sends markers before any further ordinary messages, a marker acts as a separator on each channel: everything before it was sent pre-cut, everything after it post-cut. The receiver does not need to know when the sender recorded, or what the sender’s clock said — the position of the marker in the channel *is* the cut, communicated by construction. Coordination that would have needed agreement is replaced by ordering along the channels that already exist. That is the trick worth taking away, and it recurs: Watermarks: A Guess About Time, Made Precise Enough to Act On in stream processing are the same move applied to event time.
The assumptions, and what breaks without them
The algorithm as stated assumes FIFO channels — messages on a channel arrive in send order — and that is doing real work. It is what guarantees the marker cleanly separates pre-cut from post-cut messages; on a channel that reorders, an ordinary message sent after the marker can overtake it and be wrongly recorded as in flight. TCP between a fixed pair of endpoints gives you FIFO; a connection that reconnects, a UDP path, or a message broker with several partitions does not.
It also assumes no failures during the snapshot. A node that crashes mid-run never sends its markers, and every downstream node waits forever for a marker on that channel — the snapshot simply does not terminate. Practical implementations add a timeout and abort, which means a snapshot is best-effort under failure and needs to be retryable.
And it assumes the channels are known and finite, so a node can tell when it has heard from all of them. In a system with dynamic membership, "all incoming channels" is itself a piece of agreed state, which drags Cluster Membership: A Belief, Not a Fact into what looked like a purely local algorithm.
Where you meet this in production is stream processing. Flink’s checkpointing is a direct descendant: barriers flow through the dataflow graph exactly as markers do, operators snapshot their state when barriers arrive, and the aligned checkpoint is a consistent cut of the whole job. That is what makes a stream job restartable from a coherent point rather than from an arbitrary mess — and it is why checkpoint alignment stalls show up when one operator is slow, since a barrier can only move at the speed of the path it is travelling.
| Use | Works? | Why |
|---|---|---|
| Check a stable property (terminated, deadlocked, total conserved)protocol | Yes | A stable property true at the cut is still true now — it cannot become false |
| Restart the whole system from the cutprotocol | Yes, with in-flight messages replayed | The cut is causally closed, so continuing from it is a legal execution |
| Answer "what was the state at 14:32:00?"protocol | No | The cut corresponds to no wall-clock instant, and there is no shared instant to correspond to |
| Detect a transient property (a queue was briefly over 1000)assumption | No | A transient condition may have been true in the real execution and absent from every consistent cut |
| Take a snapshot while nodes are failingassumption | Not as stated | A crashed node never emits markers; the run never terminates without a timeout |
Key points
- There is no shared instant, so "everyone record at 12:00" is not merely imprecise — it is undefined.
- A global snapshot must capture node states *and* the messages in flight on every channel.
- A cut is consistent when no recorded event depends on an unrecorded one; it need not correspond to any real moment.
- Chandy-Lamport replaces coordination with ordering: a marker on a channel *is* the cut, so no node asks any other node anything.
- It assumes FIFO channels and no failures during the run; both assumptions bite in practice.
- Stream-processing checkpoint barriers are this algorithm in production, and this is what makes a job restartable.
The chain, answered
Every field here is required, which is why no lesson in this domain can recommend a design without naming what an operator sees when it fails, what survives the partition, what repairs it afterwards, and the simpler thing to consider first.
- • An initiator records its own local state and sends a marker on every outgoing channel before sending any further ordinary message.
- • A node receiving its first marker records its local state immediately and records that channel’s in-flight set as empty.
- • It then sends markers on all of its own outgoing channels, again before any further ordinary message.
- • For every other incoming channel, it records the ordinary messages received between its own recording and the arrival of that channel’s marker — that set is the in-flight content of the channel.
- • A node finishes when it has seen a marker on every incoming channel.
- • The snapshot is assembled afterwards from all local states plus all channel sets; collection is a separate, unhurried step.
- • A channel reorders messages, so a post-cut message overtakes the marker and is recorded as in flight.
- • A node crashes before emitting its markers, and downstream nodes wait for a marker that will never arrive.
- • A slow path delays a marker, so nodes on that path record much later in real time than others — legal, but it stretches the snapshot’s duration.
- • Membership changes mid-run, so "all incoming channels" means different things at different nodes.
- • Recorded state is large, and the act of recording it perturbs the system being observed.
- • Snapshot never completes: an operator sees a snapshot in progress for hours and rising memory on several nodes, because one crashed participant’s marker never arrived and everyone is still buffering channel records.
- • Barrier alignment stall: in a stream job, one slow operator holds up the barrier and every downstream operator buffers input waiting for it. Throughput drops across the whole job and the cause is one task, not the job.
- • Snapshot that does not add up: a totals check on the snapshot fails, and the reason is that channel contents were not recorded — someone captured node states only, and the difference is exactly the in-flight messages.
- • Snapshot-induced latency spike: p99 rises sharply and periodically, at exactly the checkpoint interval, because recording large state stalls processing on each node in turn.
- • False alarm on a transient property: a check run against the snapshot reports "no deadlock" or "no backlog" while the real execution had both — the cut is not obliged to contain a transient condition.
- • Notably little. There is no leader, no agreement round, no clock synchronisation — the ordering already present in the channels carries all the information the algorithm needs.
- • What it does require is that every node participates and that the channel set is known, which is a membership dependency rather than an agreement one.
- • Assembling the result is a separate collection step and can be as leisurely as you like; the snapshot is already consistent by the time collection begins.
- • Practical implementations add a coordinator to initiate, time out and retry — which reintroduces a small amount of coordination for operability, not for correctness.
- • A completed snapshot remains a valid consistent cut regardless of what happens afterwards — it is a statement about the past, and the past does not change.
- • An incomplete snapshot is worthless rather than wrong: partial results must be discarded, not used.
- • A stable property found true in a completed snapshot is still true, which is the entire basis for using snapshots to detect termination or deadlock.
- • The system under observation keeps running throughout; no ordinary message is ever blocked by the algorithm itself.
- • Detect: track snapshot duration and the count of channels still awaiting a marker; a stuck snapshot is visible as a specific channel, not as a vague slowness.
- • Contain: bound the buffering a node will do for channel recording, and abort the snapshot rather than let recording exhaust memory.
- • Recover: abort and retry. Snapshots are idempotent in the sense that a fresh run produces a different but equally valid cut.
- • Reconcile: on restart from a snapshot, replay the recorded in-flight messages into their destination channels — omitting them loses exactly the £100 from the naive example.
- • Verify: run a stable-property check, such as a conservation invariant, over the assembled snapshot. If it does not balance, the channel sets are the first suspect.
- • Snapshot duration, per run, with the identity of the last channel to report — that identity is the diagnosis.
- • Buffered channel-record size per node during a run, which is what turns a stuck snapshot into an out-of-memory event.
- • Processing latency correlated against snapshot start times, to see whether recording is perturbing the system.
- • Snapshot success rate, separated from snapshot latency; a snapshot that always aborts and retries can look healthy on a duration graph.
- • For stream jobs: barrier alignment time per operator, which localises a stall to one task immediately.
- • Restarting a distributed computation from a coherent point — the production use, via stream-processing checkpoints.
- • Detecting stable properties: has the computation terminated, is there a distributed deadlock, does a global invariant still hold.
- • Debugging: a causally consistent global state is enormously more useful than a pile of unrelated per-node dumps.
- • Answering questions about a specific wall-clock instant — the algorithm cannot, and no algorithm can.
- • Detecting transient conditions, which a consistent cut is under no obligation to contain.
- • Systems where a brief global pause is genuinely acceptable: stopping the world for 200 ms is far simpler and gives a stronger result.
- • Very large per-node state, where the cost of recording dominates and a snapshot becomes a periodic performance event.
- • Stop the world briefly. Unfashionable, trivially correct, and the right answer for small systems that can absorb a short pause.
- • Take per-node checkpoints independently and accept an inconsistent set, if you only need each node to restart itself — this is Recovered State Is a Checkpoint Plus the Log After It without the global coordination.
- • Log everything to a single ordered log and derive any past state by replaying a prefix; a total order makes "global state" a query rather than an algorithm.
- • Use vector clocks to reason about causality after the fact instead of capturing a cut at all, when the question is about relationships between events rather than a state to restart from.
Drag the cut until the £100 stops vanishing
| Works? | Why | |
|---|---|---|
| Check a stable property (terminated, deadlocked, total conserved)protocol | Yes | A stable property true at the cut is still true now — it cannot become false |
| Restart the whole system from the cutprotocol | Yes, with in-flight messages replayed | The cut is causally closed, so continuing from it is a legal execution |
| Answer "what was the state at 14:32:00?"protocol | No | The cut corresponds to no wall-clock instant, and there is no shared instant to correspond to |
| Detect a transient property (a queue was briefly over 1000)assumption | No | A transient condition may have been true in the real execution and absent from every consistent cut |
| Take a snapshot while nodes are failingassumption | Not as stated | A crashed node never emits markers; the run never terminates without a timeout |
What people believe, and what is true
Just have every node record at the same timestamp.
There is no shared timestamp, and even with perfect clocks the messages in flight between nodes would still be missing from the result.
A consistent snapshot shows the system as it was at some moment.
It shows a state the system *could* have been in — causally closed and legal. Requiring it to be a real instant would require a global clock that does not exist.
The algorithm pauses the system.
No ordinary message is ever delayed by the protocol itself. Only the cost of recording local state pauses anything, and that is an implementation concern.
Only node states need recording.
Channel contents are half the snapshot. Omit them and conserved quantities do not balance — the classic vanishing-money result.
This is only of theoretical interest.
It is what a stream processor does on every checkpoint. Barriers are markers, aligned checkpoints are consistent cuts, and the alignment stalls you see in production are this algorithm meeting a slow operator.
Go deeper
Only the levels this lesson can honestly fill — a missing level is a claim nobody had.
Overview
You cannot photograph a distributed system at one moment, because there is no one moment. Instead, capture a state where every recorded effect has its recorded cause — plus the messages that were on the wire. That is a consistent cut.
Practical
You mostly meet this as stream-processing checkpoints. Watch barrier alignment time per operator to find the one task holding a snapshot up, bound the buffering a node will do while recording, and always replay recorded in-flight messages on restart. A snapshot that will not complete is one specific channel, and your metrics should name it.
Advanced
The core move is replacing coordination with ordering. The marker carries the cut along the channels that already exist, so no node needs to know another node’s state or clock — the same substitution that makes Watermarks: A Guess About Time, Made Precise Enough to Act On work for event time and Lamport Clocks: Consistent With Causality, Blind to Concurrency work for causality. Where it stops working is exactly where that ordering is not available: reordering channels, dynamic membership, or a participant that fails mid-run.
Internals
The FIFO assumption can be dropped at a price. Non-FIFO variants tag every ordinary message with a snapshot epoch so the receiver can classify it regardless of arrival order, trading per-message overhead for the channel assumption. Failure tolerance is usually bought by making the snapshot itself a consensus-committed decision — a coordinator proposes epoch N, and the cut is defined by the epoch rather than by marker arrival, which is roughly what unaligned checkpointing does when it records in-flight buffers instead of waiting for barriers to line up. Each variant swaps one of the original assumptions for a concrete cost, and the choice is a good exercise in reading which assumption your system actually violates.
Apply it
- 💬 Why can you not snapshot a distributed system by having every node record at the same time?
- 💬 What makes a cut consistent, and why does it not have to correspond to a real instant?
- 💬 Walk through Chandy-Lamport and explain what the marker is actually doing.
- 💬 Which of the algorithm’s assumptions does a message broker with multiple partitions violate?