05

Replication

Replication means keeping a copy of the same data on several machines. You do it to survive failures, to serve more reads, and to keep data close to users. If the data never changed this would be easy. The difficulty is in handling changes: who accepts writes, how changes reach the other copies, and what readers see in the meantime. Kleppmann covers three answers (a single leader, several leaders, no leader at all). Each one trades durability, availability and consistency differently. Below, the leader really streams its log, followers really lag behind, and quorums really overlap or miss.
The client writes to the leader, which appends the write to its replication log and streams entries to three followers at different distances. The mode decides when the client gets its acknowledgement: right away, after the fastest follower confirms, or after all of them do.

Leader streams its log to followers

t = 0.0 s
clientAleader·e1log length 0Boff 00 behindCoff 00 behindDoff 00 behind

Squares are log entries on their way to followers; green dots are follower acknowledgements. Click a follower to take it down or bring it back.

Replication logs, aligned by offset

A ldrBCDoffset 0offset 0 · filled = acknowledged to client · hollow = not yet · ✕ = acknowledged then lost
acknowledged writes
0
client wait per write
0 ms
leader-local only
writes blocked
0
waiting for follower acks
acked, only on leader
0
would vanish if the leader died now
acknowledged writes lost
0
0 failovers

Try: in asynchronous mode, raise the lag to 1.5 s and crash the leader. The acknowledged writes the new leader never received are lost (✕). Switch to synchronous and nothing acknowledged is lost, but take one follower down (click C) and every write blocks. Semi-sync keeps both properties as long as one follower is healthy.

Failover: detecting a dead leader, and the ways it goes wrong

repl-failoverrepl-split-brain
Followers watch the leader's heartbeats. After a timeout with no heartbeat, the most up-to-date follower is promoted with a new epoch number. A frozen leader looks exactly like a dead one, and what happens when it comes back depends on whether the system fences it.
fencing

Asynchronous replication with automatic failover

t = 0.0 s
clientAleader·e1log length 0Boff 0Coff 0Doff 0

Click a downed node to restart it. A restarted old leader still believes it is the leader.

Logs: entries colored by the epoch of the leader that accepted them

A ldrBCDoffset 0offset 0 · filled = acknowledged to client · hollow = not yet · ✕ = acknowledged then lost
failovers
0
writes refused
0
no leader reachable
acked writes lost
0
writes to old leader
0
split-brain writes

Event log

  1. Press Play, then kill or freeze the leader.

Try: freeze the leader for 2.5 s with a 1.5 s timeout. The followers give up on it and elect a new leader, even though nothing crashed. The old leader wakes up still thinking it is in charge, and with fencing off clients keep writing to it (red entries): split brain. Turn fencing on and repeat: it sees the newer epoch, steps down, and throws away its unreplicated writes. Raise the timeout above the freeze and no failover happens, but a real crash then takes longer to recover from.

This has happened in production. In one GitHub incident, a MySQL follower that was behind was promoted, and the auto-increment primary keys it then handed out had already been used by the old leader. Those ids were also stored in Redis, so some private data was shown to the wrong users. Throwing away writes is never purely a database-internal decision.

Replication lag: your post vanished, and time ran backwards

replication-lagread-your-writesmonotonic-reads
Reading from followers scales reads, but those followers are eventually consistent. Here you post comments and reload the page while the load balancer sends reads to a fast or a slow follower. Each read is checked against two guarantees, and you can switch the routing rule that provides them.
route reads
other users posting
autopilot user

Space-time diagram (last 10 s)

t = 0.0 s
youleader#0follower 1#0follower 2#0
━ your posts━ other users' posts■ read OK ■ anomalydashed = still replicating · #n = applied log position
page loads
0
served by leader
–
extra load on the leader
own post missing
0
read-your-writes violated
went back in time
0
monotonic reads violated
last page load
Press Play: the autopilot user posts, then reloads twice.

Try: with “any follower”, about half the reloads land on the slow follower 2 and your post seems to disappear, or the page shows fewer comments than a moment ago. “Read-your-writes” fixes the first problem only. “Monotonic” pins you to follower 2, so you never go back in time, but you still miss your own post until it catches up. “Both” tracks the newest position you have written or seen, and only replicas at least that far along serve your reads.

