Global State and Distributed Snapshots

A distributed system consists of $N$ processes communicating by message passing over channels, with no shared memory and no perfectly synchronized clocks. The global state of the system at a conceptual instant is the tuple of all local process states together with the multiset of messages currently "in flight" on each channel — sent but not yet received. Because there is no global clock, no external observer can literally look at all processes at once; any attempt to collect states by querying processes one at a time risks capturing a picture that is causally impossible, e.g., recording that process $Q$ has received a message from $P$ while $P$'s recorded state shows it never sent it. To make "in flight" and "consistent" precise, model the execution of the system as a set of local events (send, receive, internal) totally ordered per process, with the happened-before relation $\rightarrow$ from Logical Clocks and Lamport Timestamps capturing causality across processes: if $a$ is a send and $b$ the matching receive, then $a \rightarrow b$. A cut $C$ is a subset of the global history that includes a prefix of each process's local events — formally, for each process $i$ a cut point after which no more events of $i$ are included. A cut is consistent if it is closed under happened-before restricted to receive-implies-send: whenever a receive event is in $C$, the corresponding send event is also in $C$. Equivalently, a consistent cut never "sees" the effect of a message without having "seen" its cause. The global state induced by a consistent cut is well-defined and internally coherent: for every channel, the messages "in transit" recorded by the snapshot are exactly those sent (by the sender's recorded prefix) but not yet received (by the receiver's recorded prefix). This state need not correspond to any real simultaneous instant across all machines — the local recording points can happen at wildly different real times — but it is a state that some valid re-execution of the system could pass through, which is exactly what makes properties detected on it ("no deadlock," "total money = X") trustworthy. This connects tightly to Vector Clocks, whose comparability structure captures precisely which pairs of events are causally ordered and therefore which cuts can be consistent.

The Chandy-Lamport algorithm assumes: 1. Communication channels are point-to-point, FIFO (messages arrive in the order sent), and 2. Channels are reliable (no loss, no duplication), and 3. The underlying communication graph is strongly connected (a message can reach any process from any other, possibly indirectly). The snapshot is triggered by one or more processes (an "initiator") and propagates via special control messages called markers, which are distinguishable from application messages but travel over the same channels, preserving FIFO order relative to them. The key trick: a marker on a channel acts as a divider — everything the receiver got before the marker belongs to the "before" snapshot state; everything the receiver gets after the marker (and before markers on all channels) is a message that was "in flight" at snapshot time and must be recorded as channel state.

