Quorums and Replication Protocols

A quorum is the minimum number of replicas that must participate in an operation for it to be considered successful. In leaderless replicated stores such as Dynamo, Cassandra, and Riak, quorums replace a single leader: a client or coordinator sends each operation to all replicas and waits for a subset to respond. Choosing the size of that subset controls the trade-off between consistency, latency, and availability. Background mechanisms then repair replicas that missed updates.

Quorum parameters

Each key is stored on N replicas. A write is acknowledged after W replicas confirm it. A read waits for responses from R replicas and returns the newest value among them.

• N: replication factor.
• W: write quorum.
• R: read quorum.

If R + W > N, every read quorum and every write quorum share at least one replica. The read therefore contacts at least one replica that has the latest acknowledged write, and picking the newest version returns it.

graph LR
    subgraph "N = 5 replicas"
        R1["replica 1"]
        R2["replica 2"]
        R3["replica 3"]
        R4["replica 4"]
        R5["replica 5"]
    end
    W["write, W = 3"] --> R1
    W --> R2
    W --> R3
    Rd["read, R = 3"] --> R3
    Rd --> R4
    Rd --> R5

Here the write reached replicas 1 to 3 and the read reached replicas 3 to 5. Replica 3 is in both sets, so the read sees the write.

Worked example with N = 3, W = 2, R = 2: key x holds version 1 on all replicas. A write of version 2 is acknowledged by replicas A and B, and replica C has not received it yet. A read contacts B and C and receives version 2 and version 1. The coordinator compares versions, returns version 2, and (through read repair) sends version 2 to C. A read that contacts A and C or A and B also includes a replica with version 2. No pair of two replicas out of three can consist of C alone, so every read pair includes A or B.

Typical configurations

Cassandra names these levels ONE, QUORUM (majority, floor(N/2) + 1), and ALL. LOCAL_QUORUM limits the quorum to the local data center to avoid cross-region latency.

What quorums do not guarantee

R + W > N is not linearizability. Cases where a read can still return stale or inconsistent data:

• Concurrent writes. Two writes to the same key with no defined order require conflict resolution (last-write-wins or version vectors, see time, clocks, and ordering). Last-write-wins with timestamps can discard a write when clocks are skewed.
• Read concurrent with a write. A write that has reached fewer than W replicas is not yet acknowledged. A reader may see it from one replica and another reader may not.
• Failed writes. A write that reached fewer than W replicas returns an error, but the replicas that received it keep it. Later reads can return the value from a write the client was told failed.
• Sloppy quorums (below) remove the overlap guarantee.
• Restored replicas. If a replica holding a new value fails and is restored from an older copy, the count of replicas holding the new value can fall below W.

Making single-key operations linearizable requires additional steps: readers perform read repair synchronously before returning, writers read the latest version before writing, and conflicts are not resolved by last-write-wins with timestamps. Alternatively, a consensus protocol (Raft) per key range can be used (Cassandra lightweight transactions, for example, use Paxos for conditional updates).

The overlap condition can be checked by enumerating all subsets. Each subset of replicas is a bitmask, and two sets overlap when the bitwise AND is nonzero:

import java.util.ArrayList;
import java.util.List;

public class Main {
    static List<Integer> subsets(int n, int k) {
        var result = new ArrayList<Integer>();
        for (int mask = 0; mask < (1 << n); mask++) {
            if (Integer.bitCount(mask) == k) {
                result.add(mask);
            }
        }
        return result;
    }

    static boolean alwaysOverlap(int n, int w, int r) {
        var readSets = subsets(n, r);
        for (int writeSet : subsets(n, w)) {
            for (int readSet : readSets) {
                if ((writeSet & readSet) == 0) {
                    return false;
                }
            }
        }
        return true;
    }

    public static void main(String[] args) {
        int[][] cases = {{3, 2, 2}, {3, 1, 3}, {3, 1, 1}, {5, 3, 3}, {5, 2, 3}};
        for (int[] c : cases) {
            int n = c[0], w = c[1], r = c[2];
            System.out.println("N=%d W=%d R=%d overlap=%b (R+W>N: %b)"
                    .formatted(n, w, r, alwaysOverlap(n, w, r), r + w > n));
        }
    }
}

N=3 W=2 R=2 overlap=true (R+W>N: true)
N=3 W=1 R=3 overlap=true (R+W>N: true)
N=3 W=1 R=1 overlap=false (R+W>N: false)
N=5 W=3 R=3 overlap=true (R+W>N: true)
N=5 W=2 R=3 overlap=false (R+W>N: false)