Consistent prefix reads: the answer before the question

consistent-prefix
Figure 5-5 of the book. In a partitioned database each partition replicates on its own, so there is no global order of writes. A causally dependent pair of writes can reach an observer in the wrong order.

Two partitions, each with its own leader and follower

Mr. PoonsP1 leaderP1 followerMrs. CakeP2 leaderP2 followerobserverwrites questionreads itwrites answeranswer visible before question0s2s4s6s8s

What the observer sees at 5.0 s

Mrs. Cake
About ten seconds usually, Mr. Poons.
Mr. Poons
How far into the future can you see, Mrs. Cake?
anomaly window
1.8 s
from 2.0 s to 3.8 s

Try: drag the playhead into the red band. The P2 follower is fast and the P1 follower is slow, so the observer sees the answer first. Lower the P1 lag below the gap between question and answer and the anomaly disappears without any fix. That is the danger: it only appears under load. Switch to “same partition” and the single follower log keeps the writes in order whatever the lag.

Multi-leader: two datacenters edit the same page

multi-leaderconflict-resolutionrepl-lww
Each datacenter has its own leader, so writes stay local and survive a broken inter-datacenter link. When both users change the wiki title before hearing about the other's change, each leader detects a conflict: it holds a value the incoming write had never seen. How it resolves the conflict decides whether the replicas converge, and what gets lost.
conflict resolution
link

Two datacenters, a leader in each

converged
datacenter 1 · leaderclock 0.00Apage title1 writes seendatacenter 2 · leaderclock 0.00Apage title1 writes seen
conflicts
0
concurrent writes detected
writes discarded
0
overwritten

Event log

  1. Press “Both at once” to write concurrently in both DCs.

Try: “Both at once” with arrival order. Each DC applies the other's write last, so DC 1 ends at C and DC 2 at B, and they stay different forever. Switch to LWW and they converge, but one write is silently dropped. Now set the DC 2 clock to −1.5 s, let user 1 write, and have user 2 write a moment later: LWW keeps the older write. Merge and siblings lose nothing. Cut the link, write on both sides, then restore it to see offline edits conflict.

Leaderless quorums: w + r > n

leaderlessquorumread-repairsloppy-quorum
Dynamo-style replication. The client writes to all n replicas of a key and treats the write as successful after w acknowledgements. It reads from r replicas and keeps the highest version. Take nodes down, write and read, and count how many of the possible read sets could miss the latest write.
read repair
sloppy quorum
w + r = 4 > n = 3 → every read set overlaps every write set

One key, n home replicas (click a node to make it unreachable)

other nodes (not home)clientN1v1N2v1N3v1X1–X2–

Green ring = holds the latest successful write (v1). Red ring = stale. White ring = in the last read set. Yellow lines = writes accepted by nodes outside the key's home set.

read sets that miss
0%
0 of 3 possible r-subsets
stale reads
0/0
latest success
v1
failed writes
0
not rolled back

Event log

  1. Take a node down, write, bring it back, then read.

Try: with n=3, w=2, r=2, click N3 to take it down, write v2, bring N3 back. N3 is stale, but every 2-node read set includes a fresh node, and read repair fixes N3 when it gets read. Now set r=1: one read in three hits the stale node. Then set r=2, turn sloppy quorum on, take N2 and N3 down, and write. X1 accepts the write on N2's behalf, so w is met. Bring N2 and N3 back and take N1 down: every read is stale, even though w + r > n, until you run the hinted handoff.

Even with w + r > n, the book lists edge cases. A sloppy quorum may place the w writes outside the n home nodes. Concurrent writes need a way to order them, and LWW loses some. A write concurrent with a read may or may not be seen. A write that failed on most nodes is not rolled back on the ones that took it. And a node holding a new value can fail and be restored from a stale replica. Quorums make stale reads less likely. They don't make the system linearizable.

Concurrent writes: happens-before and version vectors

repl-happens-beforeversion-vector
Two clients add items to the same shopping cart. The server can't use clocks to tell whether one write overwrote the other or whether they were concurrent. Instead it uses version numbers, one per replica, that clients send back with each write.
server
replicas

