Part 4 · Redundancy

Replication I: Leaders, Followers, Lag, and Failover

Through lesson 07, everything lived on one node — one storage engine, one write-ahead log, indexes and caches all serving demand from a single machine. Caching pushed that machine further, but it cannot outrun three limits: the machine dies, the read traffic outgrows even a warm cache, or the users sit an ocean away. The fix is redundancy — keep copies of the data on more than one node, for availability and read scale. But the instant a second copy exists, the copies can disagree, and the gap between them is where a whole family of user-visible anomalies lives. This is the first lesson where data stops living on one node and becomes a distributed thing.

Source grounding

Original synthesis inspired by Martin Kleppmann, Designing Data-Intensive Applications, Ch.5 (Replication) — single-leader replication, replication lag, the read-after-write / monotonic-read / consistent-prefix anomalies, and leader failover. Built here, not reproduced.

Linear position

Prerequisite: Lesson 03 (the ordered, durable write-ahead log) — you know that a storage engine records every change as an append-only log before applying it. Lesson 06 (reliability) — you know a node can crash at any moment.
New capability: Stand up multiple copies of a dataset behind one leader, reason precisely about the staleness a reader can see, name and fix the three classic read anomalies, and design a failover that does not silently lose acknowledged writes or split into two leaders.

The plan

Five moves. (1) Establish why we copy data and the unavoidable cost — copies disagree across time. (2) Build single-leader replication on top of the lesson-03 log: the leader appends, followers replay, and the sync-vs-async choice sets the durability/availability trade. (3) Define replication lag and the three read anomalies it causes — read-after-write, monotonic reads, consistent-prefix — each with its fix. (4) Put a number on it: a 700 ms-lagging follower and a user who writes-then-reads, and how read-your-writes routing drives the miss rate to zero. (5) Failover — detecting a dead leader, promoting a follower, and the three hazards (lost writes, split brain, the timeout), with forward pointers to fencing (15) and consensus (17).

1 · Why copy data at all — and the bill that comes with it

What this is: the three reasons to replicate, and the one cost they all share.

A single node is a single point of failure, a single throughput ceiling, and a single location. Replication — keeping a copy of the same data on several nodes — buys back all three:

Availability. If one node crashes, another copy can keep serving. The data outlives any single machine.
Read scale. Reads can be spread across many copies, so the system serves more read traffic than one node could. (Writes are a different story — we will see why.)
Lower-latency local reads. Put a copy near the user. A reader in Singapore hits a Singapore replica instead of paying a 150 ms round-trip to Virginia.

None of this is free, and the cost is the entire subject of this lesson: the copies can disagree. Writes happen over time, and networks are slow and unreliable, so at any instant some copies have seen the latest write and others have not. Replication is therefore really a question about time: who knows what, when, and what is a reader allowed to see while a copy is still catching up? If the data never changed, replication would be trivial — copy it once, never think again. All the difficulty comes from change propagating at finite speed.

2 · Single-leader replication, built on the log

What this is: route every write through one node, ship its log to the rest.

The simplest scheme is single-leader replication (also called primary/backup or master/slave). One replica is the leader (primary); the rest are followers (read replicas, secondaries). The rule is asymmetric:

All writes go to the leader. The leader is the one authority on the current value. Clients never write to a follower.
Reads may go to the leader or any follower. This is the lever that buys read scale and local reads.

How does a write reach the followers? Through the same mechanism you met in lesson 03. When the leader applies a write, it appends that change to its replication log — an ordered, append-only record of every change, the very same idea as the write-ahead log from lesson 03, now doing double duty as the wire format between nodes. Each follower opens a connection to the leader, receives the log entries in order, and replays them against its own copy. Because the log is ordered, every follower applies the writes in exactly the leader's order; the only thing that differs between followers is how far behind they are.

What travels in that log varies. Statement-based ships the SQL text itself — compact, but breaks on nondeterministic functions like NOW() or auto-increments that differ per node. Write-ahead-log shipping ships the low-level on-disk byte changes — exact, but couples followers to the leader's storage-engine version. Logical (row-based) ships the resulting row changes (which rows, which new values) — deterministic and storage-version-independent, and the common modern default. The method differs; the ordered-log idea does not.

writes only client ───────────────▶ LEADER (the one authority) │ append to replication log (the lesson-03 log, on the wire) │ ┌─────────────┼─────────────┐ ▼ ▼ ▼ FOLLOWER A FOLLOWER B FOLLOWER C (replays the log in order; each at its own offset) ▲ ▲ ▲ └─────────────┴─────────────┘ reads may go here ◀───────── client reads

The synchronous / asynchronous choice

The one knob that matters most is when the leader tells the client the write succeeded relative to when the followers have it:

