Quorum replication

Deep dive

How it works

Parts:

• ring and preference list — maps a key to its N replicas (consistent hashing)
• coordinator — any node; fans out requests, counts W or R answers
• replica store — versioned values and tombstones per key
• version tag — timestamp plus writer id, or a vector clock
• repair paths — read repair, hinted handoff, Merkle-tree anti-entropy

Operations: write, read, read repair, hinted handoff, anti-entropy (steps below).

1. Map key → N replicas (preference list) with consistent hashing.
2. Write: coordinator tags the value with a version (timestamp + writer id, or a vector clock), sends to all N, returns after W acks.
3. Read: send to N (or R plus speculative retries), wait for R, return the newest; resolve or return siblings if versions are concurrent.
4. Read repair: write the newest version back to replicas that returned stale data (synchronously for linearizable reads, else in the background).
5. Hinted handoff: if a replica is down, another node stores a hint and replays it later (sloppy quorum: availability up, intersection guarantee gone).
6. Anti-entropy: replicas periodically compare Merkle trees (hash trees) of key ranges and stream the differing ranges.

         write(k, v@t)                   read(k)
coord ──► A ✓  B ✓  C ✗(slow)   coord ──► A v@t   C v@t-1   (B not needed)
         W=2 reached → ok                 R=2 → return v@t, repair C

Code

use std::cell::{Cell, RefCell};
use std::collections::HashMap;

#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct Version { pub ts: u64, pub writer: u32 }   // ts from an HLC; writer breaks ties

#[derive(Clone, Debug)]
pub struct Versioned { pub ver: Version, pub value: Option<Vec<u8>> } // None = tombstone

pub struct Replica { pub up: Cell<bool>, data: RefCell<HashMap<String, Versioned>> }

impl Replica {
    pub fn new() -> Self { Replica { up: Cell::new(true), data: RefCell::new(HashMap::new()) } }
    fn get(&self, k: &str) -> Option<Option<Versioned>> {      // outer None = no answer
        self.up.get().then(|| self.data.borrow().get(k).cloned())
    }
    fn put(&self, k: &str, v: &Versioned) -> bool {
        if !self.up.get() { return false; }
        let mut d = self.data.borrow_mut();
        let newer = d.get(k).map_or(true, |cur| v.ver > cur.ver); // apply only if newer
        if newer { d.insert(k.to_string(), v.clone()); }
        true
    }
}

pub struct Coordinator<'a> { pub id: u32, pub replicas: Vec<&'a Replica>, pub r: usize, pub w: usize }

impl<'a> Coordinator<'a> {
    // Real systems send to all replicas in parallel and return on the W-th ack.
    pub fn write(&self, k: &str, value: Option<Vec<u8>>, ts: u64) -> Result<(), usize> {
        let v = Versioned { ver: Version { ts, writer: self.id }, value };
        let acks = self.replicas.iter().filter(|r| r.put(k, &v)).count();
        // A failed write may still be visible later: it landed on `acks` replicas.
        if acks >= self.w { Ok(()) } else { Err(acks) }
    }

    /// linearizable = true: ABD write-back so no later read can see an older value.
    pub fn read(&self, k: &str, linearizable: bool) -> Result<Option<Vec<u8>>, ()> {
        let mut answers: Vec<(&Replica, Option<Versioned>)> = Vec::new();
        for r in &self.replicas {
            if let Some(v) = r.get(k) { answers.push((*r, v)); }
            if answers.len() == self.r { break; }
        }
        if answers.len() < self.r { return Err(()); }
        let newest = answers.iter().filter_map(|(_, v)| v.clone()).max_by_key(|v| v.ver);
        let Some(newest) = newest else { return Ok(None) };
        let mut fresh = 0;
        for (rep, v) in &answers {
            let stale = v.as_ref().map_or(true, |v| v.ver < newest.ver);
            if !stale { fresh += 1; } else if rep.put(k, &newest) { fresh += 1; } // read repair
        }
        if linearizable {
            for r in self.replicas.iter().filter(|r| !answers.iter().any(|(a, _)| std::ptr::eq(*a, **r))) {
                if fresh >= self.w { break; }
                if r.put(k, &newest) { fresh += 1; }
            }
            if fresh < self.w { return Err(()); }
        }
        Ok(newest.value)
    }
}

