Skip to content

Coordination and distributed state

Leader election, consensus, distributed transactions and unique ids.

01

Leader election

advanced

Many systems need exactly one node doing something: running a scheduled job, accepting writes, compacting a table, coordinating a cluster. Leader election is how that one node is chosen and how it is replaced when it dies, and the whole difficulty lies in the second half.

Two nodes both believing they lead is a split brain, and it corrupts data quietly rather than loudly. Both are behaving correctly by their own reasoning; the system has simply given two nodes the same certainty. That is why the failure is usually discovered later, in inconsistent data, rather than at the moment it happens.

The naive implementation is a lock row with a timeout, and it fails for a reason worth internalising: there is no bound on how long a process can be paused. Garbage collection, a hypervisor descheduling the VM, a page swapped to disk, a suspended laptop. The lease expires, a new leader is elected, and the old one wakes up with no idea that any time has passed and writes as though it still leads.

You cannot prevent the pause, so the storage layer has to reject the write. Fencing tokens are the mechanism: the lock service issues a monotonically increasing number with each grant, every write carries it, and the resource refuses anything with a number lower than the highest it has seen. The zombie's write is rejected by the thing being written to, which is the only layer that can know.

Leases need clocks, and clocks are the other soft spot. A lease based on absolute timestamps depends on clock synchronisation between the holder and the service; one based on elapsed monotonic time does not, and monotonic time is what you want here. Anything comparing wall-clock timestamps across machines to decide who leads is depending on NTP for correctness, which is a dependency nobody chose.

Almost nobody should implement this from scratch. etcd, ZooKeeper and Consul exist, do it correctly and have had their edge cases found by other people, and a database lease with fencing tokens is a reasonable middle path when adding a coordination service is not worth it. The good outcome is to spend your thinking on what the leader does rather than on how it is chosen.

Check yourself

Why is split brain usually discovered long after it happens?

grant 43write with 43, acceptedwrite with 42, rejectedOld leaderpaused 30sNew leadertoken 43Lock serviceissues rising tokensStoragehighest seen: 43
ServiceEdge / CDNData storeDashed = asynchronousConsidered, not chosen
You cannot stop the pause, so the resource rejects the write
Evidence · 1 source
  1. 01
    Kleppmann, How to do distributed locking

    That a process pause invalidates a lease-based lock without the holder knowing, and that fencing tokens checked by the resource are the mitigation.

Ask about this

Answers are generated and can be wrong. The topic above is the reviewed version.

02

Consensus: Raft and Paxos

advanced

Consensus is getting a group of nodes to agree on a value despite some of them failing or being unreachable. Paxos came first and is famously hard to reason about; Raft was designed for understandability, which is why it is what almost everything built since uses. Both give the same guarantee: one agreed sequence of decisions that survives a minority failing. The mechanism is a majority quorum. A value is committed once more than half the members have accepted it, which means two conflicting majorities cannot exist, because any two majorities of the same group overlap in at least one member. That overlap is the whole proof, and it is worth being able to state, because everything else in the algorithm is machinery serving it.

Cluster sizing follows from arithmetic rather than from tradition. Tolerating f failures needs 2f+1 members, so 3 tolerates one and 5 tolerates two. Four members tolerate exactly one, the same as three, while costing an extra machine and an extra acknowledgement, which is why even sizes are avoided. The common explanation about avoiding tie votes is wrong: a majority of four is three, so a majority cannot tie.

Raft's structure is worth knowing in outline because it explains the failure modes. One leader takes all writes and replicates them to followers; a term number increases with each election so stale leaders are recognisable; a follower that misses heartbeats becomes a candidate and asks for votes. Randomised election timeouts stop every follower standing at once, which is the same desynchronisation trick as jitter in a retry policy.

The cost is a round trip to a quorum on every write, so consensus is correctness bought with latency, and the price is set by geography. A cluster inside one data centre pays a millisecond; one spanning continents pays the worst inter-region round trip on every single write, which is why global systems shard so that each shard's consensus group stays local rather than stretching one group across the world.

Finally, know what it is not for. Consensus is for metadata, membership, leadership, configuration, small critical state. Running your primary data path through a consensus group means every write pays quorum latency, which is why systems that need both usually keep a small consensus group deciding who may write and a fast replication path doing the writing.

Check yourself

Why do Raft implementations randomise the election timeout?

replicatereplicateno answercommittedcommittedcorrectness bought withlatencyWriteproposed by theleaderNode 1acceptsNode 2acceptsNode 3unreachableMajority of 32 is enough,tolerates 1 lossA quorum round tripper write, priced bydistance
ClientData storeServiceExternalDashed = asynchronousConsidered, not chosen
Any two majorities overlap, which is the entire proof
Evidence · 1 source
  1. 01
    Ongaro and Ousterhout, In search of an understandable consensus algorithm (USENIX ATC 2014)

    That Raft was designed for understandability rather than novelty, the 2f+1 sizing, term numbers, and randomised election timeouts.

Ask about this

Answers are generated and can be wrong. The topic above is the reviewed version.

