Replication & Consistency Models

Replication solves three distinct problems, and it is important to keep them separate because they lead to different design choices.

1. Fault Tolerance (High Availability): If a single machine holds your database and it crashes, your service is down. With n replicas, you can tolerate f failures (for some f < n) and keep serving requests. This is the most cited motivation, but it comes with the highest consistency burden — the surviving replicas must somehow agree on what the current state of the data is.

2. Read Scalability (Throughput): A single machine can only serve so many requests per second. If your workload is read-heavy (as most web workloads are), you can route reads to any replica and multiply your read throughput by the number of replicas. A replica set of 5 machines can serve roughly 5× the read load of a single machine. Writes, however, still must be coordinated — you cannot simply multiply write throughput by adding replicas (that requires sharding/partitioning).

3. Geographic Locality (Latency): If your users are in New York and your database is in Singapore, every read takes ~200 ms of round-trip latency. By placing replicas in multiple geographic regions, you can serve reads locally, dramatically reducing latency. This is the motivation behind CDNs, geo-distributed databases like CockroachDB, and global replication in Spanner.

Each motivation introduces tension: more replicas → higher consistency overhead. Read scaling and geographic distribution demand that stale reads be tolerable at least some of the time, pushing toward weaker consistency models. Fault tolerance can be achieved with strong consistency, but at the cost of latency (you must wait for quorum acknowledgment) and availability (if too many replicas fail, you block).

# The core tension, expressed simply:
# We have 3 replicas. A write arrives at replica 1.
# Before we ack the write to the client, how many replicas
# must confirm they received it?

# Option A: Wait for all 3  -> Strong consistency, but if ANY replica is slow/down, write blocks
# Option B: Wait for 2 of 3 -> Can tolerate 1 failure, reads need to check 2 replicas
# Option C: Wait for just 1 -> Fast writes, but replicas diverge; stale reads possible

REPLICATION_FACTOR = 3
write_quorum = 2   # W
read_quorum  = 2   # R
# Quorum condition for strong consistency: W + R > REPLICATION_FACTOR
assert write_quorum + read_quorum > REPLICATION_FACTOR  # 4 > 3 ✓

Primary-backup (also called leader-follower or master-slave) is the simplest and most widely deployed replication architecture. One replica is designated the primary (leader); all writes go through it. The primary applies the write to its local state and then forwards it to the backups (followers). Reads can go to the primary (for strong consistency) or to backups (for read scaling, at the cost of possible staleness).

Synchronous vs. Asynchronous replication:

• Synchronous: The primary waits for at least one backup to acknowledge the write before confirming to the client. This guarantees durability — if the primary crashes immediately after acknowledging, the backup has a copy. But it means write latency is at least the RTT to the nearest backup.

• Asynchronous: The primary acknowledges the client immediately and forwards the write to backups in the background. This gives lower write latency but means acknowledged writes can be lost if the primary crashes before the backup catches up. PostgreSQL's streaming replication supports both modes.

Failover: When the primary fails, one backup must be promoted to primary. This is harder than it sounds. How do backups know the primary is truly dead and not just slow? (This is the split-brain problem.) How do we ensure the elected backup has the most up-to-date state? Systems like ZooKeeper, etcd, and Raft-based systems solve this with a proper leader election protocol.

Replication lag: In async replication, backups can fall behind. Reading from a lagging backup gives stale data — you may read a value that has since been overwritten. This lag is typically milliseconds in a healthy system but can grow to minutes under high load or network partition.

# Simplified primary-backup write path (pseudocode)
class Primary:
    def write(self, key, value):
        # 1. Apply write locally
        self.store[key] = value
        self.log.append((key, value))

        # 2. Replicate to backups
        acks = 0
        for backup in self.backups:
            # Synchronous: wait for ack before proceeding
            if backup.replicate(key, value):
                acks += 1

        # 3. Acknowledge client only after at least 1 backup confirms
        if acks >= 1:  # semi-sync mode (e.g., MySQL semi-sync)
            return "OK"
        else:
            # Rollback or mark dirty — this is where complexity lives
            raise ReplicationError("No backup acked")

