Replication

1. Why replicate?

Replication means that you keep a copy of the same data on several machines connected by a network. Each machine that holds a copy is a replica. There are three reasons to do this:

In this chapter, each machine can hold the whole data set. Chapter 6 covers partitioning (sharding), which splits data that is too big for one machine. Here is the key point: if the data never changes, replication is easy. You copy it once. All the difficulty comes from changes. Every write must reach every replica, and it must reach them in a sensible order.

Three algorithms do this, and almost all databases use one of them:

2. Leaders and followers

The most common method is leader-based replication. Also called master–slave, or active/passive.

1. One replica is the leader (or primary). All writes go to the leader. The leader writes the new data to its own storage first.
2. The other replicas are followers (read replicas, secondaries, hot standbys). The leader sends each change to its followers in a replication log. A follower applies the changes in the same order as the leader.
3. A client can read from the leader or from any follower. But writes are accepted only by the leader.

Many databases use this: PostgreSQL (since version 9.0), MySQL, Oracle Data Guard, and SQL Server AlwaysOn. Some NoSQL systems use it too, such as MongoDB and RethinkDB. Message brokers such as Kafka and RabbitMQ (in highly available queues) use it. So do network filesystems and DRBD.

Synchronous or asynchronous replication?

Here is a website example. A user changes their profile picture. The client sends the change to the leader. The leader sends it to the followers. Then the leader tells the client that the update worked. The question is: does the leader wait for the followers before it says "OK"?

Here is the trade-off:

• Synchronous: the follower has an up-to-date copy. If the leader dies, the write is safe. But if the synchronous follower is slow or down, all writes wait, or fail. One node outage stops the whole system.
• Asynchronous: the leader keeps working even if all followers fall behind. But a write that the client was told about can be lost if the leader dies before it is copied. So the write is not durable, even after a success message.

In practice, one follower is often synchronous, and the others are asynchronous. If the synchronous follower slows down, an asynchronous one takes its place. This gives an up-to-date copy on at least two nodes. People call this semi-synchronous.

Setting up a new follower

You cannot copy the data files while the database is changing. The files would show different parts of the data at different times. So you do this, without downtime:

1. Take a consistent snapshot of the leader. Most databases can do this for backups, without locking everything.
2. Copy the snapshot to the new follower.
3. The follower asks the leader for every change since the snapshot. The snapshot must be tied to an exact position in the replication log. PostgreSQL calls this the log sequence number. MySQL calls it the binlog coordinates.
4. When it has processed the backlog, it has caught up. Then it follows the live stream.

Node outages and failover

Nodes fail, and they also go down for planned work, such as a security patch. The goal is that the system keeps working, and the impact stays small.

Follower failure: catch-up recovery

Each follower keeps a log of the changes it received. After a crash or network break, it knows the last change that it applied. It asks the leader for everything after that point. Then it is caught up again.

Leader failure: failover

If the leader fails, one follower must become the new leader. Clients must send their writes to the new leader. The other followers must follow the new leader. This is failover. It can be manual, or automatic. An automatic failover does these three steps:

1. Detect that the leader failed. There is no perfect way. Most systems use a timeout. If a node does not answer for, say, 30 seconds, it is assumed dead.
2. Choose a new leader. By an election, or by a controller node. The best choice is the replica with the most recent data, to lose as little as possible. Agreement is a consensus problem (Chapter 9).
3. Reconfigure the system. Clients use the new leader. The old leader, if it comes back, must accept that it is now a follower.

Failover can go wrong in four ways, and each one is a real risk:

• Lost writes. With asynchronous replication, the new leader may not have all the writes of the old one. What happens when the old leader comes back? The usual answer is to discard its unreplicated writes. That can break users' expectations about durability.
• Other systems get out of step. At GitHub, an out-of-date MySQL follower became leader. Its auto-increment counter was behind, so it reused some primary keys. Those keys were also used in a Redis store. The result: some private data was shown to the wrong users.
• Split brain. Two nodes both believe they are the leader. If both accept writes, data is lost or corrupted. Some systems shut one node down when they see two leaders. If this is not designed well, both nodes can shut down.
• The timeout is hard to choose. A long timeout means a slow recovery. A short timeout causes needless failovers. A load spike can then make things worse. For this reason, many operations teams prefer manual failover.