03

Distributed transactions and sagas

advanced

A transaction across several services cannot use a database transaction, because there is no single database holding the locks. Two-phase commit exists and does provide atomicity, and it is rare in practice for one reason: it blocks. If the coordinator fails after participants have prepared, they hold their locks and wait, and a system where one component's crash freezes several others is a poor trade for a guarantee you can usually work around.

A saga is the usual alternative: break the work into local transactions, each committing on its own, each paired with a compensating action that undoes its effect. Book the flight, book the hotel, charge the card; if the hotel fails, cancel the flight and refund. Every step commits immediately, so nothing is held, and the price is that partial states are real and visible.

Compensation is not rollback, and the distinction is the entire design problem. A rollback erases history; a compensation is a new fact that offsets an old one. The flight was booked and then cancelled, and both appear in the record. Some actions have no compensation at all: you cannot unsend an email or unpublish something someone screenshotted, which forces you to order the saga so the irreversible steps come last.

Sagas are orchestrated or choreographed, and the choice matters more than it sounds. An orchestrator holds the state machine explicitly: one service knows the sequence, which makes the flow readable and the coordinator a component to run. Choreography spreads it across event handlers with no central owner, which couples less and means the actual sequence exists only in the aggregate of everyone's subscriptions, and can be reconstructed only by reading them all.

Every step needs to be idempotent, because every step will be retried. A saga is a long-running process across unreliable networks, so duplicate deliveries and repeated compensations are normal operation rather than an exception. A compensation that runs twice must leave the same state as one that runs once, which usually means recording that it has run rather than checking whether it needs to.

The last piece is visibility. A saga in progress is a state machine with a current step, and it needs to be inspectable: which sagas are running, which are stuck, which compensations failed. Without that you have a distributed process nobody can see, and the first sign of trouble is a customer noticing that a booking exists with no payment against it.

Check yourself

How should a saga be ordered when one step cannot be compensated?

step onestep tworun backwardsnot a rollbackonly if all else heldBook a tripthree services, noshared lockFlight bookedcommitted locallyHotel failscommitted nothingCard chargedirreversible, so itgoes lastCompensatecancel the flight,refundBoth facts remainbooked, thencancelled
ClientServiceData storeExternalDashed = asynchronousConsidered, not chosen
Each step commits, so each step needs an offsetting action
Ask about this

Answers are generated and can be wrong. The topic above is the reviewed version.

04

Generating unique ids at scale

intermediate

Auto-increment ids are perfect until they are not. They need a single coordinator to hand out the next value, which is a bottleneck at high write rates and stops working entirely once the table is sharded, because two shards would issue the same numbers. They also leak information: sequential ids in a URL tell a competitor your order volume and let anyone enumerate your records.

Random UUIDs solve coordination completely and cost you index locality. Version 4 is random by construction, so consecutive inserts land in unrelated parts of the B-tree, which means constant page splits, a much larger working set in memory, and write performance that degrades as the table grows. At 16 bytes they also enlarge every secondary index that references them.

Snowflake-style ids are the usual answer at scale: a timestamp in the high bits, a machine identifier, and a per-millisecond counter, packed into 64 bits. Unique without coordination, sortable by time, and small enough to be a comfortable primary key. The cost is that each generator needs a distinct machine id, which is a small piece of configuration that must genuinely be unique, and a clock that does not move backwards.

UUIDv7 is the standardised version of the same idea and is the sensible modern default when a UUID is wanted: a millisecond timestamp in the leading bits followed by randomness, so it is globally unique with no coordination and still inserts near the end of the index. It gives most of Snowflake's benefit with none of the machine id configuration.

Time-ordered ids have a cost worth naming: they disclose creation time and, with enough samples, creation rate. Where that matters, the fix is a separate opaque public identifier for external use, with the ordered id kept internal. That is also the clean answer to enumeration, since the internal key can stay sequential and useful while nothing outside can walk it. The general shape here appears in many places: coordination buys ordering, and avoiding coordination costs locality. The good designs recover the ordering some other way, usually by putting time in the high bits, rather than by reintroducing the coordinator they were trying to remove.

Check yourself

What does a Snowflake generator need that a random UUID does not?

simplestno coordinationboth propertiesthe costthe costsortable and uniquestill discloses timeAn id per rowat high write rates,shardedAuto-incrementone coordinator,enumerableUUIDv4random, nocoordinationUUIDv7 or Snowflaketime in the high bitsBottleneck and leakreveals volume,walkablePage splitsinserts landeverywhereInserts at the endlocality kept, nocoordinatorSeparate public idopaque, for externaluse
ClientServiceData storeExternalDashed = asynchronousConsidered, not chosen
Coordination buys ordering; putting time in the high bits buys it back
Evidence · 1 source
  1. 01
    RFC 9562: universally unique identifiers

    That UUIDv7 is standardised, that it places a millisecond timestamp in the leading bits, and that it is intended to give time-ordered locality without a coordinator.

Ask about this

Answers are generated and can be wrong. The topic above is the reviewed version.