Process-level rule, restated as the algorithm each process runs: ```pseudocode // State kept per process P recorded : boolean := false local_state : ProcessState channel_state[c] : list of messages, for every incoming channel c recording[c] : boolean := false // are we still logging channel c? // Rule 1: P initiates the snapshot (or is asked to, upon first marker) procedure INITIATE_SNAPSHOT(P): if not P.recorded: P.local_state := snapshot of P's current local state P.recorded := true for each outgoing channel c_out of P: send MARKER on c_out // before any further app messages for each incoming channel c_in of P: P.recording[c_in] := true P.channel_state[c_in] := [] // start logging (empty so far) // Rule 2: P receives a MARKER on incoming channel c procedure ON_RECEIVE_MARKER(P, c): if not P.recorded: // first marker P has seen anywhere: this triggers P's own snapshot P.local_state := snapshot of P's current local state P.recorded := true P.channel_state[c] := [] // channel c is empty in the snapshot: // marker arrived before any message did for each outgoing channel c_out of P: send MARKER on c_out for each incoming channel c_in of P, c_in != c: P.recording[c_in] := true P.channel_state[c_in] := [] else: // P already recorded; this marker closes channel c's logging window P.recording[c] := false // P.channel_state[c] now holds exactly the messages that arrived // on c strictly between P's own recording and this marker if all incoming channels have recording == false: P.SNAPSHOT_DONE() // send (local_state, channel_state[]) to collector // Rule 3: P receives an ordinary application message m on channel c procedure ON_RECEIVE_APP_MESSAGE(P, c, m): if P.recording[c]: P.channel_state[c].append(m) // log it — it was "in transit" at snapshot time deliver m to the application layer as normal ``` Step by step, the protocol unfolds as follows: 1. <strong>Initiation.</strong> One or more processes decide to start a snapshot (e.g., triggered externally, or by a monitor process). Each initiator records its own local state immediately, then sends a marker on every outgoing channel before sending any further application messages. 2. <strong>First marker received.</strong> When any process $Q$ receives a marker for the first time (on some channel $c$), it immediately records its own local state — this is $Q$'s contribution to the global snapshot — and declares channel $c$'s state as <strong>empty</strong> (since the marker beat every application message that might have been in transit on $c$, by FIFO order, and no other message can now arrive on $c$ before the marker). It then sends markers on all of its own outgoing channels, and begins recording all other incoming channels. 3. <strong>Recording in-transit messages.</strong> For every incoming channel not yet closed by a marker, $Q$ appends every application message it receives to that channel's recorded log — these are exactly the messages that were sent before the sender's snapshot point but had not yet arrived when $Q$ took its own snapshot. 4. <strong>Marker on an already-open channel.</strong> When $Q$ later receives a marker on channel $c$ (after $Q$ has already recorded), this signals that all messages sent on $c$ before* the sender's snapshot have now arrived. $Q$ closes (stops recording) channel $c$; whatever accumulated in `channel_state[c]` is final. 5. Termination. Once every process has received a marker on every incoming channel, every process has recorded its local state and every channel's in-transit messages, so the entire global snapshot is complete. A designated collector (or each process independently, then merged) assembles the full picture: $N$ local states plus, for each of the $E$ directed channels, a list of in-flight messages.

Message complexity. Exactly one marker traverses each directed channel per snapshot initiation cycle — a process sends a marker on every outgoing edge exactly once (upon initiating or upon first receiving a marker). With $N$ processes and $E$ directed channels in the communication graph, the algorithm sends exactly $E$ marker messages in total, independent of how many application messages are in flight. No process needs to pause; application messages continue to flow and are simply logged or passed through. This gives $O(E)$ control messages and $O(1)$ additional messages per channel — asymptotically optimal, since every channel must be probed at least once to know it has quiesced.

Space complexity. Each process stores one local state snapshot ($O(1)$ extra beyond its normal state) plus, per incoming channel, a list of logged in-transit messages, bounded by the number of application messages sent on that channel between the two processes' snapshot points. In the worst case this is bounded by the channel's maximum in-flight capacity (bandwidth-delay product), but is typically small in practice.

Termination. The algorithm terminates because the communication graph is strongly connected and every channel carries a bounded (eventually delivered) sequence of messages: each process sends a marker on each outgoing channel exactly once, so every process receives at least one marker on each incoming channel in finite time (assuming reliable delivery), triggering closure of every channel's recording.

Correctness — why the resulting snapshot is a consistent cut. The core invariant is: > For any channel from process $A$ to process $B$: every application message that $B$'s snapshot logs as "in transit" is exactly a message that $A$ sent before $A$'s local snapshot point, and that $B$ received after $B$'s local snapshot point. This follows from three facts guaranteed by FIFO channels and the marker rule: 1. Every message $A$ sends after taking its snapshot is sent after $A$'s outgoing marker on that channel (Rule: markers are sent immediately upon recording, before any further application sends can be attributed to the post-snapshot state — more precisely, the algorithm requires the process to send its markers as its very next action after recording, which is achievable because recording is a local, atomic step). Hence any message sent after $A$'s snapshot arrives at $B$ after $A$'s marker, by FIFO. 2. By FIFO, $B$ receives $A$'s marker before any message $A$ sent after its snapshot point. So once $B$ sees $A$'s marker, every subsequent message on that channel was necessarily sent by $A$ after $A$'s snapshot — these must not be logged as in-transit (they belong to the "future" of the snapshot), and indeed Rule 2 stops logging exactly when the marker arrives. 3. Messages $A$ sent before its snapshot but not yet received by $B$ at the time $B$ takes its own snapshot are precisely the messages $B$ receives between $B$'s own recording point and $A$'s marker's arrival — and Rule 3 logs exactly these. Together these show that the recorded channel state for $A \to B$ equals precisely: {messages sent by $A$ before $A$'s cut} $\setminus$ {messages received by $B$ before $B$'s cut}. That is exactly the definition of "messages in transit across a consistent cut" for that channel. Since this holds for every channel, the overall snapshot corresponds to a globally consistent cut: no recorded local state at $B$ reflects the receipt of a message that the recorded state at $A$ shows as unsent. This is precisely the causal-consistency condition from the happened-before relation, so the assembled global state is one the system could legitimately have passed through in some interleaving — even though the local snapshot instants may be far apart in real (wall-clock) time.

