Replication: Leader-Based, Multi-Leader, Leaderless and Quorums

Version note: The code in this post was built and run with C# 14 and .NET 10.

Guess First

Bookstore's stock data lives on three replicas (A, B, C). A write counts as successful once at least two replicas acknowledge it (W=2), and a read asks two replicas and takes the newest version (R=2). A is down. A book's stock is written as 12 and the write succeeds. Then B goes down too. The write that lowers the stock to 11 reaches only C; one acknowledgement, two required, so the system tells the client "failed".

The client got an error and believes 11 was not written. A few minutes later A and B come back and someone reads the stock. Do they see 12 or 11?

Why Replicate, and Sync or Async?

Replication means keeping copies of the same data on several machines. The reasons overlap with the reasons for distributing in the first post: not losing data and continuing to serve when a machine crashes, spreading read load across machines, keeping data close to users. If data never changed, replication would be easy; all the difficulty lies in carrying changes to the copies.

The first decision is when a write counts as "done". With synchronous replication the leader does not answer the client until the replicas confirm they have the write: nothing is lost on the replica, but the slowest replica slows every write, and if one goes down writes stop. With asynchronous replication the leader writes locally and answers immediately: it is fast and a failing replica does not affect writes, but if the leader crashes at that moment, writes not yet copied are lost. The common middle ground is semi-synchronous: one replica synchronous, the others asynchronous. The synchronous-commit and asynchronous-commit modes of SQL Server Always On availability groups, PostgreSQL's synchronous_standby_names setting and MySQL's semi-synchronous replication plugin are different interfaces to the same decision.

The second decision is who accepts writes. That gives three topologies:

Leader-Based Replication and Failover

The most common model: all writes go to a single leader, the leader sends changes to followers through a replication log, and followers can serve reads. Because writes are ordered in one place there are no conflicts; the trouble starts when the leader crashes.

Failover has three steps: deciding that the leader has crashed, choosing a new leader, and redirecting clients to it. Each step carries a problem from the earlier posts. The decision rests on a timeout, which, as we saw in the second post, can raise false alarms. With asynchronous replication the new leader may not have received the old leader's last writes, and those writes are lost. If the old leader was not dead but merely slow, and comes back still believing it is the leader, two leaders accept two different writes: split brain.

Kleppmann tells a real example of this risk in DDIA (Designing Data-Intensive Applications): at GitHub in 2012 an out-of-date MySQL follower was promoted to leader. Auto-increment ids the old leader had handed out were reused by the new leader, and because the same ids were also stored in Redis, some users were shown other users' private data. A few lost writes were a small problem on their own; with another system referencing that data from outside, they became a security incident. That is why a good failover mechanism fences the old leader off: once a new leader is chosen, writes from the old one are rejected. We will look at fencing tokens in detail in the tenth post, and at how Raft elects leaders without split brain in the eighth.

Multi-Leader: Writes in Every Region, Unavoidable Conflicts

Suppose Bookstore has two data centers, in Istanbul and Frankfurt. If the single leader is in Frankfurt, every write in Istanbul pays the cross-region latency, and if the link between the regions breaks, Istanbul cannot write. Multi-leader replication gives each region its own leader; the leaders send writes to each other asynchronously. An app on your phone that works offline uses the same idea: the device is a leader too.

The price is conflicts. When an editor in Istanbul sets a book's price to 450 and an editor in Frankfurt sets it to 399 at the same time, both leaders accepted the write; replication has to decide which one wins. The options are familiar: last-write-wins (LWW) loses data through clock skew (third post), a version vector makes the conflict visible and leaves it to the application (fourth post), a CRDT (Conflict-free Replicated Data Type) designs the data type so it cannot produce conflicts (eleventh post). The safest option is to avoid conflicts: always route writes for the same record to the same leader, for example by tying a customer's cart to the customer's region.

Leaderless Replication and Quorums

In the model popularized by Amazon's Dynamo paper (2007) and used by Cassandra and Riak, there is no leader. The client (or a coordinator acting on its behalf) sends the write to all N replicas and waits for W acknowledgements. On a read it asks R replicas and takes the newest version among them. If R + W > N, the replicas that acknowledged the write and the replicas the read asks overlap in at least one node, so the read sees at least one up-to-date copy. The typical choice for N=3 is W=2, R=2: both writes and reads keep working with one node down.

Let us see it in a small simulation. Each replica keeps a value and a version per key; a replica can be told to miss the next write. In the simulation a single counter produces versions; in real systems this is a timestamp or a version vector:

namespace Bookstore.Replication;

public record Stored(string Value, long Version);

public sealed class Replica(string name)
{
    private readonly Dictionary<string, Stored> _data = [];

    public string Name { get; } = name;
    public bool IsUp { get; set; } = true;
    public bool DropsNextWrite { get; set; } // simulates a lost or delayed replication message