Sloppy quorums and hinted handoff

In a strict quorum, W and R count only the N replicas that own the key. When some of those replicas are unreachable, writes may fail even though other nodes in the cluster are healthy.

A sloppy quorum counts the first W healthy nodes reachable, even if some are not among the key's designated replicas. Availability for writes increases. The price is that a read quorum may not overlap with the nodes holding the write.

Hinted handoff completes the mechanism:

1. A replica C for a key is unreachable.
2. The coordinator writes the data to a stand-in node D, with a hint naming C as the intended owner.
3. D stores hinted data separately and periodically checks whether C is back.
4. When C recovers, D forwards the data and deletes its copy.

sequenceDiagram
    participant Co as Coordinator
    participant A as Replica A
    participant B as Replica B
    participant C as Replica C (down)
    participant D as Stand-in node D
    Co->>A: write
    Co->>B: write
    Co->>C: write (no response)
    Co->>D: write with hint for C
    A-->>Co: ack
    B-->>Co: ack
    D-->>Co: ack (W met)
    Note over C: C recovers
    D->>C: hinted data
    C-->>D: ack, D deletes hint

Hints are stored for a limited time. Cassandra, for example, stops storing hints for a node that has been down longer than a configured window, so a node down longer than that needs a repair. Hinted handoff improves durability and write availability, but it is a best-effort optimization, not a guarantee of convergence.

Read repair

Read repair corrects stale replicas during reads.

1. The coordinator sends the read to several replicas (at least R, sometimes all).
2. It compares the versions returned, using timestamps or version vectors.
3. It returns the newest value to the client.
4. It writes the newest value to replicas that returned older versions.

Read repair only repairs data that is read. Rarely read keys stay inconsistent until another mechanism fixes them. Some systems compare digests instead of full values on the extra replicas to reduce network traffic, and fetch the full data only on mismatch.

Anti-entropy with Merkle trees

Anti-entropy is a background process that compares replicas and synchronizes differences, independent of client reads. Comparing all keys would require transferring the whole dataset. A Merkle tree (hash tree) reduces the transfer.

• Leaves hold hashes of individual keys (or small key ranges).
• Each internal node holds the hash of its children's hashes.
• The root hash summarizes the entire dataset.

graph TD
    Root["root = H(H12, H34)"] --> H12["H12 = H(h1, h2)"]
    Root --> H34["H34 = H(h3, h4)"]
    H12 --> h1["h1 = H(key range 1)"]
    H12 --> h2["h2 = H(key range 2)"]
    H34 --> h3["h3 = H(key range 3)"]
    H34 --> h4["h4 = H(key range 4)"]

Two replicas exchange root hashes. If they match, the data is identical. If not, they compare children and descend only into subtrees whose hashes differ, until they find the differing key ranges. Only those ranges are transferred. The cost is proportional to the number of differences, plus O(log n) hash comparisons per difference, rather than to the data size.

Building the tree requires reading the data, so it is done in the background and often per token range. Dynamo, Cassandra (nodetool repair), and Riak use Merkle trees this way. Deleted data needs special handling: a tombstone marker replicates the deletion, and it must be kept longer than the maximum time a replica can be out of sync. Otherwise a repaired stale replica can resurrect deleted data.

Mechanisms compared

A production configuration uses all three: hinted handoff for short outages, read repair for hot data, and anti-entropy for everything else.

Leader-based replication for contrast

Leader-based protocols (leader-follower, or a consensus group) replicate through one ordered log. All writes pass through the leader, and followers apply the log in order. This gives a single order of writes and simpler conflict handling, but writes depend on the leader, and failover is required when it fails. Leaderless quorums remove the leader as a single point, at the cost of conflict resolution and repair machinery. See database replication for leader-based designs.

Interview questions

• Why does R + W > N give overlap? A read set and a write set that together contain more than N replicas must share at least one, so a read always includes a replica that acknowledged the latest write.
• Does R + W > N guarantee linearizability? No. Concurrent writes, partially applied failed writes, sloppy quorums, and clock-based conflict resolution can still return stale or inconsistent values.
• What does a sloppy quorum trade away? It accepts writes on nodes outside the key's replica set to stay available, so a later read quorum may miss them until hinted handoff delivers the data.
• Why are Merkle trees used for anti-entropy? Comparing hashes top-down finds differing key ranges with few comparisons and transfers only those ranges instead of the full dataset.
• Why must tombstones be retained for some time? A replica that missed the delete may still hold the old value. Without a tombstone, repair would copy that value back to the other replicas.