Deep-Read · DDIA · Chapter 5
Designing Data-Intensive Applications · Ch 5 · Martin Kleppmann · 2017
The data behind your bank app or online cart isn't stored on just one machine — it's copied several times and spread across different servers, data centers, even continents. DDIA Chapter 5 is about exactly that: why we copy data, how we copy it, and what goes wrong while copying. The technical name for "copying" here is replication.
Three plain reasons. One: to survive failures. With a single copy, one crashed machine takes the whole site down; with three copies, lose one and two carry on. Two: to be close. If the data lives in the US, every request from Asia crosses half the planet — slow; keep a copy nearby instead. Three: to share the load. A hundred million people reading at once will crush one machine, so you copy the data ten times and let readers spread out.
Copying a static file is trivial. The real headache is that data keeps changing. The moment you change your nickname, that change has to reach every copy. Copying takes time, which creates the most famous pitfall of all: right after you change something, other people may still see the old copy — like a letter that's been mailed but hasn't arrived yet. That gap is called replication lag.
If data can be changed, you must rule on who has the say, or chaos follows. This chapter gives three schemes, and the rest of the book rests on them:
① One machine has the say (single-leader). Pick one copy as the "master ledger"; every change goes there first, then it copies the change out to the others. Clean, no clashes — but if that master machine dies, you must hastily crown a stand-in, and that hand-over is where things most often go wrong.
② Several can change it, reconcile afterward (multi-leader). Each data center keeps a writable ledger; each edits locally and syncs with the others. Great for writing close to home — but if two people change the same cell at once, they conflict, like two people editing the same shared doc, and you must merge somehow.
③ No one's in charge, trust the crowd (leaderless). When you change something, tell several machines at once; when you read, ask several at once. As long as the machines you "wrote to" and the ones you "asked" overlap, you'll hit one that knows the latest — like sending an urgent notice to 3 colleagues, then asking any 2 later: you'll always reach someone in the know.
No free lunch: the harder you insist every copy be identical at every instant, the slower and less failure-tolerant you get; the more you chase speed and resilience, the more you must tolerate readers seeing stale data. Single-leader is simple and by far the most common (most databases default to it); multi-leader suits cross-region and offline work; leaderless is built to "keep running even if several machines are down." Choosing is really about which of fast / resilient / always-latest you want most.
Replication = copying one dataset onto many machines — for fault tolerance, closeness, and read-sharing. The hard part isn't the copying, it's propagating changes, and coping when a reader hits a copy the change hasn't reached yet. Three schemes: one-machine-decides, several-reconcile, trust-the-crowd — each trading among fast / resilient / always-latest.
Want the actual mechanisms, the quorum notation, and diagrams? → Switch to the deep read
Replication means keeping the same data on multiple machines connected by a network. It buys three things — fault tolerance (survive lost nodes), low latency (data closer to users), and read scaling (many replicas share read traffic). The hard part is never "storing copies"; it's handling the continuous changes to that data. This chapter collapses the ways of handling change into three architectures — single-leader, multi-leader, and leaderless — and forces you to confront the replication lag and consistency anomalies that asynchronous replication brings.
n = total replicas, w = nodes a write must confirm, r = nodes a read must query.This chapter opens Part II, "Distributed Data." Part I (Ch 1–4) took a single-machine view — how data is modeled, stored, retrieved, and encoded. From here on, data lives across many machines. Replication keeps whole copies in several places; the very next chapter, partitioning (Ch 6), splits the data into shards — and the two are usually combined. Replication is also the runway for Ch 7 (transactions) and Ch 9 (consistency & consensus): the "how consistent can we actually be" question this chapter plants is exactly what those later chapters answer head-on.
Say you run a read-heavy service: 50k reads/sec, 5k writes/sec, users across the US, Europe, and Asia. With a single database, three problems hit at once: that one node crashes and the whole site is down (availability); European users pay 100–200 ms per query crossing the ocean (latency); 50k QPS of reads saturates one machine's CPU/IO (throughput). Replicate the data to several machines in several regions and all three ease at once: lose a node and a standby covers, read locally to save round-trips, and multiple replicas share the reads.
But replication isn't "copy a file once" — the data keeps changing. The core difficulty: how a write propagates to every replica, what to do while replicas are briefly or lastingly out of sync, and how the system keeps working correctly when a replica or the leader dies. Ignore these and "multiple copies" gives you not reliability but harder-to-debug inconsistency. This chapter pulls that difficulty apart and gives three families of answers, each with its price.
What it is: designate one replica the leader; all writes go only to it. The leader writes each change locally and, in order, sends a replication log to the followers, which replay it to catch up. Reads can go to the leader or any follower. PostgreSQL, MySQL, MongoDB, Oracle Data Guard — and Kafka, RabbitMQ mirrored queues — default to this model.
Synchronous or asynchronous? (this model's first trade-off) After sending a write, does the leader wait for the follower's ack before replying? Synchronous guarantees the follower has the latest copy, but if that follower is slow or down, the write stalls. Asynchronous doesn't wait — high throughput — but if the leader crashes in the instant "already replied to the client, change not yet on any follower," those writes are lost forever. In practice you almost never make all followers synchronous (one stall halts everything); the common choice is semi-synchronous: keep exactly one follower synchronous and the rest async — you get "at least two up-to-date copies" without being held hostage by a single node.
Adding and recovering followers: bring up a new follower from "a snapshot of the leader at some moment + replay the log from that snapshot's position onward"; a follower that crashed and reconnects simply catches up from the log position it last recorded.
When the leader dies: failover — the most dangerous step. The flow: detect the leader is really gone (usually via timeout) → choose a new leader (election, or appointment by a controller — typically the most up-to-date follower) → reconfigure the system to it. It's riddled with traps: under async replication, the new leader may be missing the old leader's last few writes, which are often simply discarded when the old leader returns; if those auto-increment keys / external systems (e.g. caches) were already used elsewhere, data corruption follows; split brain (two nodes both claiming leadership) can occur. A too-short timeout causes spurious failovers on transient blips; too long, and recovery drags. "When to fail over automatically vs. involve a human" still has no settled answer.
Table 1 · Four ways to implement the replication log (how the leader describes changes to followers)
| Method | How | Pitfall / cost |
|---|---|---|
| Statement-based | Ship the raw INSERT/UPDATE SQL to followers to replay | Nondeterministic NOW()/RAND(), auto-increment, trigger side effects differ per node → divergence |
| WAL shipping | Ship the write-ahead log (byte-level storage-engine changes) | Tightly coupled to the engine version — mismatched leader/follower versions can't ship, so rolling upgrades are hard |
| Logical (row) log | Describe changes at the row level: which row, old/new values | Decoupled from storage internals — cross-version, and feedable to external systems (the basis of CDC, change data capture) |
| Trigger-based | Use DB triggers to write changes into a table the app ferries | Most flexible, but higher overhead and more bug-prone |
What it is: more than one node accepts writes — each leader is simultaneously a follower to the others. Single-leader is fine inside one data center; multi-leader shows up once you need multi-datacenter operation (a leader per data center, write locally, sync between centers asynchronously), offline writes (a phone calendar app edits offline and merges on reconnect — each device is like a tiny data center), or collaborative editing (many people editing one document at once).
Intuition and mechanism: the upside is low write latency at each leader and continued writes when one data center dies entirely. But it brings the headache single-leader never has — write conflicts: two leaders concurrently change the same record, and the values don't line up when synced.
How do conflicts converge? A few paths: avoid conflicts — route all writes for a given record to the same leader (simplest, but breaks if that leader must move); last-write-wins (LWW) — stamp each write with a timestamp / ID and keep the largest — easy but silently drops data; merge — e.g. concatenate both versions (collaborative editing does this); keep the conflict and let the application or user decide. Replication topology matters too: all-to-all is most robust but messages can arrive out of order and break causality, needing version vectors to reconstruct "who came first"; circular / star topologies block if a single link breaks.
What it is: drop the leader entirely — the client (or a coordinator node) sends each write to several replicas at once and reads from several at once. Amazon's Dynamo popularized this line; Cassandra, Riak, and Voldemort follow it.
The quorum mechanism — the one bit of math, in one line: with n replicas total, a write needs w acks to count as successful and a read queries r. As long as w + r > n, the r nodes you read from overlap the w nodes you wrote to by at least one node — that node holds the latest value, and the reader picks the newest by version number. In plain terms: "who wrote" and "who you asked" are guaranteed to intersect, so you always reach someone in the know. A typical config is n=3, w=2, r=2: any one replica can be down and both reads and writes still assemble a quorum, so the system keeps running.
Repairing lagging replicas: read repair — the client reads a stale value from a replica and writes the fresh value back to it; anti-entropy — a background process continuously compares replicas and fills in missing data. Sloppy quorums + hinted handoff: during a network partition, a write can land on reachable nodes that aren't among the "home" n and be handed back later — this raises availability, but the w+r>n "read the latest" guarantee no longer holds. In fact, even when the quorum is met, concurrent writes, partial write failures, and sloppy quorums can all let you read stale data — leaderless replication gives eventual consistency by default, not linearizability. It uses version vectors to recognize "concurrent writes," marking them as siblings to be merged later.
The moment you use asynchronous replication for read scaling (common in both single-leader and leaderless), you can't escape replication lag: reads from a follower return stale data. Lag is usually milliseconds to seconds, but can reach minutes when a follower stalls. DDIA gives three targeted weak-consistency guarantees, strengthening as you go down:
"Eventual consistency" is deliberately vague — it promises "eventually consistent" but says nothing about what you'll read along the way or how long you'll wait. These three cut it into concrete, engineering-operable, à-la-carte guarantees.
The three schemes aren't "which is more advanced" — each stands at a different trade-off point. The core tension is always the plain-English CAP line: when the network breaks, do you want "always read the latest (strong consistency)" or "stay readable/writable even when nodes are down (high availability)"? You can't have both.
Table 2 · Single-leader / multi-leader / leaderless: what to choose when
| Single-leader | Multi-leader | Leaderless | |
|---|---|---|---|
| Who can write | Only the one leader | Several leaders write concurrently | Any replica (client fans writes out) |
| Write conflicts | None by construction (writes serialize at the leader) | The core problem, needs a convergence rule | Concurrent writes, handled by version vectors / LWW |
| Leader-failure tolerance | Needs failover, a window, error-prone | One leader down, others keep writing | No single point; runs with several down |
| Read consistency | Strong at leader; eventual at followers | Eventual | Eventual by default; quorum approximates but isn't linearizable |
| Typical config | 1 leader + N followers, semi-sync keeps 1 | 1 leader per data center | n=3, w=2, r=2 |
| Best fit | Most OLTP: read-heavy, single region | Cross-region writes, offline, collaborative editing | Extreme write availability, tolerant of weak consistency (carts, metrics, logs) |
| Representative systems | PostgreSQL, MySQL, MongoDB, Kafka | BDR / Tungsten, CouchDB, calendar sync | Cassandra, Riak, DynamoDB |
Two operating rules of thumb: ① default to single-leader — it's the simplest, lowest-cognitive-load option and satisfies the vast majority of "read-heavy, single-region" workloads; only pay the multi-/leaderless complexity when low-latency cross-region writes, offline, or extreme write availability become hard requirements. ② sync/async isn't a global switch — semi-synchronous (one follower kept synchronous) is often the sweet spot between "risk of losing writes" and "write latency."
Replication is the default substrate of nearly every production database: PostgreSQL streaming replicas for read scaling, MySQL leader/follower to absorb reads, Kafka's multi-replica partitions so messages aren't lost, Cassandra's tunable consistency for cross-DC writes — each is a point picked in one of this chapter's three models. In interviews, "how does leader/follower do read-write splitting," "does async replication lose data / how do you read your own writes," and "what does Cassandra's w+r>n mean" are near-guaranteed — the answers all live in this chapter.
w+r>n) with read repair, hinted handoff, and vector clocks to keep the shopping cart "always writable" — sacrificing strong consistency for availability; this chapter's leaderless section derives directly from it.Dynamo: Amazon's Highly Available Key-value Store, SOSP 2007 ↗acks=all + min.insync.replicas, committing only once enough replicas confirm — this chapter's semi-synchronous idea landed in a production messaging system.Kafka Documentation · Replication ↗w+r>n equals strong consistency. It only guarantees the read and write sets intersect — not linearizability. Concurrent writes, partial write failures, and sloppy quorums can all return stale data. Strong consistency needs consensus (Ch 9).① Replication = one dataset on many machines, buying fault tolerance / low latency / read scaling; the hard part isn't storing copies, it's continuously syncing changes.
② Three architectures: single-leader (one writable, most common), multi-leader (many writable, cross-region / offline), leaderless (no single point, quorum gatekeeps).
③ Single-leader's first trade-off is sync vs async: async has high throughput but loses un-replicated writes on leader crash; semi-sync (keep one synchronous) is the common sweet spot.
④ Failover is the most dangerous step: lost writes, split brain, mis-triggers; "automatic vs manual" is unsettled.
⑤ The replication log has four implementations — statement / WAL / logical-row / trigger; logical-row is decoupled and is the basis of CDC.
⑥ Multi-leader's core problem is write conflicts: converge via avoid / LWW / merge / let-the-user-decide; LWW loses data.
⑦ Leaderless uses w+r>n read/write-set overlap to fetch the latest (typically n=3,w=2,r=2), but gives eventual, not linearizable, consistency; paired with read repair / anti-entropy / sloppy quorum.
⑧ Async replication always brings replication lag; treat it with read-your-writes / monotonic reads / consistent-prefix reads — the operable slices of "eventual consistency."
⑨ Selection rule: default to single-leader; go multi-/leaderless only when cross-region writes / offline / extreme write availability become hard requirements.