    public bool TryWrite(string key, Stored value)
    {
        if (!IsUp) return false;
        if (DropsNextWrite)
        {
            DropsNextWrite = false;
            return false;
        }
        if (!_data.TryGetValue(key, out Stored? current) || value.Version > current.Version)
        {
            _data[key] = value;
        }
        return true;
    }

    public Stored? Read(string key) => IsUp ? _data.GetValueOrDefault(key) : null;
}

public sealed class QuorumStore(IReadOnlyList<Replica> replicas, int w, int r)
{
    private long _version;

    public bool Write(string key, string value)
    {
        Stored stored = new(value, ++_version);
        int acks = replicas.Count(replica => replica.TryWrite(key, stored));
        Console.WriteLine($"  write {key}={value} v{stored.Version}: acks={acks}/{replicas.Count} need W={w} -> {(acks >= w ? "OK" : "FAILED")}");
        return acks >= w;
    }

    // The coordinator asks only R replicas, chosen by the caller (e.g. the fastest ones)
    public string? Read(string key, params string[] askFirst)
    {
        int Rank(Replica x) => Array.IndexOf(askFirst, x.Name) is >= 0 and var i ? i : int.MaxValue;
        List<Replica> asked = [.. replicas.Where(x => x.IsUp).OrderBy(Rank).Take(r)];
        if (asked.Count < r)
        {
            Console.WriteLine($"  read {key}: only {asked.Count} replicas up, need R={r} -> UNAVAILABLE");
            return null;
        }
        List<(Replica Replica, Stored? Value)> answers = [.. asked.Select(x => (x, x.Read(key)))];
        Stored? newest = answers.Select(a => a.Value).Where(v => v is not null).MaxBy(v => v!.Version);
        foreach ((Replica replica, Stored? value) in answers.Where(a => newest is not null && (a.Value is null || a.Value.Version < newest.Version)))
        {
            replica.TryWrite(key, newest!); // read repair
            Console.WriteLine($"  read repair: {replica.Name} v{value?.Version ?? 0} -> v{newest!.Version}");
        }
        string seen = string.Join(", ", answers.Select(a => $"{a.Replica.Name}:v{a.Value?.Version ?? 0}"));
        Console.WriteLine($"  read {key} from [{seen}] -> {newest?.Value ?? "(none)"}");
        return newest?.Value;
    }
}

Three scenarios: the quorums overlap, the quorums do not overlap, nodes go down:

using Bookstore.Replication;

static Replica[] Cluster() => [new("A"), new("B"), new("C")];

Console.WriteLine("N=3, W=2, R=2 (R + W > N)");
Replica[] nodes = Cluster();
QuorumStore store = new(nodes, w: 2, r: 2);
store.Write("stock:978-0134494166", "12");
nodes[2].DropsNextWrite = true;                 // C misses the next update
store.Write("stock:978-0134494166", "11");
store.Read("stock:978-0134494166", "C", "B");   // C answers first, but B overlaps the write quorum

Console.WriteLine("N=3, W=1, R=1 (R + W <= N)");
nodes = Cluster();
store = new(nodes, w: 1, r: 1);
store.Write("stock:978-0134494166", "12");
nodes[1].DropsNextWrite = true;
nodes[2].DropsNextWrite = true;                 // only A gets the update
store.Write("stock:978-0134494166", "11");
store.Read("stock:978-0134494166", "C");

Console.WriteLine("N=3, W=2, R=2, one node down, then two");
nodes = Cluster();
store = new(nodes, w: 2, r: 2);
nodes[0].IsUp = false;
store.Write("stock:978-0134494166", "12");
store.Read("stock:978-0134494166", "B", "C");
nodes[1].IsUp = false;
store.Write("stock:978-0134494166", "11");
store.Read("stock:978-0134494166", "C");
nodes[0].IsUp = true;
nodes[1].IsUp = true;                            // the "failed" write is still on C
store.Read("stock:978-0134494166", "C", "A");

N=3, W=2, R=2 (R + W > N)
  write stock:978-0134494166=12 v1: acks=3/3 need W=2 -> OK
  write stock:978-0134494166=11 v2: acks=2/3 need W=2 -> OK
  read repair: C v1 -> v2
  read stock:978-0134494166 from [C:v1, B:v2] -> 11
N=3, W=1, R=1 (R + W <= N)
  write stock:978-0134494166=12 v1: acks=3/3 need W=1 -> OK
  write stock:978-0134494166=11 v2: acks=1/3 need W=1 -> OK
  read stock:978-0134494166 from [C:v1] -> 12
N=3, W=2, R=2, one node down, then two
  write stock:978-0134494166=12 v1: acks=2/3 need W=2 -> OK
  read stock:978-0134494166 from [B:v1, C:v1] -> 12
  write stock:978-0134494166=11 v2: acks=1/3 need W=2 -> FAILED
  read stock:978-0134494166: only 1 replicas up, need R=2 -> UNAVAILABLE
  read repair: A v0 -> v2
  read stock:978-0134494166 from [C:v2, A:v0] -> 11

