Replication and Consistency

In the Agreement chapter, replication served a single purpose: fault tolerance. A protocol such as Raft keeps backup copies so that the system survives a crash, and under normal (failure-free) operation the extra replicas are pure overhead, every request still funnels through one leader, which must reach a majority before it can commit. Raft therefore makes a system safer, never faster.

This chapter takes replication seriously as a performance and availability tool, and confronts the price that comes with it. Once clients can read and write more than one copy, the copies can disagree, and the system must define what a read is allowed to return. That definition is a consistency model: a contract between the application and the data store. We will walk the full spectrum of these contracts, from the strongest (linearizability) down to the weakest (eventual consistency), see the protocols that implement each one, and locate the line, the high-availability boundary, below which a replica can keep answering even while cut off from the rest of the world. The chapter closes at the CAP trade-off and the modern distributed databases engineered around it.

1. Why Replicate?#

Fault tolerance is only one of three reasons to keep multiple copies of data.

1.1 The three motivations#

• Fault tolerance (via redundancy). Deal with failures and incorrect behaviour through redundant copies, as in the replicated state machines of the Agreement chapter. Replication provides the illusion of a single machine that never fails.
• Availability. Data may be reachable only intermittently. A local replica on a mobile node supports disconnected operation; and spreading load across replicas keeps data reachable under a traffic spike (the release of a new operating system, say) that would sink a single server.
• Performance. Two distinct gains:
  • Throughput / load balancing: sharing the workload across replicas serves more requests and shortens queues. Replicate a web server to sustain more users.
  • Latency / location awareness: place data close to the user. Processor caches, a local copy on a phone, content delivery networks (CDNs), and geo-replicated datastores are all instances of the same idea.

1.2 The latency problem#

Latency has become the dominant bottleneck of modern distributed systems. Over four decades CPU speed, memory, disk and bandwidth improved by three to five orders of magnitude; latency barely moved, because it is ultimately bounded by the speed of light.

Inspired by Rumble et al., “It’s time for low latency”, HotOS ’11. As every other metric races ahead, latency becomes the limiting factor, and placing data geographically close to users is the main way to fight it.

The physical floor is visible in inter-region round-trip times. On AWS EC2, mean RTTs range from about 22 ms (California ↔︎ Oregon) to about 363 ms (São Paulo ↔︎ Singapore). A multi-round protocol run across distant regions can take on the order of one second, plainly perceptible to a human. Concretely: Netflix stores copies of each title in data centres close to viewers (a CDN), and cloud storage such as OneDrive replicates documents across regions both for safety and for local, low-latency access.

2. The Core Problem: Consistency#

As soon as clients can read and write several replicas, conflicts arise. The archetype is a Git merge conflict: two collaborators edit a file locally and, on push, the copies have diverged. By contrast, a CDN barely has conflicts, because only Netflix writes new content and everyone else reads; a viewer receiving an episode a few seconds late is not a meaningful conflict.

The interesting case is when writes originate from multiple sources. Collaboration platforms (Google Docs, Microsoft 365, Dropbox) replicate documents onto users’ devices, support disconnected editing, and reconcile on reconnection under the assumption that simultaneous edits are infrequent. The system must then answer three questions:

• Consistency across replicas. Changing one replica demands changing all others. What happens when replicas are updated concurrently (write-write and read-write conflicts), and what behaviour do we want when they conflict?
• Scalability vs performance. Keeping replicas consistent can be so expensive that replication degrades performance: for instance, if every write must reach all replicas before any progress is allowed.
• Different requirements. Some applications need a global agreement on order (data-centric); others only need each client’s own experience to be coherent (client-centric).

If we insist on the strongest guarantee, the system behaves exactly as a single machine, i.e. linearizability, there is no escape from a single leader (or something equally expensive). This is the fundamental trade-off, captured by the CAP intuition: under network partitions or delays, one must choose between strong consistency and high availability. Fortunately consistency is not binary; a rich spectrum lies between the extremes.

3. Consistency Models#

A consistency model is a contract between the clients (processes) and the data store: it fixes what guarantees the store gives about the ordering and visibility of operations. The trade-off is one-dimensional and unforgiving:

