Replication

Replication keeps copies of the same data on several machines. It serves three purposes: fault tolerance, so that data survives the loss of a machine and the service stays up; performance, so that reads can be spread over many replicas; and latency, so that copies can sit near users in different regions. Replication would be trivial if data never changed: copy it once and you are done. All the difficulty comes from handling changes. Every write must reach every replica, replicas fail and recover in the middle of updates, and clients may read from a replica that has not yet received the latest write. This lesson surveys the main replication designs and their trade-offs; lesson 7 then defines precisely what clients can observe under each, and lessons 8 and 9 show how consensus makes replication fault tolerant without giving up consistency.

Ledgerline's ledger is the system's most important data, so it is replicated three ways within a region and asynchronously to a second region. Its read-heavy reporting service uses many replicas of a derived data set. Its shopping-cart-like draft invoices use a different design again, tuned for availability. Different data, different replication.

Single-leader replication

The most common design has one replica, the leader or primary, accept all writes. The leader applies each write locally and sends it to the other replicas, the followers, usually as an ordered stream of changes from its log. Followers apply the changes in the same order, so they pass through the same states as the leader, a little behind. Reads can go to the leader, for the freshest data, or to followers, to spread the load. PostgreSQL streaming replication, MySQL replication, MongoDB replica sets and Kafka partitions all work this way. CMPUT-291 (lesson 10) described the write-ahead log of a single database; single-leader replication essentially ships that log to other machines.

The central choice is when the leader tells the client a write has succeeded. With synchronous replication, it waits until followers confirm they have the write; the write then survives the leader's failure, but every write pays a round trip and one slow or failed follower can stall all writes. With asynchronous replication, the leader replies as soon as it has written locally and sends the change afterwards; writes are fast and followers' failures do not matter, but if the leader fails before a change reaches any follower, the acknowledged write is lost. A common compromise, semi-synchronous replication, waits for one follower out of several, so there are always two up-to-date copies.

Followers that lag behind cause replication lag: a client that writes to the leader and then reads from a follower may not see its own write. Lag is usually milliseconds but can grow to minutes under load or after a failure. Lesson 7 describes guarantees, such as read-your-writes, that hide lag from users.

Failover

When the leader fails, a follower must be promoted. Failover sounds simple and is one of the most error-prone operations in operations practice. The system must detect the failure, which is uncertain (lesson 1); choose the new leader, ideally the most up-to-date follower; reconfigure clients and other followers to use it; and make sure the old leader, if it comes back, does not keep accepting writes. Each step has failure modes. If the timeout is too short, a busy leader is declared dead and the system fails over needlessly, sometimes repeatedly. If the new leader was behind, acknowledged writes that only the old leader had are lost, or worse, conflict with new writes when the old leader returns. If both leaders accept writes, split brain corrupts data. GitHub described an outage in 2018 in which a brief network partition caused an automatic failover across data centers, leaving writes on both sides that took about 24 hours to reconcile. The robust solution is to make leader election and log replication a consensus problem, as Raft does in lesson 9: a new leader must be chosen by a majority, must hold every committed entry, and is identified by a term number that fences off the old leader.

Primary-backup and chain replication

Two classic designs make the leader-follower idea precise. In primary-backup replication, the primary executes each request and forwards the resulting state change to the backups before replying; on failure, a backup takes over. Chain replication, proposed by van Renesse and Schneider in 2004, arranges replicas in a line. Writes enter at the head and are passed down the chain; the tail acknowledges the write to the client and also answers all reads. Because a write is acknowledged only after every node has it, and reads see only what the tail has, every read returns the latest acknowledged write, giving strong consistency with a simple design. Failures are handled by removing a node from the chain and having neighbors re-send pending updates. Chain replication and its variants (CRAQ, which allows reads at any node) are used in storage systems such as Microsoft Azure's internal storage layers and several research and commercial key-value stores.

# Chain replication: writes enter at the head and flow to the tail; reads go to the tail.
class Node:
    def __init__(self, name): self.name, self.store = name, {}

def write(chain, key, value, fail_at=None):
    for i, node in enumerate(chain):
        if i == fail_at:
            return f"node {node.name} failed during propagation; write not acknowledged"
        node.store[key] = value
    return f"acknowledged by the tail, {chain[-1].name}"