Tradeoffs. The algorithm requires FIFO, reliable channels — on non-FIFO networks (e.g., UDP without sequencing) the marker trick breaks, because an application message sent after the marker could overtake it and be misclassified. It also assumes the graph is strongly connected so markers can reach every process; on a partitioned or non-strongly-connected topology, extra logical channels (or a spanning communication structure) must be assumed. The algorithm records one possible consistent global state, not necessarily the "true" simultaneous one (which may not even be a coherent concept) — for many analyses (stable predicates: predicates that, once true, remain true) this is sufficient, but transient properties cannot be safely inferred from any single snapshot.

Consider three processes $P, Q, R$ arranged so that $P \to Q$, $Q \to R$, $R \to P$, and also $P \to R$ directly (four directed channels, $E = 4$). Suppose $P$ holds $70, $Q$ holds $50, $R$ holds $30 (total $150), and the following events occur in real time (unsynchronized clocks — times shown are illustrative wall-clock, not known to the processes):

Assembling the global snapshot: local states $P=\$60, Q=\$60, R=\$30$, sum $=\$150$; in-transit messages: channel $Q\to R$ holds $20, all other channels empty. Grand total accounting for in-transit money: $150 + 20 = 170$? That looks wrong until we notice $m_2$'s $20 was already debited from $Q$'s recorded $60 (i.e., $Q$'s $60 is after sending $m_1$ but the snapshot at $Q$ happened at $t_4$, before $Q$ sent $m_2$ at $t_6$). So $Q$'s recorded $60 does not yet reflect the $20 debit for $m_2$ — meaning $m_2$'s $20 is money that, in the snapshot's consistent-cut view, has not left $Q$ and has not arrived at $R$ simultaneously; it is legitimately "in transit," counted once, and the total is:

Wait — check conservation directly: originally $70+50+30=150$. $m_1$ ($10, $P\to Q$) is fully reflected in both endpoints by snapshot time ($P$ debited to $60, $Q$ credited to $60 — net effect on the pair is a wash, total unchanged at $60+60=120$ vs. original $70+50=120$ ✓.) $m_2$ ($20, $Q \to R$) is only debited at $Q$ (already reflected: $Q$ went from a would-be $80 after receiving $m_1$, down to $60 after sending both — but wait, $Q$'s snapshot was taken at $t_4$, before $m_2$ was sent at $t_6$, so $Q$'s recorded $60 has not yet subtracted $20 for $m_2$). Let's recompute cleanly: $Q$ received $10 (now $60) then snapshot recorded — snapshot value $60. Later $Q$ sends $20 (now really $40 in $Q$'s actual post-snapshot state, but that's irrelevant to the snapshot). $R$'s snapshot recorded $30 before $m_2$ arrived. So the snapshot's version of the world is: $P=\$60$, $Q=\$60$, $R=\$30$, plus $20 "on the wire." Total $= 60+60+30+20 = 170 \ne 150$. The resolution: this reveals exactly why we must double check the trace — re-examine event $t_2$: $P$'s recorded state must reflect that $P$ already sent $m_1$ before recording (event order: $t_1$ send, $t_2$ record), so $P$'s recorded $60 is correct (started at $70, sent $10). $Q$ received $m_1$ at $t_3$, before recording at $t_4$, so $Q$'s recorded state correctly includes the $10 (now $60, from $50). So far $P+Q = 60+60=120$, matching pre-transfer $70+50=120$ — consistent, $m_1$ is "settled" and correctly not double-counted (it's not in-transit; it's been absorbed on both sides in causal order). Now for $m_2$: $Q$ sends it at $t_6$, which is after $Q$'s snapshot at $t_4$ — so this send is not part of $Q$'s recorded history at all. In a truly consistent cut, an event not yet recorded at the sender cannot have its effects appear anywhere in the snapshot, including as an "in-transit" message — because "in transit" means "sent-before-cut but not-yet-received-before-cut." Since $m_2$ was sent after $Q$'s cut, it should not appear in the channel-state log at all under a correct execution of the algorithm. This is exactly what the FIFO/marker mechanism prevents: at $t_8$, $Q$ sends the marker on $Q\to R$ immediately upon recording, i.e., at $t_4$ (per Rule: markers go out right after recording, on every outgoing channel, not deferred to $t_8$). Correcting the trace: $Q$'s marker on $Q\to R$ is actually sent at $t_4$, arriving at $R$ before $t_6$ (assuming reasonable delay), so by FIFO $m_2$ — sent at $t_6$, after the marker — arrives at $R$ after the marker, and Rule 2/3 dictates $R$ has already closed channel $Q\to R$'s recording by the time $m_2$ shows up, so $m_2$ is delivered to $R$'s application layer normally but is correctly excluded from the snapshot. The corrected snapshot is $P=60, Q=60, R=30$, no in-transit messages, total $=150$ ✓ — consistent, because $m_2$ belongs entirely to the causal future of the cut, on both ends. This trace illustrates the single most important operational detail: markers must be sent as the immediate next action after recording, on every outgoing channel, precisely so that no application message a process sends after its own snapshot can be misfiled as "in transit" at a downstream process that hasn't yet closed that channel.

The algorithm is named for two of the foundational figures in distributed computing theory. K. Mani Chandy, a systems and parallel-computing researcher, and Leslie Lamport, whose earlier work on logical clocks in "Time, Clocks, and the Ordering of Events in a Distributed System" (1978) supplied the causal machinery — the happened-before relation — on which the notion of a consistent cut depends. Their 1985 paper, "Distributed Snapshots: Determining Global States of Distributed Systems," gave the marker-based protocol described here and proved its consistency and termination properties rigorously, framing global-state detection as a first-class problem separate from, but foundational to, deadlock detection, checkpointing, and later consensus protocols.

Consistent global snapshots are a quiet workhorse underneath many production systems. Distributed stream-processing engines use a Chandy-Lamport-style barrier mechanism to checkpoint the state of long-running pipelines: a "barrier" record travels through the dataflow graph exactly like a marker, and each operator snapshots its state when barriers on all input streams have arrived, giving exactly-once processing guarantees after a failure and restart. Distributed deadlock detection algorithms build a global wait-for graph from a consistent snapshot, since checking for cycles on an inconsistent picture can produce phantom deadlocks or miss real ones. Distributed debugging and monitoring tools use consistent cuts to evaluate stable predicates — properties, like "no process holds a lock," that remain true once true — safely across an unsynchronized cluster. Database systems that need consistent, non-blocking backups of replicated or partitioned state rely on the same core insight, and the causal reasoning here directly foreshadows the coordination challenges tackled by Consensus Problem (agreeing on a single value despite failures) and by Distributed Transactions and Two-Phase Commit (agreeing on whether to commit a transaction spanning multiple machines) — both of which must, at bottom, reason about consistent global states under uncertainty.

Ask the tutor: How would the Chandy-Lamport algorithm need to change if channels could reorder messages instead of being FIFO?