Client 1

talks to A
local cart (merged siblings)
empty
version it will send
none

Client 2

talks to A
local cart (merged siblings)
empty
version it will send
none

Database

empty

Causality: which write overwrote which

No writes yet.
items added
0
siblings now
0
items lost
0
union of siblings has everything

Try: step through Figure 5-13. Each client sends the version it last read. The server overwrites exactly the values that version covers and keeps the others as concurrent siblings, so after five writes nothing is lost. Switch to last write wins and repeat: the clients overwrite each other and items vanish. With two replicas, each client writes to its own replica, and the conflict only appears once you sync them.

  • takeawayA single leader gives one write order and no conflicts. You still choose between waiting for followers (durable but fragile) and not waiting (fast, but acknowledged writes can be lost on failover). Failover needs a good timeout and fencing to avoid split brain.
  • takeawayFollowers are eventually consistent, with no bound on the lag. Decide which guarantees your users need (read-your-writes, monotonic reads, consistent prefix) and route reads accordingly. Don't pretend replication is synchronous.
  • takeawayMulti-leader and leaderless systems tolerate failures and latency, but concurrent writes are a normal event for them. They need deterministic conflict resolution and causality tracking (version vectors), or they will silently lose data, as LWW does.

Key concepts · 21

book chapter 5
  1. Single-leader replication

    One replica (the leader) accepts all writes and streams them, in order, to read-only followers.

    There is a single order of writes, so there are no write conflicts. The cost is that every write goes through one node, and reads from followers may be out of date.

    PostgreSQLMySQLMongoDBKafka
    ↑ see it in the visualization
  2. Synchronous replication

    The leader confirms a write to the client only after a follower has also confirmed it.

    An acknowledged write survives the loss of the leader. But one slow or failed follower blocks every write, so a fully synchronous setup is fragile.

    ↑ see it in the visualization
  3. Semi-synchronous replication

    One follower is synchronous and the rest are asynchronous. If the synchronous one becomes slow, an asynchronous follower takes its place.

    This guarantees an up-to-date copy on at least two nodes without letting a single straggler stop all writes.

    MySQL semi-syncPostgreSQL synchronous_standby_names
    ↑ see it in the visualization
  4. Asynchronous replication

    The leader acknowledges a write as soon as it is stored locally and sends it to followers in the background.

    Writes are fast and keep working when followers fail. However, writes that were acknowledged but not yet replicated are lost if the leader dies.

    ↑ see it in the visualization
  5. Replication log

    The stream of changes a leader sends to its followers. It can be statement-based, write-ahead-log shipping, a logical (row-based) log, or trigger-based.

    Replicating statements breaks on nondeterminism (NOW(), RAND()). Shipping the WAL ties followers to the leader's storage format. A logical row log is decoupled from the storage engine and can feed change data capture.

    MySQL binlogPostgreSQL WALDebezium
    ↑ see it in the visualization
  6. Follower catch-up

    A follower that restarts or falls behind asks the leader for every change after the last log position it processed.

    Because the log has positions, a follower that was down can recover without a full copy. New followers start from a consistent snapshot and then catch up the same way.

    ↑ see it in the visualization
  7. Failover

    When the leader fails, the system detects it (usually with a timeout), promotes the most up-to-date follower, and redirects clients and other followers to it.

    If the timeout is too short, the system fails over when nothing is wrong. If it is too long, recovery is slow. With async replication, writes the old leader had not replicated are thrown away.

    ↑ see it in the visualization
  8. Split brain

    Two nodes both believe they are the leader and both accept writes.

    Their data diverges and there is no automatic way to reconcile it. Systems use fencing (STONITH, epoch numbers) to shut down or reject the deposed leader.

    ↑ see it in the visualization
  9. Replication lag & eventual consistency

    The delay between a write on the leader and that write appearing on a follower. In normal operation it is under a second, but under load or during failures it can be minutes.

    Asynchronous followers converge only eventually, and the delay has no upper bound. Applications that read from followers have to decide which anomalies they can tolerate.

    ↑ see it in the visualization
  10. Read-your-writes consistency

    A user always sees their own updates after a page reload. Other users' updates may still be delayed.

    Without it, a user who has just posted something reloads the page and the post seems to be gone. Fixes include reading recently modified data from the leader, or tracking the user's last write position and serving reads only from replicas that have caught up to it.

    Facebook's read-from-leader window
    ↑ see it in the visualization
  11. Monotonic reads

    If a user has read a value, later reads never return an older one.

    Reads that hit a fresh replica and then a stale one make time appear to move backwards. Pinning each user to one replica (for example by a hash of the user ID) prevents this.

    ↑ see it in the visualization
  12. Consistent prefix reads

    Anyone reading a sequence of writes sees them in the order they happened.

    In a partitioned database, partitions replicate independently, so an answer can show up before its question. Putting causally related writes in the same partition, or tracking their dependencies, fixes it.

    ↑ see it in the visualization
  13. Multi-leader replication

    Several nodes accept writes (for example one leader per datacenter) and exchange their changes asynchronously.

    Writes are local in each datacenter and the system survives a datacenter or network outage. In exchange, the same record can be changed concurrently in two places, and those conflicts have to be resolved.

    MySQL TungstenBDR for PostgreSQLCouchDBGoogle Docs
    ↑ see it in the visualization
  14. Conflict resolution & convergence

    A way to make every replica reach the same final value after concurrent writes: last write wins, replica priority, merging values, or keeping all versions for the application or user to resolve.

    If each replica simply applies writes in the order they arrive, replicas end up permanently different. Any automatic rule has to be deterministic so that all replicas reach the same result.

    CRDTsBucardo custom handlersRiak siblings
    ↑ see it in the visualization
  15. Last write wins (LWW)

    Each write carries a timestamp, the highest one is kept, and the others are silently discarded.

    Replicas converge, but at the cost of durability: concurrent writes are lost, and with clock skew even a write that really happened later can lose.

    CassandraRiak (optional)
    ↑ see it in the visualization
  16. Leaderless replication

    The client (or a coordinator) sends every write and every read to several replicas in parallel. No node is in charge of ordering writes.

    There is no failover, and a few slow or dead nodes are tolerated. Stale replicas are repaired by readers and by background processes.

    DynamoCassandraRiakVoldemort
    ↑ see it in the visualization
  17. Quorum reads and writes (w + r > n)

    With n replicas, a write must be confirmed by w of them and a read must query r of them. If w + r > n, every read set overlaps every write set.

    Overlap means at least one queried replica has the latest successful write. There are still edge cases: sloppy quorums, concurrent writes, failed writes that are not rolled back, and a node restored from a stale replica.

    ↑ see it in the visualization
  18. Read repair & anti-entropy

    Read repair: a client that sees a stale response writes the newer value back to that replica. Anti-entropy: a background process compares replicas and copies the missing data.

    Read repair only fixes values that get read. Without anti-entropy, rarely read keys can stay stale on some replicas indefinitely.

    ↑ see it in the visualization
  19. Sloppy quorum & hinted handoff

    When a key's home nodes are unreachable, writes are accepted by other nodes. Those nodes hand the data back to the home nodes once they recover.

    Writes stay available during network problems, but w + r > n no longer guarantees a fresh read, because the w writes may land outside the n home nodes.

    RiakCassandra hinted handoffVoldemort
    ↑ see it in the visualization
  20. Happens-before & concurrency

    Operation B happens after A if B knew about A, depended on it, or built on it. If neither knew about the other, the two are concurrent, whatever their wall-clock times.

    When one operation happened before the other, the later one may overwrite the earlier. When they are concurrent, there is a real conflict and both values have to be kept or merged.

    ↑ see it in the visualization
  21. Version vectors & siblings

    One version counter per replica, sent with every read and write. The server overwrites the values a write has seen and keeps concurrent values as siblings.

    This tells overwrites apart from concurrent writes without depending on clocks, so no data is silently dropped. The application merges siblings, for example by taking the union of shopping carts (deletions need tombstones).

    Riak dotted version vectorsVoldemort
    ↑ see it in the visualization