def read(chain, key):
    return chain[-1].store.get(key)        # only the tail answers reads

chain = [Node("A"), Node("B"), Node("C")]      # A is the head, C the tail
print(write(chain, "acct-7", 140))
print("read at tail:", read(chain, "acct-7"))
print(write(chain, "acct-7", 180, fail_at=2))
print("head A has", chain[0].store["acct-7"], "but read at tail returns", read(chain, "acct-7"),
      "(the unacknowledged write is invisible to readers)")
chain.pop()                                # remove the failed tail; mid becomes the tail
print("after reconfiguration, tail is", chain[-1].name, "- head A re-sends pending writes")
print(write(chain, "acct-7", 180))
print("read at new tail:", read(chain, "acct-7"))

acknowledged by the tail, C
read at tail: 140
node C failed during propagation; write not acknowledged
head A has 180 but read at tail returns 140 (the unacknowledged write is invisible to readers)
after reconfiguration, tail is B - head A re-sends pending writes
acknowledged by the tail, B
read at new tail: 180

Multi-leader replication

With several data centers, a single leader forces every write to cross the world to wherever the leader is. Multi-leader replication lets each data center have its own leader that accepts writes locally and replicates them asynchronously to the others. Writes are fast and each data center keeps working if the links between them fail. The cost is that two leaders can accept conflicting writes to the same record at the same time, and the conflict is discovered only later, when the changes are exchanged. Offline-capable applications, such as calendar or note apps that work on a phone without a network, are multi-leader systems in which every device is a leader.

Conflicts can be handled in several ways. Last-writer-wins keeps the write with the highest timestamp and discards the others; it is simple but silently loses data, especially with skewed clocks (lesson 3). Applications can merge conflicting values with custom logic, or record all versions and let a person choose. And some data types can be designed so that concurrent updates always merge automatically and deterministically; these conflict-free replicated data types (CRDTs) are the subject of part of lesson 7. Ledgerline avoids multi-leader writes for its ledger, where losing an update is unacceptable, but uses them for draft invoices, which merge field by field.

Leaderless replication and quorums

Amazon's Dynamo paper (2007) popularized a design with no leader at all, later adopted by Apache Cassandra, Riak and others. A client, or a coordinator node acting for it, sends each write to all \(N\) replicas of a key and considers it successful when \(W\) of them acknowledge. A read is sent to the replicas and succeeds when \(R\) reply; the client takes the value with the newest version. If \(R + W > N\), every read quorum overlaps every write quorum in at least one replica, so a read normally sees the latest successful write.

# Quorum replication: how often does a read miss the latest write?
# N replicas; a write is acknowledged after W replicas apply it, and the rest apply it
# later (replication lag). A read asks R random replicas and returns the newest version.
import random
random.seed(450)

def trial(n, r, w, lag_prob=0.6):
    # after an acknowledged write: exactly W replicas surely have it; each other replica
    # still lags (does not have it yet) with probability lag_prob
    has = [True] * w + [random.random() > lag_prob for _ in range(n - w)]
    random.shuffle(has)
    asked = random.sample(range(n), r)
    return any(has[i] for i in asked)          # fresh if any replica asked has the write

print(" N  R  W   R+W>N   stale reads")
for n, r, w in [(3, 1, 1), (3, 1, 2), (3, 2, 2), (3, 1, 3), (5, 2, 2), (5, 2, 3), (5, 3, 3)]:
    stale = sum(not trial(n, r, w) for _ in range(20000)) / 20000
    print(f"{n:2d} {r:2d} {w:2d}   {'yes' if r + w > n else 'no ':3s}    {stale:7.2%}")

 N  R  W   R+W>N   stale reads
 3  1  1   no      39.75%
 3  1  2   no      20.11%
 3  2  2   yes      0.00%
 3  1  3   yes      0.00%
 5  2  2   no      10.80%
 5  2  3   no       3.65%
 5  3  3   yes      0.00%