In the first scenario C missed the update and was the first to answer the read, but because the read also asked B, it found the newest version and fixed C on the spot. This is called read repair; Dynamo-style systems additionally run an anti-entropy process in the background that compares replicas. In the second scenario R + W = 2 ≤ 3: the write went to one replica, the read asked another, and it returned 12 instead of 11. A fast but stale answer; acceptable for some workloads, not for stock.

The Limits of Quorums: A Failed Write Does Not Disappear

The third scenario is the opening question. With A down, W=2 is still met. When B goes down too, the write gets a single acknowledgement and the client receives FAILED; the read is refused for lack of replicas. The system gave up availability to stay consistent. But look at the last line: once A and B are back, the read returned 11, the value of the "failed" write, and read repair spread it to A. The write stayed on the one replica it reached and was never undone. The client sees a change it was told had failed take effect later.

This is not a bug in the simulation; it follows directly from what a quorum guarantees. "I could not get W acknowledgements" does not mean "the write is nowhere"; like the timeout in the first post, it means "I do not know". The limits of quorums in leaderless systems do not end there: which of two concurrent writes wins is still left to versioning (often LWW); some systems, when they cannot reach enough replicas, temporarily write to another node and hand the write over to its rightful owner later (Dynamo's sloppy quorum and hinted handoff), which temporarily breaks the R + W > N overlap; and a read racing a write can see only part of the write. That is why R + W > N alone does not provide linearizability, which we define in the next post.

Replication Lag and the Anomalies Users See

With asynchronous replication, followers can fall behind the leader by a few milliseconds, or seconds or even minutes under load. To the user this lag shows up as three typical anomalies.

A read-your-writes violation: the user submits a comment, the page reloads, the read goes to a follower that has not caught up, and the comment is missing. The fix is to read the user's own data from the leader for a while, or to give the client the version returned by the write (an LSN (Log Sequence Number) in PostgreSQL) and require the read to go to "a replica that has seen at least this version".

A monotonic reads violation: the user reloads the page twice, the first read goes to an up-to-date follower and the second to a lagging one; the comment appears and then disappears. The fix is to route a user's reads to the same replica every time (for example by hashing the user id).

A consistent prefix violation: two ordered writes (a question and its answer) land on different partitions and are replicated with different delays, so a reader sees the answer before the question. The fix is to put causally related writes on the same partition or to carry the causal order explicitly.

These guarantees are called session guarantees; Terry and colleagues defined them in 1994 for the Bayou system. In the next post we will place them on the consistency model map.

Broadcast Types and Replication

You can look at replication from another angle too: every write is a message, every replica a receiver. The question is in which order the messages are delivered. FIFO (First In, First Out) broadcast delivers messages from the same sender in the order they were sent. Causal broadcast respects happens-before: a message sent in reply to another message is delivered after it everywhere (the fix for the consistent prefix problem). Total order broadcast delivers all messages to all nodes in the same order.

Total order broadcast is the strongest tool for replication: if every replica starts from the same state and applies the same writes in the same order, they all reach the same state (state machine replication). Leader-based replication achieves this with a single leader's ordering; continuing that ordering fault-tolerantly when the leader crashes is equivalent to the consensus problem. In the eighth post we will read Raft with exactly this in mind.

Common Pitfalls

Reading from a follower and expecting to see your own writes. A DbContext configuration or a read-only connection string that spreads reads across replicas may not show the change the user just made. Decide deliberately which queries may go to a replica.

Reading a quorum failure as "the write did not happen". As you saw above, a failed write may have been partially applied. If you are going to retry, the write must be idempotent.

Setting up automatic failover without fencing. A failover that cannot stop the old leader or reject its writes leaves split brain to timing alone.

Choosing multi-leader for "faster writes" and postponing conflict resolution. Conflict resolution is part of the data model; it cannot be bolted on later.

Not monitoring replication lag. How far a follower is behind the leader (in seconds or log position) should be a metric; lag quietly growing to hours is the amount of data you lose at failover.

A Small Experiment

★☆☆ In the first scenario, change the replicas the read asks from "C", "B" to just "C" and run with r: 1. What does the read return? Does read repair run?

★★☆ Add a "rollback" to QuorumStore.Write that undoes the change on the replicas that acknowledged when there are not enough acknowledgements. Why is this approach unreliable in a real system? What happens if the rollback message is lost too?

★★★ Add a scenario where two clients write different values to the same key at the same time: one reaches A and B, the other B and C, and both get the same version number. Which value does the read return? Change the version to a version vector and keep the conflict as siblings.

In this post we saw situations where replicas cannot agree on "the newest value". In the next post we define precisely what "correct behavior" means: linearizability, sequential, causal and eventual consistency, session guarantees, and a small C# checker that decides in which model a history of reads and writes is valid.