class Backup:
    def replicate(self, key, value):
        self.store[key] = value
        self.log.append((key, value))
        return True  # ack to primary

Quorum systems generalize primary-backup by removing the special role of the primary. Instead of routing all writes to one designated node, any node can accept a write, as long as it gets acknowledgment from a sufficient number of replicas — a quorum.

The key invariant is the quorum intersection property: every read quorum and every write quorum must share at least one replica. This guarantees that a read always observes at least the most recent write.

The fundamental quorum condition:

Let n = total replicas, W = write quorum size, R = read quorum size.

For strong consistency: W + R > n

For fault tolerance: W > n/2 (to prevent two concurrent writes from both succeeding without overlap)

Example with n=5:

• W=3, R=3: tolerates 2 failures, strong consistency
• W=4, R=2: every write goes to 4 nodes, reads need only 2; better read performance, still correct
• W=5, R=1: writes are synchronous to all, reads can go to any single replica
• W=1, R=5: writes are fast, but reads check everyone — rarely used
• W=2, R=2: does NOT satisfy W+R > n (4 is not > 5) — stale reads are possible

Version vectors and conflict resolution: In a quorum read, you contact R replicas and may get different values (if a write was in progress). You resolve conflicts by taking the value with the highest version number (a logical timestamp or vector clock). DynamoDB's original design used this approach, allowing sloppy quorums for availability during partitions, at the cost of occasionally having to reconcile conflicting writes.

CRDT preview: Some data types (counters, sets, registers) can be designed so that all possible merge orders produce the same result — these are Conflict-free Replicated Data Types (CRDTs), which sidestep conflict resolution entirely by making all operations commutative and idempotent.

# Quorum read: contact R replicas, return highest-versioned value
import random

def quorum_read(key, replicas, R):
    """Contact R replicas, return value with highest version."""
    chosen = random.sample(replicas, R)
    responses = [replica.read(key) for replica in chosen]
    # responses is list of (version, value) tuples
    best = max(responses, key=lambda r: r.version)
    return best.value

def quorum_write(key, value, replicas, W):
    """Write to W replicas. Increment version."""
    # First, read current version from a quorum to get latest version
    current = quorum_read(key, replicas, W)
    new_version = current.version + 1 if current else 1
    acks = 0
    for replica in replicas:
        if replica.write(key, value, new_version):
            acks += 1
        if acks >= W:
            return True  # Success
    return False  # Failed to reach write quorum

# Quorum condition check
n, W, R = 5, 3, 3
assert W + R > n, "Quorum condition violated: stale reads possible"

A consistency model is a contract between a distributed system and the applications that use it. It specifies what values a read operation is allowed to return, given the history of all preceding writes. This is fundamentally a statement about the set of legal execution histories the system may produce.

To define consistency models precisely, we need some vocabulary:

• An operation is a read or write with an invocation (when the client sends the request) and a response (when the client receives the result). Together these define the operation's real-time interval.
• A history is a set of operations with their invocations and responses.
• Two operations are concurrent if neither's real-time interval ends before the other's begins.
• A sequential history is one where no two operations overlap in time.

The spectrum of consistency models (from strongest to weakest):

1. Strict Consistency (theoretical): every read returns the most recent write globally, using a universal clock. Impossible in a distributed system (clocks are not perfectly synchronized).
2. Linearizability (external consistency): operations appear to take effect atomically at some point within their real-time interval. Equivalent to having a single copy of the data. Achievable in practice, at the cost of coordination.
3. Sequential Consistency: all processes see operations in the same total order, and each process's operations appear in program order. Does not respect real time — a write can appear to happen "before" a concurrent write that finished first in wall clock time.
4. Causal Consistency: causally related operations are seen in causal order by all processes. Concurrent operations may be seen in different orders by different processes.
5. Eventual Consistency: if no new writes occur, all replicas will eventually converge to the same value. Says nothing about when, or what happens during convergence.

Weaker models enable higher availability and lower latency but shift the burden of reasoning about correctness onto application developers.