Real example

How replication logs work

How does the leader send its changes? There are several methods.

-- The leader logs the SQL text of each write:
INSERT INTO orders (id, created) VALUES (42, NOW());
UPDATE accounts SET balance = balance - 10 WHERE id = 7;

-- Problems:
--   NOW() and RAND() give different values on each replica.
--   "UPDATE ... WHERE" depends on the data already there.
--   Triggers and stored procedures can have side effects.
-- Fix: the leader replaces NOW() with a fixed value before it logs.

-- The leader sends the bytes of its storage log (Chapter 3):
--   "write these bytes to disk block 1830 of file 16384"
--
-- Followers build the same data structures byte for byte.
-- Works well (PostgreSQL, Oracle), but the replica must run the
-- same storage format. Upgrading to a new version often needs downtime.

{ "op": "insert", "table": "orders",   "row": { "id": 42, "created": "2026-10-08T10:00Z" } }
{ "op": "update", "table": "accounts", "key": { "id": 7 }, "set": { "balance": 90 } }
{ "op": "commit", "txid": 9981 }

-- Independent of the storage format. Old and new versions can talk.
-- Easy for other programs to read (change data capture, Chapter 11).

CREATE TRIGGER orders_audit
AFTER INSERT OR UPDATE ON orders
FOR EACH ROW
EXECUTE FUNCTION write_to_change_table();   -- the app's own copy logic

-- More flexible: copy only some rows, or to another database type.
-- But more overhead, and more bugs, than built-in replication.

Each method has a trade-off. Statement-based replication is compact, but breaks easily. WAL shipping is precise, but it ties the replicas to one storage format. Logical logs are the most flexible, because the format is separate from the storage. Trigger-based replication gives the most control, and it is the most work.

3. Problems with replication lag

Asynchronous followers are behind the leader by a small time. This is the replication lag. Normally it is under a second. Under heavy load or network trouble it can grow to minutes.

A follower shows old data while it is behind. If you run the same query on the leader and on a follower, you can get different answers. This is only temporary: when writes stop, the followers catch up. Databases call this eventual consistency. The word "eventually" is vague on purpose. There is no upper limit on the lag.

Many developers did not expect this. They built apps as if the database were one machine. The next three guarantees describe the most common surprises.

Reading your own writes

A user submits a comment, then reloads the page. If the reload reads from a lagging follower, the comment is missing. The user thinks the data was lost. This is a read-after-write problem. You need read-your-writes consistency: a user always sees their own updates. Other users may still see old data for a while.

Ways to get it:

• Read from the leader for data that the user can change. A user's own profile is an example: only the owner edits it.
• Read from the leader for a while after an update. For example, one minute.
• Track the time or log position of the user's last write. Serve reads only from replicas that have reached that position, or wait for them.
• Across devices, the last-write time must be stored centrally. And the phone and the laptop may route to different data centers.

Monotonic reads

A user refreshes a page. The first request goes to a fresh follower and shows a new comment. The second goes to a lagging follower and the comment has gone. Time seems to go backward. Monotonic reads is the guarantee that this does not happen: once you have read a value, later reads do not return an older value. It is weaker than strong consistency, but stronger than plain eventual consistency.

A simple way: send each user to the same replica every time, for example by hashing the user ID. If that replica fails, the user moves to another replica.

Consistent prefix reads

Mr. Poons asks: "How far into the future can you see, Mrs. Cake?" Mrs. Cake answers: "About ten seconds usually." These sentences have a cause and effect. Now imagine an observer who reads Mrs. Cake's answer from a fast partition, and Mr. Poons's question from a slow one. The observer sees the answer before the question. This is a causality violation.

