Replication
Leader-based, multi-leader and leaderless replication, replication lag anomalies, write conflicts, quorums and version vectors
Replication keeps a copy of the same data on several machines connected by a network, to keep data close to users (latency), to keep serving when machines fail (availability), and to spread read load (throughput). If data never changed, replication would be a one-time copy. Every difficulty in this chapter comes from the fact that data does change and every replica must somehow learn about every change.
Almost every distributed database, message broker and cache implements one of three families: single-leader, multi-leader or leaderless replication. Each makes a different bet about who may accept writes, what happens when a node is slow or dead, and what a reader can be promised. Interviewers like this chapter because it exposes whether you understand the gap between "the write returned OK" and "every replica has the write".
In one minute
- Single-leader replication routes all writes through one node; followers apply the leader's replication log in order and serve reads. PostgreSQL, MySQL, MongoDB and Kafka work this way by default.
- Synchronous followers guarantee an up-to-date copy but stall writes when slow; asynchronous followers never stall but can lose acknowledged writes on leader failure. Real systems are usually semi-synchronous.
- Leader failover is the dangerous part: lost writes, split brain, and timeouts that are too short or too long. Many teams keep a human in the loop.
- Replication logs come in four flavours: statement-based, WAL shipping, logical (row-based) and trigger-based. Logical replication decouples the log from the storage engine and enables change data capture.
- Asynchronous replication creates replication lag. Three guarantees tame the worst anomalies: read-your-writes, monotonic reads and consistent prefix reads.
- Multi-leader replication suits multi-datacenter and offline-client use cases but introduces write conflicts that must be avoided, merged or resolved by a rule.
- Leaderless (Dynamo-style) systems write to many replicas and rely on quorums (w + r > n), read repair and anti-entropy. Sloppy quorums, concurrent writes and partial failures all break the "you read the latest value" intuition.
- Detecting concurrent writes needs a happens-before relation: version numbers on one replica, version vectors across replicas. Last-write-wins is simple but silently drops data.
Leaders and followers
In leader-based replication (master–slave, primary–replica, active/passive) one replica is the leader. All writes go to the leader, which applies them locally and appends them to a replication log. Each follower consumes the log and applies writes in the leader's order. Reads may go to the leader or any follower; writes only to the leader. This is built into PostgreSQL, MySQL, Oracle Data Guard, SQL Server AlwaysOn, MongoDB and RethinkDB, into brokers such as Kafka (one leader per partition) and RabbitMQ mirrored queues, and into block replicators such as DRBD.
Synchronous versus asynchronous replication
With a synchronous follower, the leader waits for that follower's confirmation before reporting success. The follower is guaranteed up to date, so a leader crash loses nothing. The cost: one slow or unreachable follower stalls every write, and followers are slow for many reasons (crash recovery, congestion, a GC pause). Making all followers synchronous is impractical.
With an asynchronous follower, the leader sends the change and moves on. Writes stay fast and the leader keeps accepting writes even if every follower is behind or down. The price is durability: if the leader fails unrecoverably, writes it acknowledged but had not replicated are lost.
The usual compromise is semi-synchronous replication: exactly one follower is synchronous, the rest asynchronous; if the synchronous follower falls behind, an asynchronous one is promoted into its role. You get two up-to-date copies without letting any single node stall the system. MySQL's semisynchronous plugin and PostgreSQL's synchronous_standby_names implement this. Fully asynchronous replication is nonetheless common, especially with many or geographically distributed followers. Chain replication (used in Microsoft Azure Storage) shows synchronous schemes can perform well, but the trade-off between durability and write latency is fundamental.
Setting up new followers
Copying files from a live leader produces an inconsistent snapshot, and locking the database defeats availability. Instead: take a consistent snapshot without locking (the same mechanism as backups, e.g. pg_basebackup, Percona XtraBackup); copy it to the new node; the follower connects and requests all changes since the snapshot's exact log position (PostgreSQL's log sequence number, MySQL's binlog coordinates or GTID); once the backlog is processed it has caught up and streams live changes.
Handling node outages
Follower failure is routine: each follower keeps the log it received on disk, so after a crash or network blip it reconnects and requests everything after its last applied transaction (catch-up recovery).
Leader failure requires failover: promote a follower, redirect clients, and make the other followers consume the new leader's log. Automatically, that means three steps: detect the failure (usually a timeout with no heartbeat for, say, 30 seconds); choose a new leader by election or via a designated controller, ideally the most up-to-date replica (a consensus problem, Chapter 9); and reconfigure, including forcing the old leader to become a follower if it returns.
Failover goes wrong in well-documented ways:
- With asynchronous replication the new leader may lack acknowledged writes, which are usually discarded when the old leader rejoins.
- Discarding writes is especially dangerous when other systems coordinate with the database. In a 2012 GitHub incident, a lagging MySQL follower was promoted and reissued primary keys the old leader had already used; because Redis also held those keys, private data was shown to the wrong users.
- Split brain: two nodes both accept writes with no conflict resolution. Some systems shoot one node (STONITH), but a bad mechanism can shoot both.
- A timeout that is too long slows recovery; one that is too short triggers spurious failovers under load, worsening a cascading failure.
Because of these hazards many operations teams prefer manual failover even when automation exists.
Implementation of replication logs
| Approach | What is shipped | Strengths | Weaknesses | Examples |
|---|---|---|---|---|
| Statement-based | The SQL text of each write | Compact and simple | NOW(), RAND() differ per replica; autoincrement and dependent statements need identical order; triggers and stored procedures may have differing side effects | MySQL before 5.1; VoltDB (requires deterministic transactions) |
| WAL shipping | The storage engine's byte-level write-ahead log | Reuses the durability mechanism; exact copy | Coupled to engine version, so leader and follower cannot run different versions and zero-downtime upgrades are hard | PostgreSQL streaming replication, Oracle |
| Logical (row-based) | Per-row records: inserted row with all values; deleted row's key; updated row's key plus new values; commit markers | Decoupled from engine internals; works across versions; parseable by external systems | More verbose than statements | MySQL row-based binlog, PostgreSQL logical decoding, Debezium |
| Trigger-based | Whatever application code in a trigger writes to a side table | Flexible: replicate subsets, to other systems, with custom conflict logic | Higher overhead, more bugs than built-in replication | Databus for Oracle, Bucardo for PostgreSQL |
A logical log is the basis of change data capture (CDC): feeding database changes to a warehouse, search index or cache (Chapter 11). MySQL can switch from statement to row format automatically when it detects nondeterminism.
Problems with replication lag
Leader-based replication scales reads by adding followers, but only with asynchronous replication, and asynchrony means followers serve stale data. If writes stop, followers eventually catch up: eventual consistency, a deliberately vague term that promises nothing about how long "eventually" is. Lag is usually sub-second but can grow to minutes under load or network trouble, and then the following anomalies become user-visible bugs.
Reading your own writes
A user posts a comment, reloads, and it is missing because the reload hit a lagging follower. Read-after-write consistency guarantees users see their own updates (and promises nothing about others'). Implementations: read anything the user may have modified from the leader (your own profile from the leader, others' from followers); track the last update time and read from the leader for a minute afterwards, while monitoring lag; let the client remember the log position of its last write and have replicas serve it only once they have reached that position; across datacenters, route the user to the leader's datacenter. Cross-device reads are harder: the last-write position must be stored centrally, and the user's devices may need pinning to one datacenter.
Monotonic reads
Refreshing a page shows a comment, refreshing again (from a more lagged follower) hides it: time moves backwards. Monotonic reads guarantees a user never sees older state than they already saw; weaker than strong consistency, stronger than eventual. The usual implementation routes each user's reads to the same replica, by hashing the user ID, with rerouting if it fails.
Consistent prefix reads
Mr Poons asks a question, Mrs Cake answers; an observer reading from a replica where the question arrived late sees the answer first. Consistent prefix reads guarantees that writes made in a certain order are read in that order. The problem is specific to partitioned databases, where partitions have no global write order. Fixes: write causally related data to the same partition, or track causal dependencies explicitly (Chapter 9).
Living with lag
Ask how the application behaves if lag reaches minutes or hours. If the answer is "badly", design in a stronger guarantee rather than pretending replication is synchronous. Implementing read-your-writes in application code is error-prone; it is better when the database provides it, which is the territory of transactions (Chapter 7) and consensus (Chapter 9). "Transactions are too expensive, eventual consistency is inevitable" is an oversimplification.
Multi-leader replication
If the single leader is unreachable, nobody can write. Multi-leader replication (master–master, active/active) lets several nodes accept writes; each leader is also a follower of the others.
Use cases
Within a single datacenter it rarely pays for its complexity. It makes sense for multi-datacenter operation (one leader per datacenter, asynchronous replication between them: local write latency, tolerance of datacenter outages and flaky inter-datacenter links; Tungsten Replicator for MySQL, BDR for PostgreSQL, GoldenGate for Oracle, CouchDB natively), for clients with offline operation (a calendar app where each device is a leader and syncs when online, essentially multi-datacenter with an extremely unreliable network), and for collaborative editing (each user's browser is a replica, with keystroke-sized changes and no locking). The downside everywhere is the same: the same data may be modified concurrently on two leaders, and retrofitted features such as autoincrement keys, triggers and integrity constraints become problematic. Many operators treat multi-leader as dangerous territory.
Handling write conflicts
Two users change the same wiki title concurrently on different leaders; both local writes succeed, and the conflict surfaces only asynchronously, when neither user can be asked. Options:
- Avoidance. Route all writes for a record to the same leader (a user's home datacenter), so the system is effectively single-leader per record. Breaks when the home changes.
- Convergence. Replicas must end up identical. Give each write a unique ID and let the highest win (last write wins, popular and lossy); let the higher-numbered replica win (also lossy); merge values (concatenate "B/C"); or record the conflict in an explicit structure for later resolution.
- Custom logic, executed on write (Bucardo runs a handler in the background, with no user interaction) or on read (CouchDB returns all versions to the application, which writes back a merge).
Resolution is per row, not per transaction, so a multi-row transaction is resolved piecewise. Automatic approaches include conflict-free replicated datatypes (CRDTs, in Riak 2.0 and collaborative apps), mergeable persistent data structures (Git-style three-way merge) and operational transformation (Etherpad, Google Docs). Even defining a conflict is subtle: two bookings of one room for the same slot on different leaders conflict although each insert succeeded locally (Chapter 7).
Multi-leader replication topologies
| Topology | Description | Risk |
|---|---|---|
| Circular | Each node forwards writes to the next | One node failure stops the flow until reconfigured |
| Star | A root node forwards to all others (generalises to a tree) | The root is a single point of failure |
| All-to-all | Every leader sends to every other | Writes may arrive out of order on different links |
In circular and star topologies writes pass through several nodes, so each write is tagged with the nodes it has visited to prevent loops. All-to-all tolerates faults better but suffers causality problems: an insert and the subsequent update of the same row may reach a third leader in the wrong order because one link was faster. Timestamps cannot fix this (clocks are not synchronised, Chapter 8); version vectors can. Conflict detection in many multi-leader products is poor; test it.
Leaderless replication
Amazon's internal Dynamo system (2007; distinct from DynamoDB) inspired Riak, Cassandra and Voldemort, so this style is Dynamo-style. Clients send each write to several replicas directly, or via a coordinator that does not enforce an order.
Writing when a node is down
With three replicas and one down, a client writes to all three and treats two OKs as success. The recovered node is stale, so reads also go to several nodes in parallel, each response carries a version number, and the client takes the newest. Two mechanisms repair stale replicas: read repair (a read that notices a stale value writes the newer one back; good for hot keys) and an anti-entropy process (a background job that compares replicas, often with Merkle trees, and copies missing data in no particular order). Without anti-entropy, rarely read keys can stay stale indefinitely.
Quorums for reading and writing
With n replicas, writes must be confirmed by w nodes and reads must query r nodes. If w + r > n, the written and read sets overlap in at least one node, so some read response carries the latest value. Common: odd n with w = r = (n + 1) / 2. The system tolerates n − w unavailable nodes for writes and n − r for reads. Requests go to all n in parallel; w and r decide how many responses to wait for. A read-heavy workload might use w = n, r = 1.
Limitations of quorum consistency
Even with w + r > n, quorums are not strong consistency:
- A sloppy quorum may write and read different node sets, so there is no overlap.
- Concurrent writes have no defined order; LWW tie-breaking can drop writes.
- A write concurrent with a read may be visible on only some replicas; the read's outcome is undefined.
- A write that reached fewer than w nodes is reported failed but not rolled back; later reads may see it.
- A node holding a new value may fail and be restored from a stale replica, dropping below w copies.
- Timing edge cases also violate the guarantee (Chapter 9).
So Dynamo-style databases are tuned for eventual consistency; w and r adjust the probability of stale reads, not the guarantee. Read-your-writes, monotonic reads and consistent prefix need transactions or consensus. Staleness monitoring is also harder than in leader-based systems, where lag is a log-position difference; with no fixed write order, only research such as probabilistically bounded staleness (PBS) estimates it.
Sloppy quorums and hinted handoff
In a large cluster a client may be cut off from a key's "home" nodes while other nodes remain reachable. The designer chooses between returning errors and accepting writes on any w reachable nodes outside the home set: a sloppy quorum. After the network heals, the temporary hosts forward the writes home (hinted handoff). This improves write availability but weakens reads: until handoff completes, a read of r home nodes may miss the value. Riak enables sloppy quorums by default; Cassandra and Voldemort disable them.
For multi-datacenter operation, Cassandra and Voldemort count all datacenters' nodes within n but wait only for a local-datacenter quorum, replicating remotely in the background; Riak keeps client traffic local and replicates across datacenters asynchronously, like multi-leader.
Detecting concurrent writes
Replicas receive concurrent writes to one key in different orders; if each simply overwrote, replicas would diverge forever. Convergence needs a rule.
Last write wins (LWW). Attach a timestamp, keep the largest, discard the rest. Replicas converge at the cost of durability: several writes reported successful, one survives, and skewed clocks can even drop a causally later write. Cassandra supports only LWW. It is safe only when each key is written once and then immutable (UUID keys).
Happens-before. Operation A happens before B if B knows about, depends on or builds upon A. Two operations are concurrent if neither knows about the other, regardless of wall-clock time. Only concurrent operations are conflicts. The book's single-replica algorithm, illustrated with a shopping cart:
- The server keeps a version number per key, incremented on every write, stored with the value.
- A client must read before it writes; the read returns all un-overwritten values plus the latest version.
- The write carries the version from the read and the client's merge of the values it saw.
- The server overwrites all values with version ≤ that number (the client has seen them) but keeps higher-versioned values as siblings, because they are concurrent.
Merging siblings. The client merges (Riak calls concurrent values siblings). For a cart a union works, except for removals: a concurrent add and remove would resurrect the item under plain union, so deletions leave a tombstone with a version. CRDTs automate such merges.
Version vectors. With one replica a single version number suffices. When several replicas accept writes, each keeps its own version number per key and tracks the numbers it has seen from every other replica. That collection is a version vector, sent to clients on read and returned on write (Riak's "causal context", implemented as dotted version vectors). It lets the database distinguish overwrites from concurrent writes even when the write goes to a different replica than the read. Vector clocks are a related but distinct structure; version vectors are the correct tool for comparing replica states.
Design trade-offs
| Dimension | Single-leader | Multi-leader | Leaderless |
|---|---|---|---|
| Write path | All writes through one node | Any leader; conflicts possible | Any w of n replicas; conflicts possible |
| Conflict handling | None needed | Asynchronous resolution per row | Siblings and version vectors, or LWW |
| Leader loss | Failover; risk of lost writes and split brain | Other leaders continue | No leader; tolerates n − w failures |
| Cross-datacenter | Cross-DC write latency | Local writes, async replication | Local quorum, async elsewhere |
| Read consistency | Strong from leader; lag anomalies from followers | Eventual, causality issues | Eventual, tunable via r and w |
| Operational complexity | Lowest | High | Medium, many knobs |
| Systems | PostgreSQL, MySQL, MongoDB, Kafka | CouchDB, BDR, Tungsten | Dynamo, Riak, Cassandra, Voldemort |
Default to single-leader. Move to multi-leader only for multiple writable datacenters or offline clients, and design to avoid conflicts. Choose leaderless when write availability under node failure outweighs strong consistency and the data has a sensible merge (counters, sets, immutable keys).
Interview angle
-
Explain synchronous, asynchronous and semi-synchronous replication. Synchronous: the leader waits for the follower's acknowledgement, guaranteeing an up-to-date copy at the cost of latency and availability. Asynchronous: the leader confirms immediately, so acknowledged writes can be lost on leader failure. Semi-synchronous keeps one follower synchronous and promotes another if it lags, giving two durable copies without any single slow node stalling writes.
-
What can go wrong during automatic failover? Unreplicated writes are lost or discarded; a stale new leader may reissue identifiers (GitHub's Redis incident); split brain if the old leader does not realise it was demoted; and a wrong timeout causes slow recovery or spurious failovers. This is why teams keep failover manual or lean on a consensus system such as Raft or ZooKeeper for election.
-
Compare WAL shipping with logical replication. WAL shipping streams the engine's byte-level log: exact and cheap but version-coupled, blocking rolling upgrades. Logical replication streams row-level changes, is version-independent and parseable by external consumers, which makes it the basis of change data capture.
-
A user posts a comment, refreshes, and it is gone. Diagnose and fix. The refresh hit a lagging follower. Provide read-your-writes: route reads of the user's own data to the leader, or let the client carry the log position of its last write and have replicas wait to reach it. For cross-device consistency, store the position centrally and pin the user to one datacenter.
-
Why is last-write-wins dangerous, and when is it acceptable? It keeps the largest timestamp and silently discards the rest, so concurrent writes that all returned success lose data, and skewed clocks can drop a causally later write. Acceptable when keys are written once and never modified, or when losing updates is explicitly fine.
-
Does w + r > n guarantee reading the latest value? No. Sloppy quorums break the overlap; concurrent writes have no winner; a failed write lingers on some nodes; a read concurrent with a write is undefined; restoring a node from a stale replica drops below w copies. Quorums tune the probability of stale reads; linearizability needs more (Chapter 9).
-
How do you detect that two writes were concurrent rather than one overwriting the other? Track causality. On one replica, a per-key version number returned on read and sent back on write lets the server overwrite only what the client has seen and keep the rest as siblings. Across replicas, use a version vector (one counter per replica); if neither vector dominates, the writes are concurrent and must be merged.
-
Design multi-datacenter replication for a global application. Single-leader in one region: simple, but every write pays cross-region latency and a region loss means failover. Multi-leader with one leader per region and users homed to a region: local writes and region-loss tolerance, but conflicts when users move. Leaderless with local quorums and asynchronous cross-region anti-entropy: high write availability, eventual consistency. Say what each promises readers and how it resolves conflicts.
Pitfalls and misconceptions
- "Replication gives me durability." Only synchronous replication does. With asynchronous followers a leader crash can lose acknowledged writes.
- "Eventual consistency is a guarantee." It only says replicas converge if writes stop. No bound on lag, no ordering promise. Design for minutes of lag.
- "Reading from a quorum is strongly consistent." The overlap argument assumes strict quorums, no concurrent writes and no partial failures, each of which fails in practice.
- "Multi-leader is single-leader with more leaders." Conflicting writes can both succeed, autoincrement and constraints break, and conflicts are resolved asynchronously per row.
- "Timestamps order events." Clock skew makes LWW drop causally later writes. Use version numbers or version vectors for causality; treat LWW as a deliberate decision to lose data.
- "Version vectors are vector clocks." Related but different; version vectors compare replica states for a key. Riak's "vector clock" is a version vector.
- "Automatic failover beats manual." Automatic failover with aggressive timeouts has caused many cascading outages. Automatic detection with manual promotion is a respectable choice.
- "Read repair keeps replicas in sync." It fixes only values that are read. Without anti-entropy, rarely read keys stay stale and durability drops.
Key terms
| Term | Meaning |
|---|---|
| Leader (primary) | The one replica that accepts writes in leader-based replication |
| Follower (replica, standby) | A replica that applies the leader's log and serves reads |
| Semi-synchronous replication | One synchronous follower, the rest asynchronous |
| Replication lag | Delay between a write on the leader and its visibility on a follower |
| Failover | Promoting a follower to leader after leader failure |
| Split brain | Two nodes simultaneously acting as leader |
| WAL shipping | Replicating the storage engine's write-ahead log bytes |
| Logical (row-based) replication | Replicating row-level change records independent of storage format |
| Change data capture (CDC) | Consuming a database's change log from an external system |
| Eventual consistency | Replicas converge if writes stop; no bound on when |
| Read-your-writes | A user always sees their own writes |
| Monotonic reads | A user never sees data older than what they already saw |
| Consistent prefix reads | Causally ordered writes are observed in order |
| CRDT | A data structure whose concurrent updates merge automatically |
| Quorum (n, w, r) | Write to w and read from r of n replicas; w + r > n ensures overlap |
| Sloppy quorum | Accepting w writes on nodes outside the key's home set during a partition |
| Hinted handoff | Forwarding sloppy-quorum writes to their home nodes after recovery |
| Anti-entropy | Background process comparing and repairing replicas |
| Happens-before | Causal ordering: B knows about or depends on A |
| Version vector | One version counter per replica per key, used to detect concurrent writes |
Further reading
- DDIA, Chapter 5, "Replication". Reference list: https://github.com/ept/ddia-references
- DeCandia et al., "Dynamo: Amazon's Highly Available Key-value Store", SOSP 2007: https://www.allthingsdistributed.com/files/amazon-dynamo-sosp2007.pdf
- Lamport, "Time, Clocks, and the Ordering of Events in a Distributed System", CACM 1978: https://lamport.azurewebsites.net/pubs/time-clocks.pdf
- Preguiça, Baquero et al., "Dotted Version Vectors: Logical Clocks for Optimistic Replication": https://arxiv.org/abs/1011.5808
- Parker et al., "Detection of Mutual Inconsistency in Distributed Systems", IEEE TSE 1983 (the original version-vector paper)
- Bailis et al., "Probabilistically Bounded Staleness for Practical Partial Quorums", VLDB 2012
- Shapiro et al., "A Comprehensive Study of Convergent and Commutative Replicated Data Types", INRIA 2011
- PostgreSQL documentation, "Logical Replication": https://www.postgresql.org/docs/current/logical-replication.html
- Kleppmann, "A Critique of the CAP Theorem": https://arxiv.org/abs/1509.05393