Stronger guarantees → lower performance, but the developer reasons against a simpler, more intuitive system. Weaker guarantees → higher performance and availability, but more complexity is pushed onto the application.

Models can constrain different dimensions:

• Content-based (numerical bounds): replicas may differ by at most some amount on the value (ticket counts off by ≤ 3). Application- and datatype-specific.
• Time-based (staleness): a replica is at most N minutes or K versions behind (a browser cache revalidating with the origin).
• Order-based: constrain the order in which updates become visible. This is the focus of the course, and it splits in two:
  • Data-centric models fix the valid global orderings of operations across all replicas.
  • Client-centric models fix only what a single, possibly migrating, client observes.

4. Consistency Protocols#

Before the models themselves, it helps to see the three implementation families, because a model’s availability is tied to which family can implement it.

4.1 Single-leader#

One replica is the leader; the rest are followers. Every write goes to the leader, which stores it locally and then forwards it to the followers in the order it chose. A read may go to the leader (followers as pure backup, as in Raft) or, in the systems we focus on, to any replica.

Propagation to followers has three variants:

Synchronous (or semi-synchronous) replication is safer: even if k−1 replicas fail, a copy survives, and lagging followers recover by catch-up. If the leader fails, a new one is elected (failover), which is delicate (followers may be stale, the network may be partitioned) and needs a consensus protocol such as Raft. Single-leader replication underlies PostgreSQL, MySQL, Oracle, SQL Server and MongoDB, typically inside one data centre where low latency makes synchronous replication affordable; the write load can be spread by partitioning and giving each partition its own leader.

• Write-write conflicts: impossible, the leader alone orders all writes.
• Read-write conflicts: still possible, a client may read a stale asynchronous follower and miss a write it just made on the leader.

4.2 Multi-leader#

Several replicas accept writes concurrently and cross-replicate. With no single orderer, write-write conflicts become possible, and how to resolve them depends on the consistency model. Multi-leader is natural in geo-replicated settings where reaching a distant leader would be prohibitively slow, and it works well when conflicts are rare and cheap to reconcile, social networks, where writes are infrequent, one user’s writes tend to hit the same replica, and reordering two comments is harmless. Some databases run single-leader within a data centre and multi-leader across data centres; external tools (Tungsten for MySQL, GoldenGate for Oracle) also provide it.

4.3 Leaderless#

The client contacts several replicas directly for both reads and writes (sometimes via a coordinator acting on its behalf). Conflicts are avoided with quorum-based voting: a write must be accepted by a majority, and a read must contact enough replicas to be sure of seeing the latest version. Amazon Dynamo, Riak and Cassandra are leaderless.

5. High Availability and the CAP Theorem#

This distinction runs through the whole chapter.

High availability is the practical face of the CAP theorem, over Consistency, Availability and network Partition:

In the presence of network failures (P), a system can offer availability (A) or consistency (C), but not both.

In practice partitions are always possible and outside the engineer’s control, so the real choice is perpetual: strong consistency or high availability. The rest of the chapter walks from strong-but-unavailable models down to weak-but-available ones, crossing the availability boundary along the way and ending at models served entirely from a local replica. Crucially, some weak models turn out to be achievable with high availability, that is the payoff for giving up global order.

6. Data-Centric Consistency Models#

Data-centric models phrase the contract in terms of the global order of operations across all replicas: which interleavings of reads and writes are permitted. Ideally every operation would take effect instantaneously and be globally ordered by time, impossible without a single clock, and even ignoring time it takes expensive coordination to agree on order.

A useful mental model, stressed in the exercise sessions: distinguish the middleware from the application. The middleware receives updates from other replicas in whatever order the model guarantees (FIFO or causal); the application then freely chooses which variable to read, and in which order. A read returns the value the middleware currently holds for that variable.

6.1 Sequential consistency#

Two requirements: (1) global agreement, all processes agree on one single history; (2) program order preserved, if a process issues A before B, then A precedes B in the agreed sequence. Operations within a process may not be reordered, all processes see the same interleaving, and, decisively, time does not matter, the agreed sequence need not reflect wall-clock time. Sequential consistency requires global agreement on order, so it is not highly available (it needs blocking coordination). It was born as a memory model for multiprocessors and underpins the Java and C++ memory models.