Consistent prefix reads guarantees that if a sequence of writes happened in one order, readers see them in that order. A single partition does this automatically. Problems appear in partitioned databases, where different partitions lag by different amounts. Fixes: put causally related writes in the same partition, or track dependencies explicitly.

// Read-your-writes: the author always reads their own recent posts from the leader.
async function readComments(postId: string, viewer: User) {
  const fromLeader = viewer.lastWriteAt && Date.now() - viewer.lastWriteAt < 60_000;
  const db = fromLeader ? leader : pickReplica(viewer.id);   // pinned replica: monotonic reads
  return db.query("SELECT * FROM comments WHERE post_id = $1 ORDER BY created_at", [postId]);
}

4. Multi-leader replication

One leader is a bottleneck. All writes must go through it. If you cannot reach the leader, you cannot write. The next idea: let more than one node accept writes. Each leader sends its changes to the others. Each leader is also a follower of the others. This is multi-leader replication, also called master–master or active/active.

Use cases

Inside one data center, multi-leader rarely pays off, because it adds complexity. It makes sense in three cases:

• Several data centers. Each data center has a leader. Writes are local and fast. Changes go to the other data centers in the background. A data center can fail, and the others keep working. A slow or broken link between data centers does not block writes.
• Offline operation. A calendar on your phone, laptop, and tablet is a leader on each device. You can add a meeting with no network. When you connect, the devices sync. Lag can be hours or days. CouchDB is built for this.
• Collaborative editing. Google Docs and Etherpad let many people edit at once. You can lock the document (like single-leader with transactions), or make the unit of change very small (a keystroke) and resolve conflicts afterward.

The downside is serious. The same data can be changed in two places at the same time. Those changes conflict. Multi-leader replication is often called dangerous territory. Auto-increment keys, triggers, and constraints can behave badly when several leaders are involved. Most teams avoid it unless they need it.

Handling write conflicts

Two users edit a wiki page title at the same time. User 1 changes it from "A" to "B" on leader 1. User 2 changes it from "A" to "C" on leader 2. Both writes succeed locally. The conflict appears only when the changes are exchanged. With one leader, the second writer would wait or be rejected. With several leaders, both succeed. You learn about the conflict later, when it may be too late to ask the user.

Avoid conflicts first

The simplest strategy: make sure all writes for one record go to the same leader. For example, route each user's requests to their "home" data center. Then, for that user, the system acts as single-leader. This breaks when the leader must change, for example after a data-center failure.

Converge to one value

There are four common ways to reach one final value:

Most multi-leader tools let you write the rule in code. It can run on write (in the background, fast, cannot ask the user) or on read (all versions go to the app, which decides, then writes the result back). Conflicts are resolved for one row or document at a time. A transaction that changes several rows is not one unit for conflict resolution.

Replication topologies

A topology is the path that writes take from one node to another. With two leaders, there is only one: each sends to the other. With more leaders, there are several choices.