# Illustrating the difference between consistency levels
# using a simple timeline notation.
#
# History: Client A writes x=1, then Client B writes x=2.
# Both operations complete before Client C reads x.
# Under linearizability: C must read 2 (the last write).
# Under sequential consistency: C could read 1 or 2, BUT
#   all clients must agree on WHICH value they all see.
# Under eventual consistency: C might read 1, 0, or 2 —
#   but eventually all replicas will converge to 2.

# Concrete violation of linearizability (but not sequential consistency):
# Timeline (wall clock time flows right):
#   A: ---[write x=1]-------
#   B: -----------[write x=2]---
#   C1: ----------------[read x] -> returns 2  (OK)
#   C2: ----------------[read x] -> returns 1  (VIOLATES linearizability!
#                                                x=2 was written before read)
#
# Sequential consistency allows this if ALL clients see
# operations in the order: write(x=1), read(x)->1, write(x=2)
# Even if that order doesn't match real time.
pass

Linearizability (Herlihy & Wing, 1990) is the strongest practical consistency model. An execution history is linearizable if there exists a linearization — a sequential reordering of all operations such that:

1. The reordering respects the real-time ordering of non-overlapping operations (if op A completes before op B begins, A comes before B in the linearization).
2. The sequential execution in the linearization is consistent with the data type's specification (for a register: every read returns the value of the most recent preceding write).

Informally: each operation appears to take effect instantaneously at some point within its real-time interval. From a client's perspective, the system behaves as if there is a single, atomic copy of the data.

Why linearizability matters:

• Compare-and-swap (CAS) operations used in distributed locking, leader election, and optimistic concurrency control require linearizability to be correct. If two processes both read value v and both try to CAS from v to their own value, linearizability guarantees exactly one succeeds.
• Reads after writes: If I write my new profile picture and then share the link, my friends should see the new picture, not the old one. This is guaranteed by linearizability but not by eventual consistency.

Cost: Achieving linearizability requires coordination (e.g., quorum reads and writes, or Raft/Paxos consensus). This means higher latency and reduced availability. Under the CAP theorem, a linearizable system cannot remain available during a network partition — it must refuse requests from the minority partition.

Key systems: etcd, ZooKeeper, Google Spanner (for individual key operations), Redis with Raft (in Redis 7+).

# Checking if a history is linearizable:
# We have 3 operations on a single register (initial value 0):
#   Op1: write(1)   invoked at t=0, completed at t=10
#   Op2: read()     invoked at t=5, completed at t=15  -> returned 1
#   Op3: read()     invoked at t=5, completed at t=12  -> returned 0

# Is this linearizable?
# Op1 and Op2 overlap. Op1 and Op3 overlap.
# Op3 completed before Op2 started finishing.
# Try linearization: write(1) at t=7, then read@t=9->0, then read@t=13->1
# But read@t=9 must return 1 if write(1) happened at t=7... contradiction.
# Wait: read()->0 COULD be linearized BEFORE write(1).
# Linearization: read()->0 at t=3 (before write at t=0? NO, write was at t=0)
# Valid linearization: Op3(read->0) at t=1, Op1(write 1) at t=6, Op2(read->1) at t=13
# Check: Op3 returns 0 (correct, write(1) not yet happened). Op2 returns 1 (correct).
# Real-time order respected: Op1 [0,10], Op3's linearization point t=1 is in [5,12] OK
# This history IS linearizable.

# In contrast, if Op3 had returned 1 and Op2 had returned 0,
# no valid linearization would exist (write(1) cannot "unhappen").
pass

Sequential consistency (Lamport, 1979) is weaker than linearizability but still strong enough for many use cases. A history is sequentially consistent if there exists a sequential reordering of all operations such that:

1. Each process's operations appear in program order (the order they were issued by that process).
2. The sequential execution is consistent with the data type's specification.

Crucially, sequential consistency does not require the reordering to respect real-time order between operations from different processes. Concurrent writes from different processes can appear in any order, as long as all processes see the same total order.

The Lamport illustration: Consider two processes writing to two separate memory locations:

Process 1:  write(x=1)  then  read(y)
Process 2:  write(y=1)  then  read(x)

