System Design: Replication

• Read from the leader for data the user has just changed, for example for a short window after their write.
• Track the position (log sequence number or timestamp) of the user's last write, and only read from a follower that has caught up to it.
• Route a user's requests consistently to the same replica. This does not guarantee they see their write, but it keeps their view from going backwards while that replica stays up.

Monotonic reads

A user refreshes twice. The first read hits a fresh follower and shows a new order; the second hits a stale one and the order vanishes. Time seems to move backwards. Pinning each user to one replica prevents it while that replica stays available.

Consistent prefix reads

Two writes happen in order: a question, then its answer. A reader sees the answer without the question, because the writes replicated through different paths. This matters mostly in partitioned systems, where partitions replicate independently. The fix is to send causally related writes through the same partition or use a system that tracks causality.

These guarantees are called session guarantees and are covered in System Design: CAP and Consistency.

Failover and split-brain

When the leader fails, the system must pick a new one. This is failover. It has several steps, and every step can go wrong:

1. Detect the failure. Usually by missed heartbeats (regular "I'm alive" messages). Too short a timeout causes false alarms; too long means long downtime.
2. Choose a new leader. Typically the follower with the most up-to-date data.
3. Reconfigure. Tell clients and the other followers about the new leader.
4. Handle the old leader. If it comes back, it must realise it is no longer in charge and become a follower.

The dangers:

• Lost writes. With asynchronous replication, the new leader may be missing the last few writes of the old one. If the old leader returns with them, they conflict or are discarded. Discarding acknowledged writes can do real harm: if the new leader reuses an auto-incrementing ID already handed to another system, data from two users can get mixed up.
• Split-brain. Two nodes both believe they are the leader and both accept writes. The copies diverge and merging them may be impossible. This can happen when a network partition makes the old leader look dead while it is actually alive.
• Wrong timeouts. A slow but healthy leader gets replaced during a load spike, and the failover itself adds more load.

