You can't photograph a distributed system

Ask every server for its balance at the same moment and add the numbers. You can get a total that never existed, and nothing in your code will tell you so.

Take the smallest bank in the world. Account A lives on one machine, account B on another, and each starts with $500. The bank holds $1,000. Now A sends $100 to B. The money leaves A as a message, travels over the network for a few milliseconds, and lands in B.

You want an audit, so you ask both machines for their balance.

A answers first, just after it sent the money: $400. B answers just before the message arrives: $500. Your total is $900, and the bank has lost a hundred dollars.

Shift the timing by a hair. A answers just before it sends: $500. The transfer goes out and lands, and then B answers: $600. Now the total is $1,100, and the bank has printed money.

Both machines told you the truth. The bug lives in a word that appears nowhere in your code: the now in "add up the balances right now." In a distributed system, there is no now.

A better clock won't save you

The obvious fix is to timestamp everything. Read each balance "as of 12:00:00.000" and you're done, right?

Leslie Lamport took that idea apart in 1978, in Time, Clocks, and the Ordering of Events in a Distributed System. If you define correctness in terms of physical time, he pointed out, the system needs real clocks, and real clocks "are not perfectly accurate and do not keep precise physical time." So he defined the order of events without them.

What a system can actually know is narrower: whether one event happened before another. A send happens before its receive. Two events on one machine happen in the order that machine saw them. Everything else is concurrent, neither before nor after, and "which came first?" has no answer.

Justin Sheehy gave the idea its best title in an ACM Queue essay in 2015: there is no now.

If you doubt it, look at what Google built. Spanner, its globally distributed database, has GPS receivers and atomic clocks in its datacenters. Even with all that hardware, its clock API, TrueTime, doesn't return a time. It returns an interval, the earliest and latest the time could be. The paper reports that the uncertainty, half that interval's width, typically swings between about 1 and 7 milliseconds. Spanner then deliberately waits out that uncertainty before it lets anyone see a commit.

Google, with atomic clocks, gets a range. Your servers, which set their clocks over the network with NTP, certainly don't get an instant.

The photographers

So you can't freeze the system and you can't trust the clocks. Can you take a picture at all?

In 1985, K. Mani Chandy and Lamport said yes, in Distributed Snapshots. Near the top of the paper sits the best metaphor in computer science: a group of photographers trying to capture "a sky filled with migrating birds," a scene too big for one camera. They can't ask the birds to hold still. Each photographer shoots their own patch at a slightly different moment, and the job is to make the stitched-together picture mean something.

Lamport's own notes on the paper are worth the click. Chandy posed the problem over dinner, they'd both had too much wine to think about it, and Lamport worked out the answer in the shower the next morning. When he got to Chandy's office, Chandy was waiting with the same solution.

The algorithm fits in a paragraph. One machine starts by writing down its own state, its balance in our bank. Then it sends a special message, a marker, down every outgoing channel, the one-way links to its neighbors. When any machine receives its first marker, it records its own state and sends markers down all of its outgoing channels. After that, it keeps listening on each of its other incoming channels and writes down every message that arrives, until the marker comes down that channel too. Those messages are the channel's state: what was in flight when the picture was taken.

The one requirement is that channels are first-in, first-out. A marker can never overtake a message sent before it. That's what makes the line between "before the picture" and "after the picture" clean.

Back to the bank. A records $400 and sends a marker after the $100. B hears about the snapshot through a third machine, C, records $500, and starts listening on the channel from A. The $100 arrives, so B writes it down. Then A's marker arrives and B stops. The snapshot says: A has $400, B has $500, and $100 is on the wire. That adds up to $1,000. The money in flight is the whole trick.

Consistent cuts

There's a neat way to see why this works. Draw each machine as a timeline running left to right, and each message as an arrow from one timeline to another. A snapshot is a line cutting across all the timelines, and everything left of the line is "before the picture."

A cut is consistent if every receive left of the line has its send left of the line too. An inconsistent cut holds a photo of a letter that arrived but was never posted. That's your $1,100. The reverse, a send before the line and its receive after, is fine, as long as you record that message as in flight. That's the $100 you lost in the first audit.

The marker algorithm is a machine for drawing consistent cuts without stopping anything.

The picture of a moment that never happened

Here's the part that bothered me when I first worked through the paper. The snapshot isn't a photo of a real instant. Chandy and Lamport say so themselves: "Of what use is the algorithm if the recorded global state never occurred?"

Their answer is that the recorded state is one the system could have passed through. It's reachable from where the system started, and where the system ends up is reachable from it. For some questions that's all you need. Some properties never go away once they become true: the computation has finished, or every machine is waiting on another and none can move. If the snapshot shows one of those, it's true now too. That's why the paper pitches the algorithm at exactly those jobs: detecting deadlock, and detecting that a computation has ended.

In grad school I was on a team that built a simulator for distributed algorithms in Scala on Akka actors, small workers that talk only through mailboxes: snapshots, deadlock detection and leader election. Actors make the "could have happened" part impossible to ignore. Each mailbox delivers whenever the scheduler gets around to it, so the same snapshot on the same network can come back with two different pictures. Both are correct. We also built Lai and Yang's variant, which drops the first-in, first-out rule by coloring every message instead of sending markers. The bookkeeping changes. The idea doesn't: you either account for what's on the wire, or your picture lies.

Where it runs today

This isn't a museum piece. Apache Flink, a system that processes an endless stream of events as they arrive, takes its checkpoints with a variant of this algorithm, described in Lightweight Asynchronous Snapshots for Distributed Dataflows. Flink calls its markers barriers, and its documentation calls them a core element of its snapshotting. They flow in line with the records, and they never overtake them.

Flink has two modes for what's in flight, and between them they cover the only two honest options. In the first (Flink calls it aligned), a stage that gets a barrier on one input stops reading that input, and keeps reading the others until the barrier shows up on all of them. Everything ahead of the barriers gets processed first, so the wire is empty when the stage takes its picture. In the second (unaligned), the barrier jumps the queue, and in the docs' words, "checkpoints can overtake all in-flight data as long as the in-flight data becomes part of the operator state." The docs note that this version is closer to Chandy and Lamport's original.

Wait until the wire is empty, or write down what's on it. Pretending it's empty is not on the list.

Your dashboard has this bug

You probably don't run a stream processor. You almost certainly have a dashboard that adds up numbers from different services.

Say you run a small shop. The orders service says 1,000 orders placed today. The payments service says 990 charged. Is that ten customers who never paid, or ten payment events sitting in a queue? The dashboard can't tell you, because it summed two counters read at two different moments and threw away the only thing that would explain the gap. A queue is money in flight.

A cleverer dashboard won't fix it. Make "in flight" a state your code can name. A transfer that has left A but not reached B should be a row with a status of its own, not a gap between two balances. An order that's been placed but not charged should say so. If a message can be on the wire, your data model needs a word for "on the wire," and your reports need to count it.

A distributed system doesn't have a now. It has a set of messages that haven't arrived yet, and the honest picture includes them.