Under sequential consistency, it is possible for Process 1 to read y=0 AND Process 2 to read x=0 simultaneously — each process's reads see a state before the other's write, even though both writes completed before either read. Under linearizability, this is impossible.

Practical implications:

• Multi-processor CPUs with relaxed memory models (x86, ARM) do not provide sequential consistency by default. The mfence instruction on x86 or dmb on ARM inserts memory barriers to enforce ordering.
• Java's volatile keyword provides sequential consistency for the annotated variable.
• Many distributed databases that advertise "strong consistency" actually provide sequential consistency, not linearizability — they guarantee program order within a session but not global real-time ordering.

When sequential ≠ linearizable matters: Distributed compare-and-swap requires linearizability. If you build a distributed lock on top of sequential consistency alone, two processes can both believe they hold the lock.

// Java volatile: sequential consistency for a single variable
// Without volatile, the JVM/CPU can reorder instructions;
// with volatile, all threads see writes in a consistent order.

class SharedFlag {
    // volatile guarantees sequential consistency for 'ready'
    private volatile boolean ready = false;
    private int value = 0;

    // Thread 1
    void producer() {
        value = 42;     // write value first
        ready = true;   // then signal readiness (volatile write)
    }

    // Thread 2
    void consumer() {
        while (!ready) { /* spin */ }  // volatile read
        // Because of volatile, value is guaranteed to be 42 here.
        // Without volatile, CPU/JVM could reorder: ready=true before value=42
        System.out.println(value);  // guaranteed: 42
    }
}

Eventual consistency is the weakest practically used model and the most commonly misunderstood. The formal definition (Vogels, 2009):

If no new updates are made to a data item, eventually all accesses to that item will return the last updated value.

Notice what this does NOT say: it says nothing about how long "eventually" takes, what values are returned during the convergence period, or what happens if writes keep arriving. A system that loses all writes and returns null forever is technically eventually consistent (vacuously).

Why choose eventual consistency? The CAP theorem tells us that under a network partition, a system must choose between consistency (refuse to serve requests from the minority partition) and availability (continue serving, but potentially with stale data). Eventual consistency chooses availability. For systems like shopping carts, social media timelines, DNS, or CDN caches, serving slightly stale data is far better than refusing to serve at all.

Flavors of eventual consistency:

• Monotonic read consistency: If a process reads value v, subsequent reads will return v or a more recent value. Prevents "time travel" (reading an older value after a newer one).
• Read-your-writes consistency: A process always reads the effects of its own writes. Practically important: if I change my password, I should not be locked out by reading my old password hash.
• Monotonic write consistency: Writes from a single process are applied in order.
• Writes-follow-reads consistency: If a process reads value v and then writes w, w is applied after v (causal ordering for the write).

Strong Eventual Consistency (SEC): A data type provides SEC if: (a) all replicas that have received the same set of updates are in the same state (convergence), AND (b) all updates eventually propagate to all replicas. CRDTs (Conflict-free Replicated Data Types) achieve SEC by construction.

# A simple G-Counter CRDT: eventually consistent increment counter
# Each replica maintains a vector of per-replica counts.
# Merge is component-wise max. Convergence is guaranteed.

class GCounter:
    def __init__(self, replica_id, num_replicas):
        self.id = replica_id
        self.counts = [0] * num_replicas

    def increment(self):
        self.counts[self.id] += 1

    def value(self):
        return sum(self.counts)

    def merge(self, other):
        """Merge another replica's state. Component-wise max."""
        self.counts = [
            max(a, b)
            for a, b in zip(self.counts, other.counts)
        ]

# Demo: two replicas independently increment, then merge
r0 = GCounter(replica_id=0, num_replicas=2)
r1 = GCounter(replica_id=1, num_replicas=2)

r0.increment()  # r0.counts = [1, 0]
r0.increment()  # r0.counts = [2, 0]
r1.increment()  # r1.counts = [0, 1]

# After merge (in any order), both replicas converge:
r0.merge(r1)    # r0.counts = [2, 1], value = 3
r1.merge(r0)    # r1.counts = [2, 1], value = 3
# Both see value 3 — strong eventual consistency achieved.

