Distributed Systems, Part 1: Replication

Replication means keeping a copy of the same data on more than one machine. Each machine or database process is a node, and a node that holds a copy is a replica.

Copying data that never changes is easy. The hard part is getting every change to every replica, in the same order, while nodes crash and the network drops messages. Databases solve this in three ways. Single-leader, multi-leader and leaderless replication each make a different trade, and this post walks through all three.

Every section has an animated diagram. The first step plays when it scrolls into view. Next plays the following step, Prev goes back one, and Play all runs the rest without stopping. The arrow keys work too once a diagram has focus.

1. Why replicate data?

There are three reasons to keep more than one copy.

• Availability. If one node dies, the other replicas keep serving the data.
• Latency. A replica close to the users means each request travels a shorter distance.
• Read throughput. More replicas can serve more reads at the same time.

This post assumes every node can hold the whole dataset. Splitting data that is too big for one machine is called partitioning, and it is a separate problem.

2. Leaders and followers

Single-leader replication is the most common setup. One replica is the leader and the rest are followers.

Clients send every write to the leader. The leader writes the change to its local storage, appends it to its replication log, and streams the log entries to the followers. Each follower applies the entries in the same order as the leader, so every replica ends up with the same data.

• Only the leader accepts writes. A follower is read-only and rejects them.
• Clients can read from the leader or from any follower.
• Each log entry has a position number. A follower’s position shows how far through the log it has got.

PostgreSQL streaming replication, MySQL replication, Oracle Data Guard, SQL Server Always On, MongoDB replica sets, Redis replication and Kafka partitions all work this way.

3. Synchronous vs asynchronous

The leader can ship a change to a follower in two ways. With synchronous replication, the leader waits for the follower to confirm the change before it tells the client the write succeeded. With asynchronous replication, the leader replies to the client straight away and sends the change afterwards.

The tabs above the diagram switch between three setups.

• Synchronous. The follower is always up to date. One slow or dead follower blocks every write.
• Asynchronous. Writes are fast and never wait on a follower. If the leader dies, writes it already confirmed can disappear.
• Semi-synchronous. One follower is synchronous and the rest are asynchronous, so every confirmed write lives on at least two nodes.

Most databases default to asynchronous. PostgreSQL picks its synchronous followers with synchronous_standby_names and synchronous_commit. MySQL’s semi-synchronous plugin waits for one replica. Redis is asynchronous, and its WAIT command makes a client block until a given number of replicas have the write.

4. Adding a new follower

New followers replace broken nodes or add read capacity. Copying the data files is not enough, because clients keep writing during the copy and the files change underneath it. The fix is a snapshot tied to a log position.

1. Take a consistent snapshot of the leader without locking the database.
2. Copy the snapshot to the new follower.
3. The snapshot records the log position it was taken at. The follower asks the leader for every change after that position.
4. Once it has applied them, the follower has caught up and carries on like any other follower.

5. Follower failure

Each follower keeps the log of changes it has received on its own disk. When a follower crashes and restarts, it reads its last position from disk and asks the leader for everything after it. This is catch-up recovery.

It only works if the leader still has those entries. If the leader already deleted them, the follower needs a fresh snapshot.

In PostgreSQL, a replication slot makes the leader keep its WAL until the follower has received it. In Redis, the replica sends PSYNC with its replication offset. If those bytes are still in the leader’s backlog, the replica gets a partial resync. If not, it gets a full one.

6. Leader failure: failover

Losing the leader is harder. A follower has to be promoted, clients have to send their writes to it, and the other followers have to start copying from it. This is failover, and either an operator or the system itself can run it.

Automatic failover has three steps.

1. Detect. Nodes exchange heartbeats. If the leader stays silent past a timeout, say 30 seconds, the others declare it dead.
2. Choose. The best candidate is the follower with the most recent data. The nodes usually vote, which makes this a consensus problem.
3. Reconfigure. Clients send writes to the new leader. If the old leader comes back, it has to become a follower.

Each step can go wrong.

• Lost writes. With asynchronous replication, the new leader can be missing the old leader’s last writes. The system usually discards them, even though clients were told they succeeded. GitHub hit this in 2012. An out-of-date MySQL follower was promoted and reused primary keys that Redis also stored, and some private data was shown to the wrong users.
• Split brain. Two nodes both believe they are the leader and both accept writes, and the data diverges. The Split brain tab shows the standard fix. Every leader gets an epoch number, every write carries it, and nodes reject writes from an older epoch. Raft calls this number a term.
• The timeout. A long timeout means a slow recovery. A short one triggers failovers when the network is only slow, which adds load to a system that is already struggling.

7. Replication log formats