Complexity and constants

• One round trip for reads and writes, to the R-th / W-th fastest replica. Latency is an order statistic: with N = 3, W = 2 the slowest replica never matters.
• Messages: N per operation (send to all), or R plus speculative retries (Cassandra default speculative_retry = 99p).
• Storage: N × data, plus hints and tombstones.
• Staleness: with R + W ≤ N (e.g. R = W = 1) reads can be stale; PBS (Bailis et al.) predicts how stale from measured replica latency distributions, so measure instead of guessing.

Efficiency

• Cost: one round trip per read or write, to the R-th or W-th fastest replica; with N = 3, W = 2 the slowest replica never matters. Messages per operation: N (send to all), or R plus speculative retries (Cassandra default speculative_retry = 99p). Storage is N x data plus hints and tombstones. Bound: network round trip and replica tail latency.
• Levers:
  • speculative retry on a slow replica — cuts tail latency to the next-fastest replica (Cassandra docs) — read-heavy, latency-sensitive loads
  • R = W = 1 where stale reads are fine — fewest waits, but R + W <= N can be stale; measure with PBS (Bailis et al.) — caches, metrics
  • read at R, write at W per request — trade latency against consistency per operation — mixed workloads
  • background read repair and Merkle-tree anti-entropy — repair cost off the request path — long-lived clusters
  • hinted handoff — writes keep succeeding while a replica is down, at the price of the intersection guarantee — brief outages
• Naive → standard → state of the art:
  • write all N, read all N · latency of the slowest replica
  • quorum R + W > N · latency of the R-th or W-th fastest of N
  • quorum plus speculative retry, tuned by PBS predictions · tail set by the next-fastest replica

Implementation notes

• Versions: use hybrid logical clocks or server-assigned timestamps; never raw client wall clocks for LWW.
• For multi-writer correctness keep siblings: vector clocks / dotted version vectors (Riak) or CRDTs; with LWW, accept that concurrent writes drop.
• Deletes are tombstones; purge only after all replicas have seen them (repair cadence < gc_grace_seconds), or deleted data resurrects.
• Rack/zone-aware placement: the N replicas must sit in different failure domains, else one rack kills the quorum.
• Use LOCAL_QUORUM (quorum within the local DC) to keep cross-region latency off the critical path.

Pitfalls

• Believing R + W > N gives linearizability: a reader can see a new value, a later reader an older one while a write is in flight or after a partial failed write. Use ABD write-back.
• Sloppy quorums (hinted handoff) break the R/W overlap during failures.
• A "failed" write (fewer than W acks) is not rolled back; it may surface later.
• LWW with skewed clocks: the writer with the fast clock silently wins for the size of the skew.
• Read repair alone never fixes keys nobody reads: run anti-entropy.

Variants and alternatives

In production

Shopping carts and key-value stores: Amazon Dynamo (internal), Riak KV, Voldemort. Wide-column and time-series databases: Apache Cassandra, ScyllaDB. Object storage: OpenStack Swift (its proxy server counts replica answers against a quorum size). (Amazon DynamoDB, despite the name, replicates each partition with Multi-Paxos.)

Implementations

Sources

• Giuseppe DeCandia et al., Dynamo: Amazon's Highly Available Key-value Store, SOSP 2007. paper
• David K. Gifford, Weighted Voting for Replicated Data, SOSP 1979. paper
• Hagit Attiya, Amotz Bar-Noy, Danny Dolev, Sharing Memory Robustly in Message-Passing Systems, JACM 1995. paper
• Peter Bailis et al., Probabilistically Bounded Staleness for Practical Partial Quorums, VLDB 2012. paper
• Apache Cassandra documentation, Dynamo (architecture, consistency levels). docs
• OpenStack Swift documentation, Proxy (quorum_size). docs