The CAP theorem (Brewer's conjecture 2000, proven by Gilbert and Lynch 2002) states that a distributed system can satisfy at most two of the following three properties simultaneously:

• C — Consistency (linearizability): every read receives the most recent write or an error.
• A — Availability: every request receives a response (not an error), though it may not be the most recent write.
• P — Partition Tolerance: the system continues operating despite arbitrary message loss between nodes (network partitions).

The key insight: P is not really optional. Networks do partition. Hardware fails. Even within a data center, network switches fail, cables get cut, packets get lost. A system that cannot tolerate any partition is not a realistic distributed system — it is just a single machine with some network calls. Therefore, the real choice is between CP (consistency + partition tolerance) and AP (availability + partition tolerance).

• CP systems (e.g., ZooKeeper, etcd, HBase, BigTable): During a partition, the minority partition will refuse to serve requests (or return errors) to avoid returning stale data. Correct, but may be unavailable.
• AP systems (e.g., Cassandra, CouchDB, original DynamoDB): During a partition, all nodes continue serving requests, but may return stale or conflicting data. Highly available, but application must handle inconsistency.

CAP is often misapplied. The theorem is about the worst case — a partition is occurring. Most of the time, networks work fine. The PACELC model (Abadi, 2012) extends CAP: Else (when there is no partition), there is a tradeoff between latency and consistency. Spanner accepts higher latency for linearizability; DynamoDB accepts eventual consistency for lower latency. The real engineering decision is not just CAP but PACELC.

Misconception: CAP does not say you must choose exactly two. You can offer linearizability within a data center (CP) while also offering availability for reads from a read replica (AP) for different operation types.

# CAP in practice: ZooKeeper (CP) vs Cassandra (AP)
#
# ZooKeeper: CP
# - Leader election via ZAB protocol (Zookeeper Atomic Broadcast)
# - Reads/writes go through leader; majority quorum required
# - During partition: minority partition REFUSES requests
#   (throws ConnectionLossException)
#
# Cassandra: AP (tunable)
# - No single leader; any node can handle any request
# - During partition: both sides continue serving
#   possibly returning stale data
# - Consistency level is tunable per-query:

# In CQL (Cassandra Query Language):
#   CONSISTENCY ONE    -- AP: returns first response
#   CONSISTENCY QUORUM -- CP-ish: majority must agree (W+R > RF)
#   CONSISTENCY ALL    -- max consistency, min availability

# The 'replication_factor' and chosen consistency level determine the tradeoff:
# RF=3, W=QUORUM(2), R=QUORUM(2): W+R=4 > RF=3 => linearizable for this partition key
# RF=3, W=ONE(1),   R=ONE(1):     W+R=2 < RF=3 => eventual consistency only
pass

Between sequential consistency and eventual consistency sits causal consistency — a model that enforces ordering for causally related operations but allows different processes to see concurrent operations in different orders.

Causal ordering (Lamport, 1978): Operation A causally precedes operation B (written A → B) if:

1. A and B are in the same process and A occurred before B in program order, OR
2. A is a write and B is a read that returned A's value, OR
3. There exists C such that A → C and C → B (transitivity).

Operations that are not causally related are called concurrent. A causally consistent system guarantees that if A → B, every process sees A before B. Concurrent operations can be seen in any order.

Vector clocks (Fidge/Mattern, 1988) implement causal tracking. Each process maintains a vector V of n counters (one per process). When process i sends a message, it increments V[i] and attaches the vector. The receiver takes component-wise max and increments its own component.

Two events V1 and V2 are causally ordered (V1 → V2) iff V1[k] ≤ V2[k] for all k and V1 ≠ V2. They are concurrent if neither dominates the other.

Practical use: DynamoDB uses vector clocks (called "version vectors") to detect conflicting writes. When two concurrent writes occur (neither causally precedes the other), DynamoDB returns both versions and lets the client (or a user-defined merge function) resolve the conflict. This is the "shopping cart" bug — DynamoDB would return two cart states and merge them by union, sometimes adding back items the user had deleted.

# Vector clock implementation
from copy import deepcopy

class VectorClock:
    def __init__(self, process_id, num_processes):
        self.pid = process_id
        self.clock = [0] * num_processes

    def tick(self):
        """Increment own component on internal event or send."""
        self.clock[self.pid] += 1
        return deepcopy(self.clock)

    def receive(self, remote_clock):
        """Merge on message receive, then tick own component."""
        self.clock = [max(a, b) for a, b in zip(self.clock, remote_clock)]
        self.clock[self.pid] += 1

    def happens_before(self, other_clock):
        """Does self causally precede other?"""
        return (all(a <= b for a, b in zip(self.clock, other_clock))
                and self.clock != list(other_clock))

    def concurrent_with(self, other_clock):
        """Neither causally precedes the other."""
        return (not self.happens_before(other_clock)
                and not VectorClock._dominates(other_clock, self.clock))

    @staticmethod
    def _dominates(vc1, vc2):
        return all(a <= b for a, b in zip(vc1, vc2)) and vc1 != vc2

# P0 sends to P1, P1 sends to P2 — causal chain
p0 = VectorClock(0, 3)
p1 = VectorClock(1, 3)
p2 = VectorClock(2, 3)

sent = p0.tick()          # p0: [1,0,0]
p1.receive(sent)          # p1: [1,1,0]
sent2 = p1.clock[:]       # p1 sends: [1,1,0]
p2.receive(sent2)         # p2: [1,1,1]

print(p0.happens_before(p2.clock))  # True: [1,0,0] -> [1,1,1]

Quorum reads/writes and primary-backup are protocols, but consensus algorithms are the theoretical foundation that makes them safe. Raft (Ongaro & Ousterhout, 2014) is the most teachable consensus algorithm and underlies etcd, CockroachDB, TiKV, and many other production systems.

Raft decomposes consensus into three sub-problems:

1. Leader Election: Time is divided into terms (monotonically increasing integers). At the start of each term, a leader is elected via randomized timeouts — if a follower doesn't hear from the leader within an election timeout (150–300ms), it starts a new election. A candidate wins by collecting a majority of votes. At most one leader per term.

2. Log Replication: All writes go to the leader. The leader appends the entry to its log and sends AppendEntries RPCs to all followers in parallel. An entry is committed when a majority of nodes have written it to their logs. The leader then applies committed entries to its state machine and notifies followers.

3. Safety: Raft's key invariant: if any server has applied a log entry at index i, no other server will ever apply a different entry at index i. This is guaranteed by the leader completeness property: a candidate can only win an election if its log is at least as up-to-date as a majority of nodes.

Why Raft provides linearizability: Because all writes go through one leader per term, and the leader processes them in order, and clients can only read from the leader (or check leadership with a quorum), the system behaves as a single atomic register.

Raft vs Paxos: Paxos (Lamport, 1998) is the original consensus algorithm, but its description is famously incomplete and underspecified for practical implementation. Raft was designed explicitly for understandability — in user studies, participants understood Raft significantly better than Paxos.

# Raft leader election (simplified pseudocode)
# Each server runs this logic concurrently.

class RaftNode:
    FOLLOWER, CANDIDATE, LEADER = 'follower', 'candidate', 'leader'

    def __init__(self, id, peers):
        self.id = id
        self.peers = peers
        self.current_term = 0
        self.voted_for = None
        self.state = self.FOLLOWER
        self.votes_received = 0

    def on_election_timeout(self):
        """Called when no heartbeat received within timeout."""
        self.state = self.CANDIDATE
        self.current_term += 1
        self.voted_for = self.id  # vote for self
        self.votes_received = 1
        for peer in self.peers:
            peer.request_vote(self.current_term, self.id,
                              self.last_log_index, self.last_log_term)

    def on_vote_received(self, term, granted):
        if term == self.current_term and self.state == self.CANDIDATE:
            if granted:
                self.votes_received += 1
            majority = (len(self.peers) + 1) // 2 + 1
            if self.votes_received >= majority:
                self.state = self.LEADER
                self.send_heartbeats()  # assert leadership

    def on_append_entries(self, term, leader_id, entries):
        if term >= self.current_term:
            self.state = self.FOLLOWER  # step down
            self.current_term = term
            self.reset_election_timer()  # heartbeat received