Sequential consistency is expensive even on one machine#

Consider two Java threads sharing x = 0, y = 0:

int x = 0, y = 0;
// Thread 1          // Thread 2
x = 1;              if (y == 1) {
y = 1;                  // Is x guaranteed to be 1 here?
                    }

Under sequential consistency, y == 1 would force x == 1, because Thread 1 wrote x before y. Java does not guarantee this: reading x == 0 after y == 1 is legal. The reason is that each core has its own cache; shared memory is itself a replicated store seen through those caches, and enforcing sequential consistency would demand a cache-coherency protocol on every access, blocking compiler reordering and adding heavy overhead. Java and C++ therefore make it opt-in: a synchronized block flushes and synchronises caches at its boundaries, and a volatile variable is always read/written past the cache. You pay for consistency only where you need it.

Implementing it: single-leader#

Because sequential consistency needs a single agreed order, it cannot be highly available: two partitioned replicas both accepting writes would produce diverging sequences. The standard implementation is single-leader:

• one leader; all writes go to it, and it propagates updates to followers in its chosen order;
• reads may be served by any replica;
• channels are assumed FIFO (if the leader sends A before B, every follower gets A before B: achieved with sequence numbers on reliable links);
• each process is sticky: it always reads the same replica. This is essential, a process that hopped replicas mid-session could read a fresh value from one and then a stale value from another, breaking the agreed order.

In the worked run above (x starts at 0, A writes 1), C conceptually read before the write took effect and B after; all processes agree on the order R(x)0 → W(x)1 → R(x)1. What matters is the logical sequence everyone observes, not real time. The protocol assumes no failures (failover needs Raft), FIFO channels, sticky processes, and that only the latest state matters (followers apply updates on receipt, no log needed in this simplified model).

Implementing it: leaderless quorum#

Sequential consistency can also be reached without a leader, by having the client (or a proxy) contact several replicas directly. With N replicas, let N_W be the number that must acknowledge a write and N_R the number that must answer a read. Sequential consistency holds if:

$$N_R + N_W > N \quad\text{(read-write overlap)} \qquad N_W > N/2 \quad\text{(write-write exclusion)}$$

• Write-write exclusion ($N_W > N/2$): every write locks more than half the replicas, so two writes cannot hold disjoint majorities, their locked sets overlap and the writes are forced to serialise. Concurrent writes become impossible.
• Read-write overlap ($N_R + N_W > N$): any read touching $N_R$ replicas necessarily includes at least one replica that took part in the latest write, so it cannot miss the newest value.

With N = 12, N_W = 10, N_R = 3 is valid (10 > 6, 13 > 12); N_W = 12, N_R = 1 is the read-optimised ROWA extreme; but N_W = 6, N_R = 7 is invalid because $N_W = N/2$ lets two disjoint groups of six accept different writes. Operators tune the split by workload: small $N_W$ / large $N_R$ favours writes; large $N_W$ / small $N_R$ favours reads. Because replicas can briefly hold stale values, two background mechanisms repair them, read repair (a read that spots a lagging replica pushes it the fresh value) and anti-entropy (a background process that periodically compares replicas and fills gaps even with no client activity). Like single-leader, the quorum protocol needs blocking synchronisation (a write cannot complete without reaching a quorum), so sequential consistency is not highly available under either implementation.

6.2 Linearizability#

Linearizability is stronger than sequential consistency: it adds a real-time constraint.

Also called strong, external or atomic consistency, it is the strongest guarantee under replication: fundamentally a recency guarantee, once a client’s write completes, all clients see its effect, giving the illusion of a single copy despite physical replication. It needs no global clock: each operation has a start (issue) and end (response), and the protocol may take as long as it likes within that interval to lock replicas and propagate; once the interval closes, the new state must be visible to everyone.

Reusing the single-leader run (x starts at null, A writes a): the order R(x)null → W(x)a → R(x)a is a valid sequential history. It is not linearizable if, at the same real-time instant, B reads null while C reads a, two clients simultaneously disagreeing on the current value, which linearizability forbids.