Defences against split-brain include: requiring a majority of nodes to agree on the leader (a quorum, so two groups cannot both reach majority), using leases (a leader's authority expires unless renewed), and fencing (the storage or followers reject requests from an old leader using an increasing token). Tools such as etcd and ZooKeeper exist largely to provide this agreement reliably.

Multi-leader replication

In multi-leader (or multi-primary) replication, several nodes accept writes, and each forwards its changes to the others. The usual reason is geography: a leader in each of Asia, Europe and the Americas lets users write locally, and a regional outage does not stop the other regions. Offline-capable apps, such as a calendar syncing between phone and laptop, are another example: every device is effectively a leader.

The price is write conflicts. Two users edit the same record in different regions at the same moment, both writes are accepted locally, and the system must decide the result:

• Last write wins (LWW): keep the later timestamp. Simple, but it silently throws away the other write, and clock differences make "later" unreliable.
• Merge: combine the changes, as collaborative editors and conflict-free replicated data types (CRDTs) do.
• Application resolution: store both versions and let the application or the user decide.
• Avoid conflicts: route all writes for a record to one "home" region, which removes most conflicts by design.

Uniqueness constraints and auto-increment IDs do not work across leaders without extra care. Use multi-leader only when you truly need local writes in several places or offline operation.

Leaderless replication and quorums

In leaderless designs, such as Dynamo-style systems like Cassandra, no node is special. A client (or a coordinator node) sends each write to several replicas and reads from several replicas.

Let N be the number of replicas holding a piece of data, W the number that must acknowledge a write, and R the number that must answer a read. If R + W > N, every read set overlaps every write set in at least one node, so a read will include at least one replica that saw the latest successful write.

For example, with N = 3, W = 2 and R = 2: a write succeeds when two of three replicas confirm, a read asks two of three, and at least one node in the read has the newest value, which the client picks by version. You can tune the numbers: W = 1 and R = 3 favours fast writes, W = 3 and R = 1 favours fast reads, and each reduces availability for the expensive operation.

Two cautions keep this honest:

• Overlap alone does not give the strict guarantee of a single leader. Concurrent writes, sloppy quorums (writing to other nodes temporarily when the usual ones are unreachable), partial write failures and clock-based conflict resolution can all produce stale or lost values. Treat quorums as "better odds", not as a transaction.
• Replicas that missed writes are repaired by read repair (fixing stale copies when a read notices a mismatch) and background anti-entropy processes.

Choosing a replication design

The same choice as a decision flow:

Most teams land on the bottom right, and the diagram shows why: the other exits each require a stated need. A sensible start for most products is one leader with two or three followers in different availability zones, semi-synchronous replication, non-critical reads (reports, catalogues) served by followers, and "the thing I just changed" served by the leader or a caught-up follower. Consider multi-region writes only if latency or regulations demand them.

In practice: moving reports to read replicas, then "my order disappeared"

The situation

A B2B ordering SaaS serves 2,000 wholesale customers, who place orders through a web app and read monthly sales reports. A team of nine runs a Spring Boot application on Kubernetes against a managed PostgreSQL primary. Month-end report runs are heavy aggregate queries, and at 09:00 on the first working day the primary hits 90% CPU and order placement slows from about 200 ms to over 2 seconds (illustrative). The team adds a read replica and sends reports to it. It works, so someone flips a framework setting to send every GET to the replica.

Within a week support logs "my order disappeared" tickets. A customer places an order, is redirected to the order list, and sees nothing, so some place it again. Assume duplicate orders reach about 0.5% of orders on busy days. Replica lag is normally under a second but rises to 20 seconds or more while the report queries run.

Starting architecture

The replica that serves customers is also the one running multi-minute report queries, so reads and reports share a lag problem.

Options considered

The decision

Rollout and verification

Ship the routing as a small decision made once per request:

Release it behind a flag for 10% of sessions first. Before that, replay a recorded month-end load against staging with reports running, and check that lag alarms fire. Dashboards that prove it: replica lag (replay_lag from pg_stat_replication on the primary), the percentage of reads routed to the primary and why, report duration, primary CPU, and the rate of repeated order submissions per customer, which is the user-visible symptom. A target of zero "disappeared order" tickets over two month-ends is the real test.

The 2 a.m. failure

At month end, the page reads "Replica lag above 30 s" and soon after "Primary CPU above 90%". A long report is delaying replay on the replica (on a PostgreSQL standby, replay waits for conflicting queries up to a configured limit, which had been raised for reports), the lag guard sends reads to the primary, and the primary becomes the next bottleneck.

First three checks: on the primary, SELECT client_addr, state, replay_lag FROM pg_stat_replication;; on the replica, SELECT pid, now() - query_start AS age, left(query, 60) FROM pg_stat_activity WHERE state = 'active' ORDER BY age DESC LIMIT 5;; and the dashboard panel showing the share of reads routed to the primary. Mitigation: cancel the offending report with pg_cancel_backend(pid), confirm lag drains, and pause scheduled report generation until traffic calms. Follow-up: a statement timeout on report connections, reports on their own replica (as decided), a cap on how much fallback traffic the primary accepts, and report schedules moved away from the 09:00 order peak.

Architect's notes

• Measure lag, not just whether the replica is up. Alert on replay lag in seconds and on the oldest long-running query on each replica.
• Ask in the review: "Which reads must see the user's own write?" List them. The answer is usually shorter than people fear.
• Do not route by HTTP method alone. A GET right after a POST is the classic trap, so write the exceptions into the design.
• Separate the failover replica from the reporting replica. A standby that analysts can saturate is not a standby.
• Fallbacks need their own limits. A guard that sends all reads to the primary moves the outage rather than preventing it.
• Defer multi-region writes. Cross-region read replicas answer most locality needs at a fraction of the conflict risk.

Common mistakes

• Treating replicas as backups. Deletes and corruption replicate instantly. Keep independent, point-in-time backups and test restores.
• Reading from followers without handling lag. Users see their own changes vanish. Use read-your-writes routing or read from the leader for fresh data.
• Assuming asynchronous replication loses nothing. Acknowledged writes can disappear on failover. Decide your acceptable loss (your RPO) and choose semi-synchronous replication if it is near zero.
• Never rehearsing failover. Untested failover breaks in surprising ways. Run drills and time them.
• Replicating but putting all copies in one zone. One power or network failure still takes everything down. Spread nodes across availability zones.
• Adopting multi-leader to "scale writes". It adds conflicts without increasing total write capacity for a single record. If writes outgrow a node, partition the data instead.
• Using last-write-wins casually. It drops data silently, and clock skew picks the wrong winner. Choose conflict handling on purpose.

Key takeaways

• Replication copies data for availability, read scale and locality. It does not add storage capacity and it is not a backup.
• Leader-follower is the default. One write node means no conflicts and a clear order of changes.
• Synchronous trades latency for durability. Asynchronous trades possible data loss for speed; semi-synchronous sits in between.
• Lag creates visible bugs. Read-your-writes, monotonic reads and consistent prefix are the guarantees users notice.
• Failover is the risky moment. Detection, election, reconfiguration and old-leader handling can each fail.
• Split-brain needs majority, leases or fencing. Never rely on timeouts alone to decide who is leader.
• Multi-leader means conflicts. Prefer routing writes to one home region unless you truly need otherwise.
• Quorums improve odds, not certainty. R + W > N creates overlap, but sloppy quorums and concurrency still allow anomalies.

Practice exercises

Exercise 1 — Starter (30 min): A leader-follower database has one leader and two followers. For each of three modes (asynchronous, semi-synchronous, fully synchronous), write down what a client sees when the leader crashes immediately after acknowledging a write, and whether that write survives. Present it as a table.

Exercise 2 — Intermediate (2 hours): Design the read path for the photo app's profile page. List the five pieces of data on the page (own bio, photo grid, follower count, other users' recent comments, notification badge). For each, decide whether it may be read from a lagging follower, must be read from the leader, or needs a caught-up follower. Explain how your application would implement read-your-writes using a stored write position.

Exercise 3 — Advanced (half day): A leaderless store uses N = 5. Compute, for W and R from 1 to 5, which combinations satisfy R + W > N, and how many node failures each tolerates for reads and for writes. Then describe one concrete scenario (sloppy quorum or concurrent writes) where even a combination satisfying R + W > N returns a stale value, and propose a detection or repair mechanism.

Related reading

• System Design: Reliability Basics — redundancy, failover and backups from the Foundations course.
• System Design: Sharding — how to split data when one leader cannot hold or write it all; the next lesson.
• System Design: CAP and Consistency — the precise consistency vocabulary that replication lag makes necessary.
• System Design: Consensus and Leader Election — how a cluster safely agrees on a leader.
• PostgreSQL documentation — reference for streaming replication and failover behaviour in one widely used database.