There are four common ways for the leader to encode the changes it sends. The diagram compares the first and the third.

• Statement-based. The leader sends each SQL statement and the followers run it again. NOW(), RAND(), auto-increment columns and triggers can give a different result on each replica. MySQL used this before version 5.1, and today it switches to row-based logging for statements that aren’t deterministic.
• Write-ahead log shipping. The leader sends its low-level storage log. That log describes bytes on disk, so every replica must run the same storage engine version, which makes upgrades without downtime hard. PostgreSQL streaming replication and Oracle physical standby work this way.
• Logical, or row-based. The leader sends the new values of each changed row, in a format that doesn’t depend on the storage engine. MySQL’s row-based binlog and PostgreSQL’s logical decoding are examples, and Debezium reads these logs for change data capture.
• Trigger-based. Code inside the database copies the changes. It is flexible, but slower and easier to break.

Redis sits between the first and the third. It sends write commands, but rewrites them to be deterministic first.

8. Replication lag

With asynchronous replication, a follower can fall behind the leader. That delay is replication lag. It is usually under a second, but under heavy load or a network problem it can grow to seconds or minutes.

If writes stop, the followers eventually catch up and match the leader. This is eventual consistency, and it causes three common anomalies. Each tab shows one.

• Read-your-writes. A user always sees the data they wrote. One fix is to read anything the user can edit from the leader, for example for a minute after their last write. Another is for the client to remember the log position of its last write and only read from replicas that have reached it.
• Monotonic reads. Once a user has seen newer data, a later read never shows older data. The fix is to send each user to the same replica every time, for example by hashing the user ID.
• Consistent prefix reads. If writes happen in a certain order, every reader sees them in that order. The fix is to write causally related data to the same partition.

Synchronous replication would avoid all three, but read-heavy setups have many followers. Waiting on all of them is too slow, and any single bad node would block writes. So these systems stay asynchronous and the application handles the lag.

9. Multi-leader replication

In multi-leader replication, more than one node accepts writes, and each leader is also a follower of the others. It fits three cases. The first is several datacenters. The second is apps that work offline, where each device acts as a leader. The third is real-time collaborative editing. Inside a single datacenter, the extra complexity rarely pays off.

Each datacenter has its own leader, so users write to a nearby node and writes stay fast. If one datacenter goes down, the others keep accepting writes. The cost is that two leaders can accept conflicting writes to the same data. Auto-increment keys, triggers and constraints also cause trouble here.

Resolving conflicts

• Avoid them. Send all writes for a given record to the same leader.
• Last write wins. Keep the write with the highest timestamp and drop the rest. This is simple, but it loses data without telling anyone, and clock skew between datacenters makes it worse.
• Merge. Keep every value and let the application or the user pick one.
• CRDTs. These data types merge concurrent changes on their own.

Topologies

The leaders also need a route for passing writes to each other.

10. Leaderless replication

Leaderless replication has no leader at all. The client sends each write to several replicas in parallel, and reads from several replicas in parallel too. Amazon described this design in its 2007 Dynamo paper, so these systems are often called Dynamo-style.

• Each value carries a version number, and the client keeps the newest one it receives.
• Read repair. When a read finds a replica with an old value, the client writes the newer value back to it.
• Anti-entropy. A background process compares replicas and copies missing data, including data nobody reads.
• Sloppy quorums and hinted handoff. If the usual replicas are unreachable, other nodes accept the writes for a while and hand them back later.

Cassandra, ScyllaDB, Riak and Voldemort are leaderless. Amazon DynamoDB shares the name but is a different system. It uses a leader for each partition.

11. Quorums

A leaderless system stores each value on n replicas. A write must be confirmed by w of them, and a read must get replies from r of them. If w + r > n, the replicas written to and the replicas read from must overlap, so every read reaches at least one replica with the latest value.

Move the sliders and draw new random sets to test the rule.

A common choice is an odd n with w = r = (n + 1) / 2, such as n = 3, w = 2, r = 2. A smaller w makes writes faster and a smaller r makes reads faster, but the sum has to stay above n to keep the guarantee.

Even with w + r > n, edge cases can return old values. Sloppy quorums, concurrent writes, writes that reached fewer than w replicas, and a replica restored from an old backup all break it. A quorum makes a fresh read very likely. It is not a hard guarantee like linearizability.

Summary

• Single-leader replication is simple and gives one order of writes. Failover is the hard part.
• Asynchronous replication causes lag, and lag breaks read-your-writes, monotonic reads and consistent prefix reads.
• Multi-leader and leaderless replication accept writes on many nodes, so they have to resolve conflicts.
• Quorums with w + r > n help leaderless systems return recent values.

Self-check

Answer each question before opening it.