To make the single-leader protocol linearizable, add two-phase locking:

1. Lock and propagate. The leader locks all replicas and sends the value; replicas hold it but do not yet serve it. Reads arriving now are deferred.
2. Unlock. Once all replicas have acknowledged synchronously, the leader unlocks and they begin serving the new value.

This closes the window in which some replicas have the value and others do not, so the write looks instantaneous, at the cost of two communication rounds instead of one and reads blocked during a write.

Linearizability is strictly stronger than sequential consistency, which is already unavailable, so it is not highly available either.

6.3 Causal consistency#

Causal consistency is the first model below the availability line: it is the strongest model still implementable in a highly available way.

The intuition is a group chat. Each phone is a local replica; messages are writes, reading the chat is a read.

• Causally related. P1 posts “Distributed systems are the best.” P2 reads it and replies “No way, too complex.” If P3 sees the reply before the original, the conversation is nonsense: P2’s message is a consequence of reading P1’s, so the two are causally related and everyone must see them in the same order.
• Concurrent. P1 posts about a new MacBook while P2, independently, posts about enjoying replication. Neither replies to the other, so P3 and P4 may see them in either order.

This is exactly the happens-before relation from the Synchronization chapter, and it can be read through the same bridge used there: a write and the read that observes it behave like a message sent and received, so “P2 read P1’s write, then wrote” is the same shape as “P2 received P1’s message, then sent one.”

Why it is highly available is best seen through airplane mode: switch your phone offline and keep reading and writing your local chat. While disconnected you cannot receive anyone’s messages, so you cannot causally reply to them, all your offline writes are concurrent with everyone else’s. On reconnection the system reconciles everything while preserving causal order; no global agreement is needed. The one assumption is that clients do not migrate between replicas: hopping replicas could carry causal knowledge from one to another that the second is unaware of. (Migration can be supported with extra metadata, see the client-centric models.)

Implementation: vector clocks. Every write is stamped with a vector clock recording what its writer knew (the latest write it had seen from every other process). A replica applies an incoming write only once it has applied all writes that causally precede it, as the attached vector indicates; out-of-order writes are buffered. Scalar clocks are insufficient because causal dependencies span multiple processes, you must track what each writer knew about every other writer. Causal consistency fits multi-leader naturally: each node accepts writes locally, stamps them, and propagates, withholding dependent updates until their prerequisites arrive.

6.4 FIFO consistency#

FIFO consistency is the weakest data-centric model here. It guarantees exactly one thing:

If channels are already FIFO, the model comes for free; otherwise each process tags its writes with a per-sender sequence number, and a replica buffers out-of-order writes until the gaps fill. FIFO is trivially highly available: a process only orders its own writes, so it can read and write while fully disconnected, and per-sender sequence numbers restore order on reconnection.

6.5 Eventual consistency#

Eventual consistency is weaker still, it does not predicate on the order of operations at all.

Two replicas may apply the same two writes in different orders and, for a while, hold different states (if x = 1 and x = 2 both propagate, some replicas end at 1, others at 2, depending on arrival order). Used naively this is dangerous, but combined with the right data types it is safe and useful. It fits read-heavy workloads (few conflicting writes), applications that tolerate temporary divergence (a slightly out-of-order social feed), geo-distributed multi-leader deployments, and cases where simplicity matters, no vector clocks, quorums or locks, just propagate and converge.

CRDTs (conflict-free replicated data types) are the key mechanism. A CRDT is engineered so that all concurrent updates merge deterministically, regardless of arrival order, always yielding the same final state. The trick is to propagate the operation rather than its result: if operations are commutative and associative, order does not matter. A counter starting at 2 receiving +4 and +2 reaches 8 either way. An append-only list (social-media comments) tags each element with a unique, roughly time-ordered identifier (e.g. a 64-bit timestamp plus node id) and inserts in identifier order, so all replicas converge to the same list, the Last-Writer-Wins (LWW) strategy when the identifier also decides conflicts on the same key. The costs: temporary inconsistency is visible (a message may reorder when a delayed update lands, fine for social media, unacceptable for a bank balance); a hard-limited counter (ticket sales) may briefly oversell by a few before converging; and anything requiring strict ordering or invariants needs a stronger model.