Synchronous replication. The leader waits for a follower to confirm it has the write before reporting success to the client. The acknowledged write is guaranteed to exist on at least two nodes. The cost: every write now pays the slowest follower's round-trip, and if that follower is down or slow, the write blocks — availability of writes is coupled to the follower.
Asynchronous replication. The leader reports success as soon as it has committed locally, and ships the log to followers afterward, in the background. Writes are fast and unaffected by slow followers — but the followers lag, and a write that was acknowledged can be lost entirely if the leader dies before the followers received it.

In practice almost nobody runs fully synchronous (one slow node would freeze all writes). A common middle ground is semi-synchronous: exactly one follower is synchronous (so an acknowledged write always exists on two nodes), and the rest are asynchronous (so they don't drag down write latency). If the synchronous follower is slow, an async one is promoted to take its place.

Choice	Buys	Costs
Synchronous follower	Acknowledged write survives the leader dying	Write latency = slowest follower; a slow/dead follower blocks writes
Asynchronous follower	Fast writes, tolerant of slow/dead followers	Followers lag; acknowledged write can be lost on failover
Read from followers	Read scale + low-latency local reads	Reads can be stale — the three anomalies below
Read from leader	Always fresh, simplest semantics	Leader becomes the read bottleneck; loses the local-read win

3 · Replication lag and its three anomalies

What this is: the gap between leader and follower, and the three ways users feel it.

Replication lag is the amount of time (or, equivalently, the number of log entries) by which a follower trails the leader. Under light load it is milliseconds; under heavy write load, network trouble, or a follower catching up after a restart, it can stretch to seconds or minutes. Because reads go to whichever follower the client happens to hit, lag is not an abstract internal metric — it is directly visible to users as data that is briefly wrong. This is the eventual-consistency regime: stop writing and the followers will catch up, so the system is consistent eventually, with no bound on how long "eventually" takes. Three specific anomalies recur so often they have names, and each has a known fix. This lesson is their home; lesson 16 will build the stronger guarantee (linearizability) that rules them out entirely.

Anomaly 1 — read-after-write (read-your-own-writes)

A user submits a write, then immediately reads — and does not see their own write, because the read landed on a follower that hasn't received it yet. You update your profile photo, the page reloads from a lagging replica, and the old photo is back. The user's own action appears to have failed. This is the most jarring anomaly because it violates a contract users assume without being told: I should always see the result of my own actions.

Read-your-writes consistency is the guarantee that a user always sees their own writes (it says nothing about other users' writes). The fixes:

Read your own writes from the leader. For data the user could have just modified (their own profile), route those reads to the leader. Reads of data they cannot have modified (someone else's profile) stay on followers.
Track a write timestamp / log position. Remember the log offset of the user's last write (e.g., in their session). When they read, route to a follower that has caught up past that offset, or wait until one has.
Sticky routing by recency. For a short window after a write (say, one minute), pin that user's reads to the leader or to a known-fresh replica.

Anomaly 2 — monotonic reads (going backwards in time)

A user makes two reads in a row and the second is older than the first — time appears to move backwards. The first read hits an up-to-date follower and shows a comment; the page refreshes, the second read hits a more-lagging follower, and the comment vanishes. Each read individually is a valid past state; the problem is the sequence regressing.

Monotonic reads guarantees that if a user has seen the data at some point, later reads never show an earlier state. The standard fix is sticky routing: make sure each user always reads from the same replica (e.g., hash the user ID to a replica), so they advance through that replica's timeline monotonically and never jump to a more-stale one. (Note the shard-key idea here — pinning a user to a replica — connects to partitioning in lesson 11.)

Anomaly 3 — consistent-prefix reads (effect before cause)

An observer sees writes in an order that violates causality — the answer before the question. Mr. Poons asks "how far away are you?"; Mrs. Cake replies "about ten seconds"; but a reader whose partition for Cake's reply lags less than the partition for Poons's question sees the reply first, and it reads like nonsense. This anomaly shows up specifically when different pieces of data are partitioned across nodes that lag independently (lesson 11), so there is no single ordered log tying the related writes together.

Consistent-prefix reads guarantees that if a sequence of writes happens in a certain order, anyone reading them sees them in that order (no gaps that reorder cause and effect). The fix is to ensure causally related writes go through the same ordered log / partition, so their relative order is preserved — which, deep down, is why total ordering and consensus (lesson 17) matter.

Why "just make everything synchronous" is the wrong reflex

You could eliminate all three anomalies by reading only from the leader, or by replicating fully synchronously. But that throws away the read scale and local-latency wins that motivated replication in the first place, and couples write availability to your slowest node. The engineering move is not to pick one global setting; it is to choose the guarantee per operation: read-your-writes only for data the user just touched, monotonic reads via sticky routing, leader reads only for the few invariants that truly cannot tolerate staleness.

4 · Worked number — how often does a read miss its own write?

Make the read-after-write anomaly concrete. Suppose your followers lag the leader by 700 ms on average, and a user writes, then immediately fires a read that is routed to a random replica. Take a simple, deliberately conservative model: you have one leader and three asynchronous followers (four replicas total), the user's read lands on a uniformly random one, and "immediately" means the follower has not yet received the write if the read arrives within the lag window.

t=0ms user writes → leader has it, value = NEW t=0ms user reads → routed to a random replica of {leader, fA, fB, fC} leader : has NEW now → fresh fA,fB,fC: each receives NEW ~700ms later → still has OLD at t=0 → STALE P(read is stale) = P(landed on a follower) × P(that follower hasn't caught up yet) = (3/4) × (read arrives inside the ~700ms lag window ≈ 1) ≈ 0.75

So in this naive setup, roughly 3 out of 4 immediate self-reads miss the user's own write — because three of the four replicas are followers, and at the instant of an immediate read essentially none of them has the 700 ms-old write yet. Even if you only spread reads across the three followers (never the leader), the miss rate is ~100% for that first immediate read. This is not a rare edge case; it is the default behavior of naive follower reads, and it is exactly the "ghost write" users report.

Now apply read-your-writes routing: for the short window after a user's write, send that user's reads to the leader (or to a replica known to have passed the write's log offset). The leader always has the write, so the miss rate for self-reads drops from ~75% to 0% — while every other user's reads keep going to followers and keep the read-scale benefit. You paid for read-your-writes only on the tiny slice of traffic that needs it: a user reading data they themselves just changed, within a one-minute window. The widget below lets you feel how lag and the number of async followers move the stale-read probability.

Replication lag vs. stale reads

A user writes at t=0, then reads Δt milliseconds later from a random replica (1 leader + f async followers). Each follower receives the write after a random delay centered on the average lag. The bars show, over many trials, the fraction of those immediate self-reads that are stale (miss the write). Then flip on read-your-writes routing and watch the stale fraction collapse — because the read is pinned to the leader, which always has the write.

average lag: 700 ms async followers f: 3 read after write Δt: 0 ms read-your-writes routing

Replicas (leader + f)

—

Stale self-reads

—

Routing

—

5 · Failover — where the hidden assumptions surface

What this is: replacing a dead leader, and the three ways that goes wrong.

Everything above assumed the leader stays up. It won't. Failover is the process of detecting that the leader has died and promoting a follower to become the new leader. It is deceptively hard, and almost every step has a way to go wrong.

1Detect. Decide the leader is dead. There is no clean signal — only silence. Nodes send heartbeats; if the leader misses them for a timeout, it is presumed dead. The timeout is a guess (see hazard below).

2Choose. Elect a new leader — ideally the follower with the most up-to-date data (the largest log offset), to minimize lost writes. This is itself a distributed agreement problem (consensus, lesson 17).

3Reconfigure. Point clients and the other followers at the new leader, and ensure the old leader, if it returns, steps down to a follower rather than continuing to accept writes.

Failure modes

Lost async writes. If replication was asynchronous, the promoted follower may be missing writes the old leader had acknowledged. Those writes are gone. If the old leader rejoins and its log conflicts with new writes on the same keys, those writes are usually discarded — quietly violating the durability the client was promised.
Split brain. The old leader didn't actually die (it was just slow or network-partitioned), but a new leader was promoted. Now two nodes believe they are leader and both accept writes. With no mechanism to stop one, their data diverges and may corrupt. This is the central hazard.
The timeout dilemma. Too short a failover timeout and a brief network blip or GC pause triggers a needless failover (and possibly a cascade as load shifts). Too long and the system is unavailable for writes while everyone waits. There is no universally right value — it is a tunable trade between false failovers and downtime.

Decision checklist

What is the contract: can an acknowledged write be lost? If no, you need at least one synchronous follower.
What is the maximum stale-read window you will expose, and which operations get read-your-writes or leader reads instead?
How is a dead leader detected, and what is the heartbeat timeout — chosen against your real network jitter and GC-pause distribution?
What prevents two leaders? (Forward pointer: a monotonically increasing fencing token that the storage layer checks, lesson 15; leadership decided by consensus, lesson 17.)
Is leader election automatic or operator-driven? Automatic is faster but more prone to false failovers.
Do clients re-discover the leader after failover, or will they keep writing to a demoted node?

The two hardest hazards — split brain and lost writes — are not fully solved here; they are solved by machinery later in Part II. Fencing tokens (lesson 15) let the storage layer reject writes from a leader that has been superseded, even if that old leader still thinks it is in charge. Consensus / total-order broadcast (lesson 17) is how a group of nodes agrees on exactly one leader and one order of writes despite failures. For now, the takeaway is to name the contract rather than hope the defaults are safe.

Checkpoint exercise

Try it

You run a feature store for an ML serving system. Feature values are written by an ingestion job (the writer) and read at inference time by many model servers (the readers), which are spread across three regions. (a) Pick single-leader with regional read replicas, and state the sync/async choice and why. (b) A data scientist updates a feature definition and immediately queries it to verify — which anomaly threatens this, and exactly how do you give them read-your-writes without sending all serving traffic to the leader? (c) The leader region goes down. Write the failover contract: can an acknowledged feature write be lost, what is your detection timeout, and what stops a recovered old leader from corrupting the registry? Tie part (c) to the fencing/consensus pointers above.

Where this points next

Single-leader replication is the easy case: one node decides the order of all writes, so there is never a conflict — at most some followers are behind. But that one leader is also the write bottleneck and the single point that must fail over. Lesson 09 relaxes the constraint and lets more than one node accept writes — multi-leader (a leader per region) and leaderless (any replica, with quorum reads and writes where w + r > n). The instant two nodes can write the same key concurrently, the comfortable single-order world is gone and conflict resolution becomes part of the application's contract. Everything you learned here about lag and staleness still applies; what changes is that "who is right?" no longer has an obvious answer.

Where is truth?

For single-leader replication, the system of record is the leader — the one node that decides the order of writes and holds the authoritative current value. The copies / derived views are the followers, each replaying the leader's replication log at its own offset. The freshness budget is the replication-lag budget — the maximum staleness a follower read may expose (and the read-your-writes window for data a user just touched). The owner is whoever runs the failover policy and the heartbeat/timeout config. The deletion path is a delete written to the leader, which propagates down the same log to every follower. The reconciliation/repair path is a lagging or restarted follower re-syncing by replaying the leader's log from its last offset. The evidence it is correct is per-follower lag metrics, checksum/anti-entropy comparison against the leader, and a fencing token proving no superseded leader is still accepting writes.

Takeaway

Replication keeps copies of data to buy availability, read scale, and local-read latency — and the unavoidable price is that copies disagree across time. Single-leader replication routes all writes through one leader, which appends them to a replication log (the same ordered, durable log from lesson 03) that followers replay in order; the sync/async choice trades write latency and availability against the risk of losing an acknowledged write. The gap between leader and follower is replication lag, and it surfaces as three named anomalies — read-after-write (fixed by reading your own writes from the leader or tracking your write's log offset), monotonic reads (fixed by sticky per-user routing so time never runs backwards), and consistent-prefix reads (fixed by keeping causally related writes in one ordered partition). A naive immediate self-read against a 700 ms-lagging fleet misses its own write ~75% of the time; read-your-writes routing drives that to zero on exactly the traffic that needs it. Failover — detect, choose, reconfigure — risks lost async writes and split brain, and its timeout trades false failovers against downtime; the deeper fixes are fencing tokens (15) and consensus (17).

Interview prompts

Why replicate data, and what does it cost? (§1 — availability, read scale, local-read latency; the cost is that copies disagree across time, so a reader may see stale data.)
What is the replication log and how does it relate to the WAL? (§2 — it is the same ordered, append-only change log from lesson 03; the leader appends each write and followers replay it in order, so all followers apply writes in the leader's order and differ only in how far behind they are.)
Synchronous vs. asynchronous replication — the trade? (§2 — sync guarantees an acknowledged write survives the leader dying but couples write latency/availability to the slowest follower; async is fast and tolerant of slow followers but lags and can lose acknowledged writes on failover. Semi-sync keeps one sync follower.)
Define the three replication-lag anomalies and one fix each. (§3 — read-after-write: see read from leader / track write offset; monotonic reads (time going backwards): sticky per-user routing; consistent-prefix (effect before cause): keep causally related writes in one ordered partition.)
A user writes then immediately reads from a random replica with 700 ms lag — what fraction miss their write, and how do you fix it? (§4 — ~75% with one leader + three async followers, since the read usually lands on a follower that hasn't received the write yet; read-your-writes routing pins that user's reads to the leader and drops the miss rate to 0%.)
Walk through failover and its hazards. (§5 — detect (heartbeat timeout), choose (most up-to-date follower, via consensus), reconfigure (redirect clients, demote old leader); hazards are lost async writes, split brain (two leaders), and the timeout trade between false failovers and downtime.)
What is split brain and what prevents it? (§5 — two nodes both believing they are leader and both accepting writes, diverging the data; prevented by fencing tokens that the storage layer enforces (lesson 15) and by deciding leadership through consensus (lesson 17).)