• Circular (MySQL's default): a write may pass through several nodes. Each write is tagged with the IDs of the nodes it passed through. A node ignores a write that carries its own ID, so there are no endless loops. But if one node fails, the circle breaks.
• Star: simple, but the root is a single point of failure. It can be generalized to a tree.
• All-to-all: the most fault-tolerant, because messages can take different paths. But some links are faster, so writes can overtake each other.

Overtaking causes a real problem. Client A inserts a row on leader 1. Client B updates that row on leader 3. Leader 2 may receive the update first. From its view, the update is for a row that does not exist. Timestamps do not solve this, because clocks are not perfectly in sync (Chapter 8). The book's answer is version vectors, shown below.

5. Leaderless replication

The last approach drops the leader completely. The client sends a write to several replicas at once. Or a coordinator node does it for the client. No node decides the order of the writes. This style is called Dynamo-style, after Amazon's Dynamo system. Riak, Cassandra, and Voldemort use it.

Writing when a node is down

Imagine three replicas, and one is rebooting. There is no leader, so there is no failover. The client writes to all three in parallel. Two accept. The third misses the write. If the client needs only two acknowledgments, the write succeeds. The client ignores the missed one.

When the third node comes back, it has stale data. Reads may return it. So a read also goes to several nodes in parallel, and the client uses version numbers to pick the newest value. Two mechanisms bring the stale node up to date:

• Read repair. When a read finds a stale replica, the client writes the newer value back to it. Works well for data that is read often.
• Anti-entropy. A background process compares replicas and copies missing data. It does not follow any order, so it can take a long time. Values that are rarely read can stay stale for a long time.

Quorums for reading and writing

How many replicas must agree? With n replicas:

• A write needs w acknowledgments.
• A read asks r replicas.

If w + r > n, then the set of nodes that got the write and the set you read must overlap in at least one node. That node has the latest value. This is the quorum condition. A common choice is n = 3 with w = r = 2. Another is n = 5 with w = r = 3.

The quorum condition also tells you how many failures you survive:

• With n = 3, w = 2, r = 2: one node can be down. Reads and writes still work.
• With n = 5, w = 3, r = 3: two nodes can be down.
• With w = n and r = 1: reads are fast, but one failed node stops all writes.

Limits of quorums and sloppy quorums

Quorums look like a guarantee. They are not one. Even with w + r > n, you can read stale values:

• Concurrent writes: it is not clear which came first. Only merging is safe. A timestamp (LWW) can lose writes because of clock skew.
• A write runs at the same time as a read. The read may see the new value on some replicas and not others.
• A write succeeded on fewer than w nodes, and it was reported as failed. It is not rolled back. A later read may return it.
• A node with the new value fails, and is restored from an old copy. The count of new copies can fall below w.

Most quorum systems therefore do not give you read-your-writes, monotonic reads, or consistent prefix reads by default. Stronger guarantees need transactions or consensus (Chapters 7 and 9).

Now the sloppy quorum. Suppose the home nodes for a key are unreachable. You can (a) return an error, or (b) accept the write on any reachable nodes, even ones that are not home nodes. Option (b) is a sloppy quorum. When the home nodes come back, the temporary copies are sent to them. This is hinted handoff.

A sloppy quorum increases write availability: as long as any w nodes are up, you can write. But it breaks the overlap. The latest value may sit outside the home nodes, so a read of r home nodes can miss it. A sloppy quorum is only a promise that the data is stored on w nodes somewhere. Riak enables it by default. Cassandra and Voldemort disable it.

Multi-datacenter operation with leaderless replication

Cassandra and Voldemort count the replicas in all data centers in n. The client waits only for a quorum in its local data center. Writes to the other data centers happen in the background. Riak keeps client traffic in one data center, and copies between clusters asynchronously.

Detecting concurrent writes

Two clients write the same key at the same time, to a three-node store. Client A's write reaches node 1 first, and node 2 gets A then B, while node 3 gets B then A. Each node sees a different order, and nobody has a global clock. If each node just keeps the last value it received, the nodes stay different forever.

The common fix is last write wins: attach a timestamp and keep the biggest. This is simple, and Cassandra uses it. But the writes are concurrent*, so neither one happened "first". LWW silently drops one of two successful writes. It can also drop writes that are not concurrent, when clocks are skewed. LWW is fine when losing data is acceptable, for example in a cache. Another safe trick: write each key only once, and never update it. Cassandra users often use a UUID as the key.

type VV = Record<string, number>;   // replica ID to counter

// Returns "before", "after", "equal", or "concurrent".
function compare(x: VV, y: VV): string {
  const ids = new Set([...Object.keys(x), ...Object.keys(y)]);
  let xLess = false, yLess = false;
  for (const id of ids) {
    const a = x[id] ?? 0, b = y[id] ?? 0;
    if (a < b) xLess = true;
    if (b < a) yLess = true;
  }
  if (xLess && yLess) return "concurrent";   // neither knows the other
  if (xLess) return "before";
  if (yLess) return "after";
  return "equal";
}