7. Client-Centric Consistency Models#

Every model so far was data-centric and assumed sticky clients bound to one replica. Client-centric models drop that assumption and ask a narrower question: what is a single, migrating client guaranteed to observe as it moves between replicas, regardless of what other clients see?

The diagrams change accordingly. There is one client; the x-axis is its own clock (left is earlier); each row (L1, L2, …) is a replica location it connects to at different times. WS(x_1) is the write set known at L1; WS(x_1; x_2) means L2 knows everything L1 knew plus writes made at L2.

The four properties, each a different ordering across a migration:

• Monotonic reads (read → read). After reading a value of X, any later read of X returns the same or a newer value: never an older one. The client must not “go back in time”: if it read an update at L1 and migrates to L2, L2 must already hold at least what L1 held.
• Monotonic writes (write → write). A client’s write of X completes before its next write of X. If it writes at L1 then migrates to L2 and writes again, L2 must first have received the earlier write, so the second is layered on top.
• Read your writes (write → read). After a client writes X, its every later read of X reflects that write (or a newer one). The client always sees its own writes, even after migrating.
• Writes follow reads (read → write). If a client reads X and then writes X, the write lands on replicas at least as up-to-date as the one it read from, so a write based on something the client read cannot be applied on top of an older state: it prevents “writing into the past.”

7.1 Implementing client-centric consistency#

When a client migrates, the new replica knows nothing of its history, so that context must travel with the client, exactly like web cookies or tokens, and in keeping with stateless (REST) servers that keep no per-client state. The client maintains two lightweight sets, on the client side for scalability:

• a read set: identifiers of the latest writes it has observed through reads;
• a write set: identifiers of the writes it has performed itself.

The protocol is then simple: every write gets a unique id from the store; the client records each id it encounters; on migration it presents both sets to the new replica; the replica serves the request immediately if it has already applied everything in the sets, otherwise it blocks (or signals a retry) until background propagation catches it up. Note these are sets of write identifiers, not vector clocks, kept minimal by dropping ids that later writes overwrite.

8. The Consistency Spectrum#

From strongest to weakest, with the high-availability boundary falling between sequential and causal consistency:

The implication chain Linearizable ⟹ Sequential ⟹ Causal ⟹ FIFO is the workhorse for reasoning: proving a schedule sequential proves it causal and FIFO; proving it violates FIFO proves it violates all three. These models govern individual read/write operations; grouping operations into transactions adds a further isolation layer, left to later courses.

9. Replication Design Strategies#

Beyond the consistency model, a replicated system makes three further design choices.

Replica placement. Permanent replicas are configured at deploy time (traditional databases, CDN edge nodes); server-initiated replicas are spawned or removed by load (cloud auto-scaling for a Black-Friday spike); client-initiated replicas appear on demand from access patterns (browser and CDN caches).

What to propagate.

Propagating operations saves bandwidth but the receiver must re-execute them (CPU cost) and they must be deterministic: any call to a non-deterministic function (current time, a random number) yields different results on different replicas, so such side effects must be resolved before propagation (evaluate time() once and ship the result).

How to propagate: push vs pull.

A common hybrid pushes a lightweight notification on a new version and lets the client pull the data only when it reads, minimal bandwidth when reads are rare, while keeping clients aware of staleness. The push/pull choice combines with the protocol family: leader-based propagation is synchronous, asynchronous or semi-synchronous (as in §4.1); leaderless replicas are reconciled by read repair and anti-entropy (as in §6.1), whose replica-to-replica exchange can itself be push- or pull-based.

10. ACID, CAP and Distributed Databases#

Traditional relational databases offer ACID guarantees, and in a distributed setting each property maps to a coordination protocol:

All four require coordination, hence synchronisation, so they all sit on the non-highly-available side of CAP. Since partitions are always possible, the practical choice is perpetual, consistency (strong guarantees, may block during a partition) or availability (always responds, guarantees relaxed), and it recurs independently across atomicity, isolation and replication.

11. Summary#

12. Exam questions#