The simulation shows the overlap condition at work: whenever \(R + W > N\), the read always reaches a replica holding the write, and smaller quorums sometimes miss it, more often the smaller they are. Tuning \(R\) and \(W\) trades latency against freshness and availability. \(N = 3, W = 2, R = 2\) is a common default. Setting \(W = N\) makes reads cheap (\(R = 1\)) but a single unavailable replica blocks writes. Setting \(W = 1\) makes writes very available but risks losing a write acknowledged by only one replica that then fails.

Quorums are not as strong as they look. The overlap argument assumes writes and reads are well defined points in time, but real systems have concurrent writes, partially failed writes (a write that reached fewer than \(W\) replicas is reported as failed yet may still be visible on some), and replicas that recover with old data. Leaderless stores therefore usually provide eventual consistency rather than linearizability (lesson 7), unless extra mechanisms are added.

Keeping replicas in sync: read repair, hinted handoff and anti-entropy

Leaderless systems use three mechanisms to make replicas converge. With read repair, when a read finds some replicas holding an older version, the coordinator writes the newest version back to them. With hinted handoff, when a replica is unreachable during a write, another node accepts the write on its behalf, keeping a hint, and forwards it when the replica returns; using such stand-ins to reach \(W\) acknowledgments is called a sloppy quorum, which improves availability but means \(R + W > N\) no longer guarantees overlap with the intended replicas. Finally, anti-entropy processes run in the background, comparing replicas and copying missing data.

Comparing two replicas key by key would require sending every key across the network. Dynamo and Cassandra instead use Merkle trees, named after Ralph Merkle, who patented the idea in 1979. Each leaf is the hash of one key's value, and each internal node is the hash of its children, so the root summarizes the whole data set. Two replicas compare roots; if they match, the data is identical. If not, they compare children and descend only into subtrees whose hashes differ. Finding a few differences among millions of keys takes a number of comparisons proportional to the number of differences times the depth of the tree. Git, blockchains and certificate transparency logs use Merkle trees for the same reason.

# Anti-entropy with Merkle trees: find the differing keys by comparing hashes top-down.
import hashlib

def h(s): return hashlib.sha256(s.encode()).hexdigest()[:8]

def build(items):                          # items: sorted list of (key, value)
    level = [h(f"{k}={v}") for k, v in items]
    tree = [level]
    while len(level) > 1:
        level = [h(level[i] + (level[i + 1] if i + 1 < len(level) else "")) for i in range(0, len(level), 2)]
        tree.append(level)
    return tree                            # tree[-1][0] is the root

def diff(t1, t2, keys):
    compared, todo = 0, [(len(t1) - 1, 0)]
    found = []
    while todo:
        lvl, i = todo.pop()
        compared += 1
        if i >= len(t1[lvl]) or t1[lvl][i] == t2[lvl][i]:
            continue                       # identical subtree: skip all keys below it
        if lvl == 0:
            found.append(keys[i])
        else:
            todo += [(lvl - 1, 2 * i), (lvl - 1, 2 * i + 1)]
    return found, compared

keys = [f"acct-{i:04d}" for i in range(1024)]
a = [(k, 100) for k in keys]
b = list(a)
for i in (17, 600):                        # replica b missed two updates
    b[i] = (keys[i], 100 + i)
ta, tb = build(a), build(b)
print("roots equal?", ta[-1][0] == tb[-1][0])
found, compared = diff(ta, tb, keys)
print("differing keys:", sorted(found))
print(f"hash comparisons: {compared} (instead of {len(keys)} key-by-key comparisons)")

roots equal? False
differing keys: ['acct-0017', 'acct-0600']
hash comparisons: 39 (instead of 1024 key-by-key comparisons)

State machine replication

The most general approach, introduced by Lamport in 1978 and surveyed by Fred Schneider in 1990, treats the service as a deterministic state machine: given the same initial state and the same sequence of commands, every replica produces the same state and outputs. Replication then reduces to two requirements: every replica receives the same commands, and in the same order. The ordered sequence is a replicated log, and agreeing on the next entry of the log, despite failures, is exactly the consensus problem of lessons 8 and 9. Determinism matters: a command that reads the local clock, generates a random number or depends on thread scheduling would make replicas diverge, so such values are chosen once, by the leader, and recorded in the log entry. ZooKeeper, etcd, Spanner's Paxos groups, CockroachDB's ranges and Kafka's KRaft metadata all use state machine replication.

Worked examples