Chapter 05 - Replication - Notes
Chapter 05 - Replication - Notes
Replication means keeping a copy of the same data on multiple machines connected via a
network. It is one of the two main strategies (the other being partitioning, Ch. 6) for
distributing data.
Why Replicate?
High availability — keep the system running even if a machine (or whole datacenter)
goes down.
Disconnected / offline operation — keep working through network interruptions.
Latency — place data geographically close to users for faster interaction.
Scalability — serve a higher volume of reads than one machine could handle, by
spreading reads across replicas.
The hard part isn’t storing copies — it’s handling changes to replicated data. There are
three main approaches: single-leader, multi-leader, and leaderless replication. Almost all
distributed databases use one of these.
One replica is designated the leader (master). All writes go to the leader.
Other replicas are followers (read replicas / slaves). They receive the leader’s stream
of data changes (the replication log / change stream) and apply them in the same
order.
Reads can be served by the leader or any follower; writes only by the leader.
Used by PostgreSQL, MySQL, Oracle Data Guard, SQL Server AlwaysOn, MongoDB,
RethinkDB, Espresso — and beyond databases: Kafka, RabbitMQ HA queues, DRBD.
Synchronous vs. Asynchronous Replication
Synchronous Asynchronous
If sync follower is
Risk down/slow, all writes Weaker durability
block
Making all followers synchronous is impractical — any single node outage halts all
writes.
Semi-synchronous: usually one follower is synchronous, the rest asynchronous. If
the sync follower becomes slow/unavailable, an async one is promoted to sync.
Guarantees an up-to-date copy on at least two nodes.
Fully asynchronous is common (especially with many or geo-distributed followers):
leader keeps processing even if all followers lag, but a confirmed write isn’t guaranteed
durable.
Chain replication is a synchronous variant that avoids data loss with good
performance (used in Microsoft Azure Storage). There’s a deep link between
replication consistency and consensus (Ch. 9).
A plain file copy is insufficient — data is always in flux. Process (usually without downtime /
locking): 1. Take a consistent snapshot of the leader (most DBs support this for backups).
2. Copy the snapshot to the new follower. 3. Follower requests all changes since the
snapshot — requires an exact position in the replication log (PostgreSQL: log sequence
number; MySQL: binlog coordinates). 4. Once the backlog is processed, the follower has
caught up and streams new changes live.
Follower failure → catch-up recovery: Each follower logs changes locally. After a
crash/network blip, it knows the last processed transaction, reconnects, requests the missed
changes, and catches up.
Leader failure → failover: Promote a follower to leader, reconfigure clients to write to it,
make other followers follow it. Can be manual or automatic. Automatic steps: 1. Detect
⇒
failure — usually a timeout (e.g., no response for 30s assumed dead). No foolproof
detection. 2. Choose a new leader — election by majority, or appointment by a controller
node. Best candidate = replica with most up-to-date data (minimize data loss). This is a
consensus problem (Ch. 9). 3. Reconfigure — route writes to the new leader; ensure the
old leader steps down to follower if it returns.
Failover is fraught — things that go wrong: - Lost writes (async): new leader may not
have all old-leader writes; the common fix is to discard unreplicated old-leader writes —
violating durability expectations. - Cross-system inconsistency: discarded writes are
dangerous when coordinated with external stores. GitHub incident: an out-of-date MySQL
follower was promoted; its autoincrement counter lagged, reused primary keys already used
in Redis → private data disclosed to wrong users. - Split brain: two nodes both believe
they’re leader. If both accept writes with no conflict resolution, data is lost/corrupted. Safety
mechanism = fencing / STONITH (Shoot The Other Node In The Head) — but poorly
⇒
designed fencing can shut down both nodes. - Timeout tuning: too long slow recovery;
⇒
too short unnecessary failovers (a load spike or network glitch can trip it), worsening an
already-struggling system.
Because of these hazards, some teams prefer manual failover even when
automatic is supported. These trade-offs (consistency, durability, availability,
latency) are fundamental distributed-systems problems (Ch. 8–9).
stored procs). Workarounds exist but are fragile. MySQL used this pre-5.1; now
defaults to row-based if any nondeterminism. VoltDB uses it but requires deterministic
transactions.
Write-ahead log (WAL) shipping — ship the storage engine’s append-only log (the
same bytes written to disk). Used by PostgreSQL, Oracle. Downside: very low-level
⇒
(which bytes in which disk blocks) tightly coupled to the storage engine and
version. Usually blocks zero-downtime upgrades (can’t run different DB versions on
leader vs. follower).
Logical (row-based) log replication — a separate log format decoupled from the
storage engine, describing writes at row granularity (inserted = all new values; deleted
= key/identifying info; updated = key + changed values), with a commit marker per
transaction. MySQL’s binlog (row-based mode) works this way. Easier to keep
backward-compatible (different versions/engines on leader vs. follower) and easier for
external systems to parse → change data capture (Ch. 11).
Trigger-based replication — move replication into the application layer via DB
triggers/stored procedures (or tools reading the log, e.g., Oracle GoldenGate; Databus
for Oracle; Bucardo for Postgres). More flexible (replicate a subset, cross-DB, custom
conflict logic) but higher overhead and more bug-prone.
Read-scaling architecture: route reads across many followers, writes to the leader. Great
for read-heavy web workloads; add followers to add read capacity. Realistically requires
asynchronous replication (sync to all followers would be too fragile).
But async followers can serve stale data. Run the same query on leader vs. follower and get
different results — an apparent inconsistency. This is temporary: stop writing and followers
catch up → eventual consistency. The replication lag is usually sub-second but can
balloon to seconds/minutes under load or network problems. “Eventually” is deliberately
vague — there’s no upper bound.
Problem: user submits data (to leader), then reads it (from a lagging follower) → looks
like the write was lost.
Read-after-write (read-your-writes) consistency: a user always sees their own
updates after reload. Says nothing about other users’ updates.
Techniques:
Read potentially-self-modified data from the leader (e.g., always read your own
profile from leader, others’ from followers).
If most things are user-editable, use other criteria: read from leader for N minutes
after the user’s last update; or avoid followers lagging > threshold.
Client remembers timestamp of its last write; serve reads only from replicas
caught up to that timestamp (logical timestamp like LSN, or system clock — then
clock sync matters).
Cross-device consistency adds complexity: last-update metadata must be
centralized; different devices may route to different datacenters.
2. Monotonic Reads
Problem: user reads from a fresh follower, then a lagging one → sees data go
backward in time (a comment appears, then disappears).
Monotonic reads: stronger than eventual consistency, weaker than strong
consistency. Guarantees a user, reading in sequence, never sees older data after
newer data.
Technique: each user always reads from the same replica (e.g., chosen by a hash of
user ID). Must reroute if that replica fails.
Think through how the app behaves if lag grows to minutes/hours. If “no problem,” fine;
otherwise design for a stronger guarantee.
Don’t pretend async is sync — that’s “a recipe for problems down the line.”
Handling these issues in app code is complex and error-prone. Transactions exist so
the database provides stronger guarantees and the app stays simple. Many distributed
DBs abandoned transactions claiming they’re too costly — an “overly simplistic” view
(revisited in Ch. 7 & 9).
Multi-Leader Replication
Also master–master / active/active. More than one node accepts writes; each leader is
also a follower to the other leaders. Each write must be forwarded to all other nodes.
Within a single datacenter, multi-leader rarely justifies its added complexity.
Use Cases
Each DC operates
Failover promotes a
DC outage tolerance independently; catches
follower in another DC
up later
Clients with offline operation — e.g., a calendar app across phone/laptop. Each
device has a local DB acting as a leader; async multi-leader sync between devices
(lag can be hours/days). Architecturally identical to multi-DC taken to the extreme.
CouchDB is designed for this.
Collaborative editing — e.g., Google Docs, Etherpad. Local replica updated instantly,
async-replicated to others. Lock the whole doc = single-leader with transactions; small
change units (per keystroke) without locks = multi-leader with conflict resolution.
The biggest problem with multi-leader: the same data modified concurrently on different
⇒
leaders conflicts (doesn’t happen single-leader).
Sync vs. async detection: single-leader blocks/aborts the second write; multi-leader
detects the conflict asynchronously, often too late to ask the user. Making detection
synchronous defeats the point (might as well use single-leader).
Conflict avoidance (most common recommendation): route all writes for a given
record to the same leader (“home” datacenter per user). Breaks down when the
designated leader must change (DC failure, user relocation).
Converging toward a consistent state: there’s no defined write order, so every
scheme must ensure replicas reach the same final value. Methods:
Last write wins (LWW) — pick highest unique ID/timestamp; popular but
dangerously prone to data loss.
Highest-numbered replica wins — also implies data loss.
Merge the values (e.g., concatenate “B/C”).
Record the conflict explicitly and resolve later (possibly prompting the user).
Custom conflict resolution logic — runs on write (background handler, can’t prompt
user; e.g., Bucardo/Perl) or on read (store all conflicting versions, return them on next
read for app/user to resolve; e.g., CouchDB). Usually applies per row/document, not
per transaction.
Automatic conflict resolution research:
CRDTs (Conflict-free Replicated Data Types) — sets/maps/lists/counters that
auto-merge sensibly (Riak 2.0). Two-way merges.
Mergeable persistent data structures — track history like Git; three-way
merge.
Operational transformation — algorithm behind collaborative editors (Google
Docs, Etherpad); designed for concurrent editing of ordered lists (text).
Amazon shopping cart cautionary tale: handler preserved added items but not
removed ones → removed items reappeared.
What is a conflict? Some are obvious (two writes to the same field). Others are subtle
(e.g., double-booking the same meeting room — a constraint violated across leaders
even if each checked availability).
A topology describes the paths writes propagate along: - All-to-all — every leader sends to
every other. Best fault tolerance (multiple paths, no single point of failure), but messages
can overtake each other → causality/ordering problems (an update arrives before its insert).
- Circular — each node forwards to one other (MySQL default). - Star / tree — a root
forwards to all others.
In circular/star, writes pass through several nodes; each write is tagged with the node
identifiers it has passed through to prevent infinite loops (a node ignores writes
already tagged with its own ID).
Circular/star weakness: one failed node interrupts replication flow (manual
reconfiguration usually needed).
All-to-all ordering: timestamps aren’t enough (clocks can’t be trusted to sync); use
version vectors. Many systems implement conflict detection poorly (e.g., PostgreSQL
BDR lacked causal ordering; Tungsten for MySQL didn’t try to detect conflicts). Read
the docs and test thoroughly.
Leaderless Replication
Abandon the leader entirely — any replica accepts writes directly from clients. Revived by
Amazon’s Dynamo; open-source Dynamo-style systems: Riak, Cassandra, Voldemort.
(Note: AWS DynamoDB is different — single-leader.) Clients (or a coordinator node that
doesn’t enforce ordering) send writes to several replicas.
No failover. Client sends each write to all replicas in parallel; succeeds once enough
(e.g., 2 of 3) acknowledge. A down replica simply misses the write.
When the down node returns, it has stale data. So clients also send reads to several
nodes in parallel and use version numbers to pick the newest value.
Catching up missed writes: - Read repair — on a parallel read, the client detects a stale
replica and writes the newer value back to it. Works well for frequently-read values. - Anti-
entropy process — a background process continuously copies missing data between
replicas (no particular order; possibly delayed). Without it, rarely-read values may stay
missing (reduced durability). Voldemort lacks anti-entropy.
With n replicas, every write needs w acknowledgments, every read queries r nodes.
⇒
Quorum condition: w + r > n a read overlaps with the latest write on at least one
node, so you expect an up-to-date value.
Common: n odd (3 or 5), w = r = (n+1)/2.
Tolerance: with n=3, w=2, r=2 → tolerate 1 down node. With n=5, w=3, r=3 → tolerate
2.
Reads/writes go to all n in parallel; w and r just define how many acks to wait for.
Tune for workload: many reads, few writes → w=n, r=1 (fast reads, but one failed node
blocks all writes).
More than n nodes can exist in the cluster, but each value lives on only n nodes
(enables partitioning, Ch. 6).
A strict quorum (w+r>n) contrasts with a sloppy quorum.
Even with w + r > n, stale reads still happen in edge cases: - Sloppy quorum used (no
guaranteed overlap of read/write node sets). - Two concurrent writes — order undefined;
must merge, else LWW can lose writes via clock skew. - A write concurrent with a read
may be reflected on only some replicas — undetermined which value is read. - A write
partially succeeded (< w replicas) isn’t rolled back where it succeeded — later reads may
or may not return it. - A node with a new value fails and is restored from an old replica →
count of replicas with the new value can drop below w. - Unlucky timing (linearizability edge
cases, Ch. 9).
A network interruption can cut a client off from the n home nodes for a value even
though many nodes are alive. Trade-off: error out, or accept writes on other reachable
nodes?
Sloppy quorum: still require w/r successful responses, but allow nodes outside the
designated n “home” nodes (analogy: locked out of your house, sleep on a neighbor’s
couch).
Hinted handoff: once the network recovers, the temporarily-accepted writes are
forwarded to the proper home nodes (neighbor sends you home).
Boosts write availability but breaks the overlap guarantee — even with w+r>n you
can’t be sure to read the latest value until handoff completes. So a sloppy quorum
isn’t really a quorum — just a durability assurance.
Defaults: enabled in Riak; disabled in Cassandra and Voldemort.
Multi-datacenter: leaderless suits multi-DC (tolerates conflicts, interruptions, latency).
Cassandra/Voldemort include all-DC nodes in n but usually wait only for a local-DC
quorum. Riak keeps client↔︎node traffic local to one DC and replicates cross-DC async
(multi-leader-style).
Detecting Concurrent Writes
Conflicts occur even with strict quorums (and during read repair / hinted handoff), because
events arrive in different orders at different nodes. Naïve last-write-wins overwriting →
permanent inconsistency. Replicas must converge, but most implementations are poor —
the application developer must understand the DB’s conflict handling.
Last write wins (LWW): force an arbitrary order via timestamps; discard older.
Cassandra’s only method; optional in Riak. Achieves convergence at the cost of
durability — concurrent writes (all reported successful) silently dropped; can even
drop non-concurrent writes under clock skew. Safe only if keys are written once and
immutable (e.g., use a UUID per write).
Server keeps a version number per key, incremented on each write, stored
with the value.
On read, server returns all non-overwritten values + the latest version number.
A client must read before writing.
On write, the client includes the version number from its prior read and
merges all values it received. (Write responses can also return current values,
allowing chained writes.)
On receiving a write with version V, the server overwrites all values at version ≤
V (already merged) but keeps values with a higher version (those are concurrent
— siblings).
A write with no version number is concurrent with everything → overwrites
nothing.
Result: no data silently dropped, but clients must merge siblings afterward.
Merging siblings: same problem as multi-leader conflict resolution. LWW loses data.
A sensible cart merge = union of items. But deletions break union merges — a
removed item reappears. Fix: don’t truly delete; leave a tombstone (deletion marker
with version number) so merges preserve the removal. CRDTs (e.g., Riak datatypes)
can merge siblings automatically, including deletions.
Version vectors: with multiple replicas (no leader), a single version number is
insufficient. Use a version number per replica per key. Each replica increments its
own and tracks numbers seen from others. The collection across all replicas = a
version vector (Riak’s dotted version vector, encoded as a “causal context” string).
Sent to clients on read, returned on write; lets the DB distinguish overwrites from
concurrent writes and makes it safe to read from one replica and write to another
(may create siblings, but no data lost if merged correctly).
Version vector ≠ vector clock (subtly different); version vectors are the right
structure for comparing replica states.
Key Takeaways
Question 1
Explain why “last write wins” (LWW) conflict resolution can lead to data loss even when all
writes succeed from the client’s perspective. Under what circumstances is LWW a safe
choice, and why?
Question 2
Question 3
The chapter states that a sloppy quorum “actually isn’t a quorum at all in the traditional
sense.” Explain what property of a strict quorum is violated by a sloppy quorum, and why
this means that even with w + r > n, you cannot guarantee reading the latest written value.
Question 4
Explain the difference between “concurrency” as used in this chapter and the common
intuition that two events are concurrent if they happen “at the same time.” Why is physical
time irrelevant to the definition of concurrent operations in a distributed system?
Question 5
The chapter describes how discarding the old leader’s unreplicated writes during failover
“may violate clients’ durability expectations.” Using the GitHub incident as a case study,
explain how this problem compounds when the database is not the only storage system, and
why autoincrementing counters are especially dangerous in failover scenarios.
Question 6
Why does the chapter argue that “pretending that replication is synchronous when in fact it
is asynchronous is a recipe for problems”? Relate this to the three consistency guarantees
discussed (read-after-write, monotonic reads, consistent prefix reads) and explain why
transactions are ultimately proposed as the solution.
Question 7
User A updates their profile picture on Node 1 (the leader), then immediately views their
profile, but the read request is routed to Node 2 (a follower that hasn’t replicated yet).
Question 8
A company runs a multi-leader replicated database across three datacenters (US, EU, Asia).
A user in the US edits a wiki page title from “Alpha” to “Beta.” Simultaneously, a user in the
EU edits the same page title from “Alpha” to “Gamma.”
a. Explain why a single-leader setup would prevent this conflict but at a cost.
b. Describe three different convergent conflict resolution strategies the system could
employ.
c. If the company chooses conflict avoidance, describe how they might route writes and
under what circumstances this strategy breaks down.
Question 9
An e-commerce system uses Dynamo-style leaderless replication with n=3. Node C goes
down for maintenance. During the outage, a customer places an order (written to Nodes A
and B with w=2). Node C comes back online.
a. If another customer immediately reads from Node C alone, what will they see?
b. Describe the two mechanisms the chapter discusses for bringing Node C up to date
after it recovers.
c. Which mechanism works poorly for data that is rarely read, and why?
Question 10
A social media application uses leader-based replication. User X posts a comment. User Y
(using a follower with 200ms lag) sees the comment and replies. User Z reads from a
different follower with 500ms lag and sees Y’s reply but not X’s original comment.
Question 11
A calendar synchronization app uses multi-leader replication where each device is a leader.
A user adds “Meeting with Bob at 3pm” on their laptop (offline) and “Meeting with Carol at
3pm” on their phone (also offline). Both devices come online and sync.
Question 12
A team is evaluating replication strategies for a new globally distributed database. They
need to support: (i) writes from any geography with low latency, (ii) tolerance for entire
datacenter failures, and (iii) eventual consistency is acceptable but data loss is not.
a. Which replication approach (single-leader, multi-leader, or leaderless) best fits these
requirements? Justify your choice.
b. What is the primary trade-off they accept with this choice?
c. Name one specific operational pitfall the chapter warns about for this approach.
Question 13
Question 14
A database cluster has n = 7 replicas. The operations team wants to tolerate up to 3 node
failures for both reads and writes while maintaining the quorum condition.
a. What are the minimum values of w and r that satisfy these requirements?
b. Prove that your values satisfy w + r > n.
c. A product manager asks: “Can we reduce write latency by lowering w to 2 while
keeping r = 6?” Does this still satisfy the quorum condition? What is the trade-off?
Question 15
Consider a system with n = 3, w = 2, r = 2. A write W1 succeeds on nodes {A, B}. Then node
B fails. A subsequent write W2 to the same key succeeds on nodes {A, C}.
a. If a client now reads from nodes {B, C} (after B recovers with stale data), which
version(s) might it see? Does the quorum guarantee hold?
b. The chapter lists several edge cases where quorum guarantees break down. Which
specific edge case does this scenario illustrate?
c. How does read repair help mitigate this issue?
Question 16
A Dynamo-style database uses sloppy quorums. The system has n = 3 designated “home”
nodes for key K. During a network partition, the client can only reach 2 out of 3 home nodes
plus 1 non-home node.
a. With w = 2, can the client still write to key K using a sloppy quorum? Explain.
b. With w = 3, can the client write? Explain.
c. After the partition heals, what mechanism returns data to the correct home nodes?
What guarantee does a sloppy quorum provide vs. a strict quorum?
Question 17
Single- Multi-
Dimension Leaderless
Leader Leader
Consistency guarantees
Question 18
Compare the four replication log implementation methods discussed in the chapter: (1)
Statement-based replication, (2) Write-ahead log (WAL) shipping, (3) Logical (row-based)
log replication, and (4) Trigger-based replication. For each, identify:
Question 19
Compare read repair and anti-entropy as mechanisms for bringing replicas up to date in
leaderless systems:
Question 20
Draw a timeline showing how a consistent prefix reads violation occurs. Your diagram
should include:
Time -->
Observer: [reads Partition Y @ t3: sees answer] [reads Partition X @ t4: no question
yet!]
Question 21
Trace through the version vector algorithm for the following scenario with a single-replica
database:
For each step, show: (a) The version number assigned by the server
(b) What values the server stores
(c) Which values are siblings
(d) After A’s merge write, what is the final state?
Answer Key
Answer 1
LWW assigns timestamps to concurrent writes and keeps only the write with the highest
timestamp, discarding others. Data loss occurs because when two clients write concurrently
to the same key, both receive a success acknowledgment (they wrote to w replicas), but
only one value survives. The clients have no way of knowing their write was discarded.
LWW is safe only when: (1) a key is written once and then treated as immutable (no
concurrent updates), such as using a UUID as the key so every write creates a unique key;
or (2) in caching scenarios where lost writes are acceptable.
Answer 2
If all followers are synchronous, any single node outage blocks all writes system-wide (the
leader must wait for every follower to confirm). The more nodes you have, the higher the
probability that one is down, making the system extremely unreliable.
Semi-synchronous replication keeps one follower synchronous and the rest asynchronous. If
the synchronous follower becomes unavailable, an asynchronous follower is promoted to
synchronous. This guarantees an up-to-date copy of data exists on at least two nodes
(leader + one synchronous follower) while avoiding the brittleness of fully synchronous
replication. Fully asynchronous replication cannot guarantee durability of confirmed writes if
the leader fails.
Answer 3
A strict quorum guarantees that the set of nodes receiving writes (w) and the set of nodes
read from (r) overlap in at least one node, ensuring the read sees the latest write. A sloppy
quorum allows writes to go to nodes that are not among the designated n “home” nodes for
a key. This means w writes may end up on different nodes than the r reads, breaking the
overlap guarantee. The sloppy quorum only provides durability (data is stored on w nodes
somewhere), not a consistency guarantee that reads will find the latest value until hinted
handoff completes.
Answer 4
The chapter defines two operations as concurrent if neither knows about the other (neither
“happens before” the other). Physical time is irrelevant because in distributed systems, clock
synchronization is unreliable, and network delays can prevent one operation from knowing
about another even if they occurred far apart in wall-clock time. Two operations can be
“concurrent” even if they happen minutes apart, if network problems prevented causal
information flow. Conversely, two operations happening at the exact same wall-clock time
might not be concurrent if one causally depends on the other through some communication
path.
Answer 5
When the old leader’s unreplicated writes are discarded during failover, the new leader may
reuse primary key values (from autoincrementing counters) that the old leader had already
assigned. If other systems (like Redis) use these primary keys as references, the reuse
creates inconsistency across storage systems. In the GitHub incident, a MySQL follower
promoted to leader had a lagging autoincrement counter, causing primary key reuse. Since
Redis had already associated data with those original keys, the reuse caused private data to
be disclosed to wrong users. The fundamental problem is that the database is not the only
source of truth – external systems have stale references to discarded writes.
Answer 6
When applications assume replication is synchronous (i.e., reads from any replica return the
latest data), but it is actually asynchronous, they will encounter: - Violation of read-after-
write: users don’t see their own writes - Violation of monotonic reads: users see data “go
backward in time” - Violation of consistent prefix reads: causal ordering is broken
These bugs are intermittent (only appearing when replication lag spikes) and hard to
reproduce. The chapter proposes transactions as the proper solution because they provide
stronger guarantees at the database level, relieving application developers from
implementing complex workarounds (like routing reads to leaders, tracking timestamps, etc.)
in application code.
Answer 7
b. Three approaches:
1. Read the user’s own profile from the leader (since only the owner can edit it), read
other profiles from followers.
2. Track the timestamp of the user’s last write; for some time window after the update
(e.g., one minute), route all reads to the leader.
3. The client remembers the timestamp of its most recent write; the system ensures any
replica serving reads reflects updates at least up to that timestamp (using logical
timestamps like log sequence numbers).
Answer 8
a. Single-leader prevents the conflict because all writes go through one leader, which
serializes them (the second write would see the first and either block or abort). The
cost is that one of the users (US or EU) must write over the internet to the leader’s
datacenter, adding significant latency.
1. Last write wins: Assign timestamps; highest timestamp wins (but causes data loss –
one edit is silently discarded).
2. Merge values: Concatenate or combine changes (e.g., title becomes “Beta/Gamma”).
3. Preserve conflict: Store both versions in an explicit conflict record and let application
code (or the users) resolve it later on read.
c. For conflict avoidance, route all writes for a particular record to a designated leader
(e.g., based on the wiki page’s “home” datacenter). This breaks down when: the
designated datacenter fails and traffic must be rerouted, or when the routing rule
changes (e.g., page ownership transfers), creating a window where concurrent writes
hit different leaders.
Answer 9
a. Node C has stale data. Reading from Node C alone would return the old value (the
order would not appear).
b. Two mechanisms:
1. Read repair: When a client reads from multiple nodes in parallel and detects that
Node C returns a stale value, it writes the newer value back to Node C.
2. Anti-entropy process: A background process constantly compares data between
replicas and copies missing data to Node C.
c. Read repair works poorly for rarely-read data. If the order record is never read, read
repair is never triggered, and Node C remains perpetually stale. Without an anti-
entropy process, there is no limit to how old the value on Node C might be, reducing
durability.
Answer 10
c. Solutions:
Write placement: Ensure causally related writes go to the same partition (e.g., a
comment thread all on one partition), so they replicate together in order.
Causal dependency tracking: Use algorithms that explicitly track causal
dependencies between writes, ensuring a replica does not serve a causally-later write
until all its dependencies are also available.
Answer 11
a. Each device is essentially a “datacenter” with its own leader that accepts writes
independently. The network between them (internet/sync) is unreliable and may be
unavailable for extended periods. This is structurally identical to multi-datacenter multi-
leader replication.
b. It is a conflict because both meetings occupy the same time slot (3pm), violating the
application invariant that a user cannot have two meetings at the same time. This is a
subtle conflict – different records, but an application-level constraint is violated.
c. Appropriate strategies: (i) Preserve both bookings and alert the user to resolve the
double-booking on next sync (on-read resolution). (ii) Use CRDTs or application-level
merge logic that detects time-slot overlaps and flags them. (iii) Prompt the user to
choose which meeting to keep (application-level conflict resolution on read).
Answer 12
a. Multi-leader replication best fits. Justification: (i) Each datacenter has a local leader,
so writes are low-latency in any geography. (ii) Each datacenter continues operating
independently during failures. (iii) Asynchronous inter-datacenter replication means no
data loss within a datacenter (writes are durable locally before replicating).
b. The primary trade-off is conflict handling complexity: the same data may be
concurrently modified in different datacenters, requiring conflict detection and
resolution logic.
Answer 13
c. w + r = 2 + 2 = 4. Since 4 is NOT > 5, the quorum condition is not satisfied. You lose
the guarantee that reads will overlap with writes, meaning you may read stale values
because no node in your read set necessarily received the latest write.
d. With w = 5, r = 1: Reads are very fast (only need one response) but writes require ALL
nodes to acknowledge. Write tolerance: 0 node failures (any single failure blocks
writes). Read tolerance: 4 node failures. Trade-off: optimized for read-heavy workloads
at the cost of write availability.
Answer 14
a. To tolerate 3 node failures:
Answer 15
a. After B recovers with stale data: Client reads from {B, C}. B has W1 (“old” value from
before failure), C has W2 (latest). The client gets W2 from C, so it does see the latest
value. However, if B had recovered with data older than W1, or if the scenario involved
concurrent writes, the guarantee might not hold.
Actually, more precisely: B was down when W2 occurred, so B only has W1. C was down
when W1 occurred, so C only has W2. The client reads both W1 and W2. Which is newer?
The system must use version numbers. If it can determine W2 is newer (it was written after
W1 since it was on the same key), it returns W2. The quorum guarantee holds here because
the overlap exists (node A has both, but A isn’t in the read set). Wait – the read is from {B,
C}, which does NOT overlap with the write set of W2 ({A, C}) at node C. So the client does
see W2. The guarantee holds in this case.
b. This illustrates the edge case: “If a node carrying a new value fails, and its data is
restored from a replica carrying an old value, the number of replicas storing the new
value may fall below w, breaking the quorum condition.” If node A (which has both W1
and W2) fails and is restored from B’s data, then W2 only exists on C (1 node), which
is below w=2.
c. Read repair helps because when a client reads from multiple nodes and detects that
one node has a stale value, it writes the newer value back. This gradually brings all
replicas up to date as reads occur.
Answer 16
a. Yes. With w = 2, the client needs 2 successful write responses. It can write to the 2
reachable home nodes. Even without the sloppy quorum mechanism, this works
because 2 home nodes are reachable. (If only 1 home node were reachable, the client
could use the non-home node to reach w = 2 via a sloppy quorum.)
b. With w = 3, the client needs 3 successful responses. It can reach 2 home nodes + 1
non-home node = 3 total. With a sloppy quorum, yes – the write can succeed by
counting the non-home node. Without sloppy quorum, it would fail because only 2 of
the 3 designated home nodes are reachable.
c. Hinted handoff returns data to correct home nodes after the partition heals. The non-
home node that temporarily accepted the write sends it to the appropriate home node.
Guarantee difference: A strict quorum guarantees that subsequent reads (from r of the n
home nodes) will see the latest write. A sloppy quorum only guarantees durability (data is
stored on w nodes somewhere) but NOT that reads from home nodes will see it until hinted
handoff completes.
Answer 17
Weakest:
Strongest: sequential Medium: eventual eventual
Consistency
write ordering via convergence, consistency, edge
guarantees
single leader conflicts possible cases even with
quorums
Good: writes to
Worst: all writes Best: writes
Write latency local replicas, but
traverse internet to processed at local
(geo-distributed) must wait for w
leader’s datacenter datacenter leader
responses
High: concurrent
Conflict None: no conflicts High: must detect
writes require
handling (single serialization and resolve conflicts
version vectors
complexity point) between leaders
and merging
Hard: quorum
Hard: surprising
Ease of Easiest: one leader, edge cases,
interactions, topology
reasoning clear ordering concurrent write
issues
semantics
Answer 18
Nondeterminism
Systems requiring
Statement- Compact (just (NOW(), RAND(),
deterministic transactions
based SQL statements) triggers) breaks
(VoltDB)
replication
Tightly coupled to
Single-version
Simple (reuse storage engine; blocks
WAL homogeneous
existing log zero-downtime
shipping deployments
infrastructure) upgrades across
(PostgreSQL, Oracle)
versions
Decoupled from
storage engine; Change data capture,
Logical
backward More verbose than heterogeneous
(row-
compatible; statement-based replication, rolling
based) log
parseable by upgrades
external systems
Answer 19
a. Read repair is triggered when a client reads from multiple nodes and detects stale
values. Anti-entropy is a continuously running background process that proactively
scans for differences.
b. Neither preserves write ordering. Anti-entropy explicitly “does not copy writes in any
particular order.” Read repair is opportunistic and order-independent.
c. In a read-repair-only system, values that are rarely read may remain stale indefinitely
on some replicas. There is no limit to how old a value might be. This reduces
durability because if the nodes holding the fresh copy fail, the value is lost.
Answer 20
Time ------>
t1 t2 t3 t4 t5
| | | | |
Partition X: [Mr. Poons writes ] [replicated to
"How far can you observer's
see, Mrs. Cake?"] replica]
Explanation: Partition Y replicates quickly (low lag), so the answer is visible at t3. Partition
X replicates slowly (high lag), so the question is not yet visible at t4. Because partitions
operate independently with no global ordering, an observer reading across partitions can
see causally-later writes before causally-earlier ones.
Answer 21
Step 1: Client A writes “x = 1” with no prior version number. - Server assigns version 1 -
Server stores: {v1: “1”} - No siblings (only one value)
Step 2: Client B writes “x = 2” with no prior version number. - Server assigns version 2 -
Server stores: {v1: “1”, v2: “2”} - Siblings: “1” (v1) and “2” (v2) – both concurrent because
B’s write included no version number, meaning it’s not based on any prior state, so it cannot
overwrite v1.
Step 3: Client A reads key x. - Server returns: values [“1”, “2”], latest version number = 2 -
Client A merges: decides on “[1, 2]”
Step 4: Client A writes “x = [1, 2]” with version number 2. - Server assigns version 3 -
Server logic: version 2 is included, so overwrite all values at version <= 2 - Server stores:
{v3: “[1, 2]”} - No siblings – the merge resolved the concurrency - Final state: version 3,
value = “[1, 2]”
Key insight: By including version 2 in its write, Client A signals that it has seen and merged
all values up to version 2. The server safely discards v1 and v2, knowing they are
incorporated into the new value.