Consistency exercises are a recurring exam item, and they are mechanical: with a disciplined method they are almost impossible to get wrong. The solutions below are unofficial worked solutions, the professor does not publish official ones; where the reasoning was confirmed in class (the Q&A and exercise sessions) that is noted.

12.1 Method#

Read a schedule as writes and reads on a replicated store, one row per process, each row in program order. FIFO, causal and sequential do not predicate on time, so only per-row order and read values matter. Since every value is written by exactly one operation, each read is unambiguously tied to its write.

A reliable recipe:

1. Write the constraints. For FIFO, list each process’s writes in order (across variables). For causal, add an edge from a value read by a process to any write that process makes afterward (read → later write), and take the transitive closure with the FIFO chains. The result is a partial order on the writes.
2. Order of attack. Because sequential ⟹ causal ⟹ FIFO, if a schedule is sequential it satisfies all three; if it fails FIFO it fails all three. Often the quickest route is to test the strongest first (try to build one global sequence) and the weakest as a fallback.
3. To prove consistency, exhibit a full interleaving that works. To disprove it, show that every legal prefix leads to a contradiction: usually a read that cannot return its value.
4. The “no going back” rule. Once a reader has observed a value on a variable, it cannot later read a value that was overwritten before it. Across variables the same holds through causal chains: having read y=3 that was written after x=5, a later R(x) can no longer return an older x.
5. Middleware vs application. The middleware receives updates in FIFO/causal order; the application then reads variables in whatever order it likes. A read returns the value currently held for that variable: so a R(y) between two R(x) does not “reset” x.
6. The inference trap. A schedule that is consistent with a model proves nothing (like a passing test), but a schedule that violates a model proves the store does not implement it. So the honest conclusion is usually “the store is not X; it may be Y.”

12.2 Is the schedule FIFO / causal / sequential?#

P0:  W(x)2   W(x)5   R(x)4   W(y)3
P1:  W(y)1   W(x)4   R(y)1   R(y)3   R(x)4
P2:  R(y)1   R(x)4   R(x)5   R(y)3

            June 18                         July 12
P2:  R(y)1 R(x)2 R(y)3 R(x)5 R(x)4    R(y)1 R(x)2 R(y)3 R(x)4 R(x)5

P0:  W(y)1   W(x)2   R(y)3   W(x)6   R(y)5
P1:  R(y)1   W(y)3   R(x)2   R(x)4   R(y)5
P2:  R(x)6   W(y)5   R(x)6   W(x)4   R(x)2

12.3 Which values may be read? (the “?” variant)#

P0:  W(x)2   W(x)5   R(x)4   W(y)3
P1:  W(y)1   W(x)4   R(y)1   R(y)3   R(x)4
P2:  R(y)3   R(x)4   ?

12.4 Older schedule (five processes)#

P0:  W(x)2   R(y)2   W(x)3   R(x)3
P1:  W(x)1   R(x)1   W(y)2   W(y)3
P2:  R(x)0   W(y)1   R(y)1   R(x)2
P3:  R(x)2   R(y)2   R(x)2   R(y)3
P4:  R(x)3   R(y)2   R(y)3

R(x)0 · W(x)1 · R(x)1 · W(y)1 · R(y)1 · W(x)2 · R(x)2 · R(x)2 · W(y)2 ·
R(y)2 · R(y)2 · R(x)2 · W(x)3 · R(x)3 · R(x)3 · R(y)2 · W(y)3 · R(y)3 · R(y)3

12.5 Building on a weaker store#

13. Glossary#

Sources: van Steen & Tanenbaum, Distributed Systems* (3rd ed., 2017); Kleppmann, Designing Data-Intensive Applications (2017); Viotti & Vukolić, “Consistency in Non-Transactional Distributed Storage Systems”, ACM CSUR 2016; Bailis et al., “Highly Available Transactions”, VLDB 2014; Lamport (1979) on sequential consistency; Corbett et al. (Spanner, OSDI 2012); Thomson et al. (Calvin, SIGMOD 2012).*

Compiled from G. Cugola's lectures (A.Y. 2025/26), slides + class transcriptions. Not official course material.