Replicas drift apart. A write reaches two of three nodes while the third reboots; a schema change is announced while a rack is unreachable; a node joins and nobody has told it who else exists. Eventually consistent systems promise convergence, and this chapter is about the machinery that delivers it: anti-entropy protocols that find and fix divergence between copies, and dissemination protocols that spread information (data, metadata, membership) through a cluster without a central broadcaster that could fail or bottleneck.

Anti-entropy answers "are my copies the same, and if not, which bits differ?" with as little traffic as possible. Dissemination answers "how does every node hear about this?" in few rounds and with little redundancy, even when nodes fail mid-way. Dynamo-style databases run both: gossip for membership and metadata, read repair plus Merkle-tree repair for data.

In one minute

• Three ways to reach every node: direct broadcast from one process (simple, fragile), periodic peer-to-peer exchange (anti-entropy, slow but inevitable), and cooperative broadcast where receivers forward (gossip, fast and robust).
• Foreground anti-entropy rides on normal traffic: read repair fixes replicas that returned stale values during a read; hinted handoff stores a write for a replica that was down and replays it later. Both fix only what is touched.
• Digest reads fetch the full value from one replica and only hashes from the rest, paying for full reads only when hashes disagree.
• Background anti-entropy uses Merkle trees: hash a range into a tree, exchange trees, descend only where hashes differ. Comparison is logarithmic; building is a full scan.
• Bitmap version vectors (Riak) track per peer which operations have been seen, so reconciliation ships exactly the missing updates.
• Gossip follows epidemic models: with fanout f, information reaches N nodes in about log N rounds; redundancy is the price. Rumor mongering stops forwarding after a loss-of-interest rule.
• Plumtree pushes full messages along a spanning tree and only ids elsewhere, repairing the tree lazily. HyParView keeps a small active view and a larger passive view so nodes talk to few peers yet survive churn.
• Membership is gossiped too: SWIM piggybacks join, leave and suspect events on ping traffic; Cassandra gossips heartbeat state into a phi-accrual failure detector.

Three ways to spread state

Notification broadcast: the process holding new information sends it to every other process. The sender must know every member, retry failures and stay up until all have acknowledged; load grows with the cluster, and a mid-way crash leaves some nodes informed and some not.

Periodic peer-to-peer exchange: every node, on a timer, picks a peer and the two compare and reconcile state. No node is special and no message needs to arrive on the first try, because the next round retries. This is anti-entropy in the original Xerox PARC sense (Demers et al., 1987): thorough and slow.

Cooperative broadcast: a receiver forwards to a few others, who forward in turn. Load is shared, the originator can die after the first hop, and the message still reaches everyone with high probability. This is gossip (epidemic dissemination, rumor mongering).

Entropy measures disorder, here how far replicas have diverged. Anti-entropy reduces it in the foreground, piggybacking on reads and writes (read repair, hinted handoff), or in the background, by periodic reconciliation (Merkle trees, version-vector exchange). Foreground mechanisms are cheap and fix hot data; background mechanisms are expensive and fix everything.

Read repair and hinted handoff

Read repair mechanism

In a leaderless store the coordinator (whichever node received the client's request) fans a read out to replicas. If replicas return different versions, divergence has been detected for free. Read repair uses that moment: the coordinator picks the newest version (by timestamp or version-vector dominance) and writes it back to the replicas that were stale.

Read repair touches only the replicas that participated and only the keys read. A key nobody reads is never repaired this way, and a read at consistency level ONE compares nothing.

Variant	What the client waits for	Trade
Blocking read repair	Repair writes acknowledged by the stale replicas before the answer	Preserves monotonic reads across the replica set; higher tail latency
Asynchronous read repair	Nothing; repair happens in the background	Lower latency; a read right after may still hit the stale replica

Cassandra performs blocking read repair when a QUORUM read finds mismatches among the replicas it contacted, because answering first would let a later QUORUM read go backwards. Probabilistic background read repair (read_repair_chance) was removed in Cassandra 4.0 in favour of scheduled repair.

Digest reads

Returning every replica's full value wastes bandwidth when replicas almost always agree. A digest read sends a full data request to one replica and digest requests to the others, which return only a hash of the value. If digests match the full value's hash, the coordinator returns the value; otherwise it issues full reads to the disagreeing replicas and repairs with the newest version. The extra round trip is paid only when there is divergence, which you were going to pay for anyway.

Hinted handoff

Hinted handoff is the write-side counterpart. When a write cannot reach one of its target replicas, the coordinator (or, in Dynamo's design, a stand-in node on the ring) stores a hint: the write plus the identity of the replica that missed it. When that replica is seen alive, hints are replayed to it.

Details that matter:

• In Cassandra hints do not count toward the write's consistency level: a QUORUM write with N = 3 still needs two real acknowledgements. In Dynamo's sloppy quorum the stand-in's write does count toward W, which is why Dynamo quorums do not guarantee read/write overlap.
• Hints have a window. Cassandra keeps them for max_hint_window (three hours by default); a node down longer must be repaired another way. Hints also consume disk and replay bandwidth on the coordinator.
• Replay is an ordinary write carrying the original timestamp, so a newer value already on the target wins.

Hinted handoff shortens the divergence window for transient failures. It does nothing for data lost after it was written (disk corruption, restore from backup) or for hints on a coordinator that dies before replaying them. That is background repair's job.

Merkle trees and version vectors

Why hashing a tree

To reconcile two replicas of a large key range you want the differing keys without shipping all the keys. One hash over the whole range says whether anything differs but not what; a hash per key says what but costs as much as the data. A Merkle tree is the middle: leaves are hashes of key sub-ranges, inner nodes hash their children, the root hashes everything.

Building and comparing

A replica scans its data in key order, hashes rows into leaf buckets, then hashes pairs upward. Two replicas exchange roots; if equal, done. If not, they exchange children, descend into subtrees whose hashes differ, and stop at disagreeing leaves, which identify ranges to stream (in both directions, since each side may have writes the other lacks). Messages are proportional to tree depth times the number of differing leaves: logarithmic in data size per difference.

Costs and how systems manage them

The weakness is construction. Every write changes a leaf and all hashes up to the root, so keeping trees current under heavy writes is expensive. Most systems build trees on demand per repair session, which means a full scan of the range: in Cassandra a validation compaction that reads every SSTable for the range, I/O heavy and hence scheduled for quiet hours. Resolution is a knob: more leaves pinpoint differences and stream less, at more memory and messages; Cassandra uses a bounded depth per range and builds one tree per vnode range. Riak's active anti-entropy keeps persistent hash trees on disk and updates them incrementally, trading write-path cost for no rescans. Incremental repair (Cassandra 2.1, reworked in 4.0) marks SSTables as repaired so later sessions hash only unrepaired data; full repair rebuilds everything and remains necessary after data loss.

Merkle trees appear in Git, Bitcoin, ZFS and certificate transparency logs for the same reason: cheap verification that two large structures agree and cheap localisation of where they do not.

Bitmap version vectors

Merkle trees find differences by hashing content. Riak (2.x onward) instead tracks causality so a node knows exactly which updates a peer lacks, following Gonçalves, Almeida, Baquero and Fonte (2015). Each node keeps, per peer, a bitmap version vector entry: a base counter meaning "I have every operation from this peer up to here" plus a bitmap of operations beyond the base that arrived out of order. As gaps fill, the base advances and the bitmap shrinks, so in a healthy cluster the structure is roughly one integer per peer regardless of key count.

Anti-entropy becomes log comparison: A sends its vector to B, B sees which counters A lacks and ships exactly those operations from its per-peer log. No hashing, no full scan, cost proportional to peers rather than keys. The same structure tells Riak when a tombstone has been seen everywhere and can be collected. The cost is retaining per-peer logs until peers catch up, and the fact that operation tracking cannot detect silent corruption of a value the log says was delivered. Merkle trees and version vectors are complementary.

Gossip dissemination

Epidemic models

A node holding the information and still spreading it is infective; one that has not heard is susceptible; one that heard and stopped spreading is removed.

• SI (susceptible-infective): infective nodes never stop. Each round every infective node contacts f random peers (f is the fanout). Susceptible nodes decay roughly exponentially and the cluster is fully infected in O(log N) rounds, but infective nodes keep sending forever, so steady-state traffic grows with the number of rumours ever started. This is pure anti-entropy exchange.
• SIR (susceptible-infective-removed): infective nodes eventually lose interest. Traffic is bounded, with a small probability that some node never hears because the rumour died out first. The stopping rule is tuned so that probability is acceptable.

Push, pull, rounds and redundancy

Push gossip (infective nodes send) is efficient early, when most peers are susceptible, and wasteful late. Pull (nodes ask "what's new?") is wasteful early and efficient late, because the last susceptible node will pull from an infected peer with high probability. Push-pull combines both in one exchange; Cassandra's syn/ack/ack2 gossip is push-pull in effect.

With fanout f, roughly log base f of N rounds reach all nodes: fanout 3 in a 1,000-node cluster converges in about seven rounds, a few seconds at one round per second, with per-node load of f messages per round regardless of cluster size. Two metrics describe the cost. Message redundancy counts deliveries of a rumour a node already had; it is inherent to random selection and grows with fanout. Latency to convergence exceeds the time the last node receives the message, because under SIR nodes keep spreading until the stopping rule fires.

Rumor mongering and loss of interest

Under SIR the stopping rule is the loss-of-interest function. Probabilistic: after each round, or after each redundant contact, an infective node stops with probability 1/k. Counter: a node stops after receiving the rumour from peers who already had it some fixed number of times, on the reasoning that if everyone you talk to already knows, the rumour has saturated. Demers et al. analysed both (and "feedback" versus "blind" variants) and measured the residue, the fraction of nodes never reached. Practical systems back rumor mongering with a slower full anti-entropy pass so residue is eventually zero.

Overlay networks and hybrid gossip

Spanning trees

Random peer selection wastes bandwidth on duplicates. If nodes form a spanning tree and forward only to tree children, every node receives each message once and latency is the tree depth. The problem is fragility: a failed interior node cuts off its subtree until the tree is rebuilt, and rebuilding needs coordination. Pure trees are too brittle for the churn gossip was designed for.

Plumtree: eager push plus lazy push

Plumtree (Leitão, Pereira and Rodrigues, 2007), the epidemic broadcast tree, combines the two. Each node splits its peers into eager peers, which receive full messages, and lazy peers, which receive only message ids (IHAVE). The eager links form a spanning tree that emerges from the first broadcasts: a node receiving a full message a second time over a different link sends PRUNE to that sender, moving the link to lazy. Steady-state cost is one full copy per node plus small ids over redundant links.

Repair is lazy and local. A node that gets an IHAVE for a message it has not received in full waits a timeout; if the message still has not arrived, the tree is broken upstream, so it sends GRAFT to the announcing lazy peer, which replies with the message and promotes the link to eager. The tree heals from announcements without a global rebuild, and the ids double as a reliability check. Plumtree is used in Erlang and Elixir cluster libraries and in some blockchain gossip layers.

Partial views and HyParView

Gossip to "f random nodes" assumes every node knows the full membership, which itself must be disseminated and does not scale to very large groups. Partial view protocols give each node a small random sample of members and refresh it so the union of views forms a connected graph with good expansion.

HyParView keeps two views per node. The active view is small (fanout plus one), symmetric, and maintained over TCP connections that double as a failure detector: when a connection drops the peer is replaced. The passive view is larger (tens of nodes), maintained by periodic shuffle exchanges of samples, and used as a pool to refill the active view after failures. Joins are forwarded across active views by a random walk so newcomers land in several active views. The paper shows recovery after half the nodes fail at once, while each broadcast uses very few links. HyParView plus Plumtree is a common pairing: resilient neighbours, with a spanning tree carved out of them.

Membership and failure detection by gossip

Who is in the cluster, who is suspected dead, which schema version each node runs and which token ranges it owns are all gossiped in Dynamo-style systems, because a central membership service would be the single point of failure the architecture avoids.

SWIM

SWIM (Das, Gupta and Motivala, 2002) separates failure detection from dissemination and then rides the latter on the former. Each period a node pings one random member. On no ack, it asks k other members to ping-req the target indirectly; if none succeeds, the target is declared failed. Indirect probing avoids false positives from one congested link.

The suspicion mechanism softens this: the target is marked suspect and that is disseminated; a live target hears it and refutes with a higher incarnation number, which overrides the suspicion. Only after a suspicion timeout without refutation does the node become confirmed dead. Membership updates (alive, suspect, confirm, join, leave) are piggybacked on pings and acks, so dissemination costs no extra messages and spreads epidemically. HashiCorp's memberlist (behind Consul and Serf) implements SWIM with the Lifeguard extensions, which make an overloaded observer less eager to suspect others.

Cassandra's gossiper

Cassandra gossips once per second: each node picks one to three live peers (plus, with some probability, a seed and a node it believes is down, so partitions heal) and exchanges endpoint state: per node, a heartbeat state (a generation, the node's start time, which changes on restart, plus a version counter incremented every second) and a set of application states (status, tokens, schema version, load, data centre and rack). Each application state carries the version at which it changed, so exchanges send only entries newer than the peer reported, and generation ordering makes a restarted node's state win over stale state about its previous incarnation.

Heartbeat arrival times feed the phi-accrual failure detector (Hayashibara et al., 2004). Instead of a fixed timeout it keeps a window of inter-arrival intervals and computes phi, proportional to the negative log of the probability that a heartbeat this late would still arrive. A node is convicted when phi exceeds phi_convict_threshold (default 8). The detector adapts to each peer's network and avoids both the flapping of a short timeout and the sluggishness of a long one.

Gossip membership is weakly consistent: at any instant nodes may disagree about the cluster. Dynamo-style systems tolerate that (a coordinator that thinks a replica is up will time out and store a hint); where they need a consistent view (token ownership during bootstrap, schema agreement) they add waits or, as in Cassandra 5's transactional cluster metadata, a consensus-replicated log.

Design trade-offs

Mechanism	Fixes or spreads	Cost	Convergence	Falls short when
Read repair	Keys read, replicas contacted	Extra writes on mismatch	Immediate for hot keys	Cold keys; ONE-level reads
Digest reads	Same as read repair	Hashing; second round trip on mismatch	Same	Replicas outside the read set
Hinted handoff	Writes missed by a down replica	Hint storage and replay	When the replica returns	Hint window expires; coordinator dies
Merkle-tree repair	Whole ranges, including cold data	Full scan to build; logarithmic compare	Hours, scheduled	Heavy write load during build
Bitmap version vectors	Exactly the missing operations	Per-peer logs	Fast	Corruption of delivered data
Random gossip	Data or metadata, probabilistically	f messages per node per round	O(log N) rounds	Redundancy; needs full membership
Plumtree	Broadcast, near one copy per node	Small ids on lazy links	Tree depth plus repair timeout	Poor first-broadcast tree
HyParView	Membership sample	Few TCP connections; shuffles	Fast refill	Not a global member list
SWIM, phi-accrual	Liveness and membership	One ping per period	O(log N) periods	Weakly consistent view

For a Dynamo-style store: hinted handoff and read repair handle minute-scale blips; Merkle or version-vector repair runs on a schedule and after any outage longer than the hint window; gossip carries membership and schema; nothing here provides linearizability, so anything that needs it goes through consensus.

Interview angle

• A Cassandra node was down for six hours. How does it catch up? Hints cover only the first three hours by default, and read repair fixes only hot keys read at QUORUM. The operator runs nodetool repair (or scheduled incremental repair), which builds Merkle trees per range, compares with the other replicas and streams the differing ranges.
• Why does Cassandra block on read repair at QUORUM? If it answered before the stale replicas were fixed, a later QUORUM read could land on a set that lacks the repaired value and go backwards, violating monotonic reads.
• Why not maintain Merkle trees incrementally? Every write would update a leaf and every ancestor, serialising writes on the tree and adding I/O per write. Most systems build per session and accept a full scan; Riak's AAE keeps persistent trees and pays on the write path.
• Gossip versus a central coordinator for membership? A coordinator gives a consistent view but must itself be replicated with consensus and is a bottleneck. Gossip has no special node, costs f messages per node regardless of N, and converges in O(log N) rounds, but the view is weakly consistent and detection is probabilistic.
• How does SWIM avoid declaring a healthy node dead because of one bad link? Indirect pings through k other nodes, plus a suspect state the target can refute with a higher incarnation number before the suspicion timeout.
• Push versus pull gossip? Push spreads fast while most nodes are uninformed and wastes messages at the end; pull is wasteful early and efficient late. Push-pull takes both for slightly larger messages.
• Propagate a feature flag to 10,000 servers in seconds without a hot spot. Gossip with fanout 3 or 4 and a version number on the flag state; expect roughly eight rounds. Add a periodic anti-entropy pull for stragglers and expose the version in health checks. If every node must switch at the same instant, this is the wrong tool: that needs a barrier via consensus.
• What do bitmap version vectors buy over Merkle trees? Exact knowledge of which operations each peer lacks, so reconciliation ships only those and costs scale with peers, not data. Merkle trees need a scan but detect any content difference, including corruption.

Pitfalls and misconceptions

• Hinted handoff makes writes durable on the down replica. It makes them likely to arrive later. Durability comes from the W replicas that acknowledged.
• Read repair keeps a cluster consistent. Only keys that are read, only on replicas contacted, only when more than one replica is contacted.
• Merkle trees are cheap because comparison is logarithmic. Construction is a full scan with heavy I/O; repair is among the most expensive routine operations in a Dynamo-style cluster.
• Gossip guarantees delivery. SIR rumor mongering leaves a non-zero residue; practical systems add a slow anti-entropy pass, and "eventually" still has no bound.
• Gossip gives every node the same membership view. Views are weakly consistent; protocols on top must tolerate disagreement or add a consistent metadata layer.
• Phi-accrual is just an adaptive timeout. Its output is a confidence value, so different subsystems can act at different thresholds.
• Plumtree is a configured spanning tree. The tree emerges from the first broadcasts and is continuously repaired by IHAVE and GRAFT; it is gossip that converges to a tree.

Key terms

Term	Meaning
Anti-entropy	Any process that detects and repairs divergence between replicas
Read repair	Fixing stale replicas with the newest value found during a read; blocking or asynchronous
Digest read	A read asking most replicas for a hash of the value instead of the value
Hinted handoff	Storing a write for an unreachable replica and replaying it when the replica returns
Sloppy quorum	Dynamo's writing to the first N healthy nodes, counting stand-ins toward W
Merkle tree	Tree of hashes over key ranges; lets replicas find differing ranges with logarithmic comparison
Validation compaction	Cassandra's full scan of a range to build a Merkle tree
Bitmap version vector	Per-peer base counter plus bitmap of out-of-order operations seen; basis of Riak's log-based anti-entropy
Gossip	Cooperative broadcast where each receiver forwards to a few random peers
Fanout	Number of peers a node forwards to per round
SI and SIR	Epidemic models without and with a removed state; SIR bounds traffic at the cost of possible non-delivery
Rumor mongering	SIR-style gossip where nodes stop forwarding after a loss-of-interest rule
Plumtree	Epidemic broadcast tree: eager push on tree links, lazy ids elsewhere, repair by graft and prune
HyParView	Membership protocol with a small symmetric active view over TCP and a larger passive view refreshed by shuffles
SWIM	Membership protocol with direct and indirect pings, suspicion with incarnation numbers, piggybacked dissemination
Phi-accrual failure detector	Outputs a suspicion level from the distribution of heartbeat inter-arrival times
Generation and version	Cassandra heartbeat state: restart counter and per-second counter ordering gossip updates

Further reading

• Database Internals, Chapter 12; DDIA Chapter 5 for leaderless replication from the application side.
• Alan Demers et al., "Epidemic Algorithms for Replicated Database Maintenance", PODC 1987.
• Giuseppe DeCandia et al., "Dynamo: Amazon's Highly Available Key-value Store", SOSP 2007, https://www.allthingsdistributed.com/files/amazon-dynamo-sosp2007.pdf
• João Leitão, José Pereira and Luís Rodrigues, "Epidemic Broadcast Trees", SRDS 2007 (Plumtree), and "HyParView: a Membership Protocol for Reliable Gossip-based Broadcast", DSN 2007.
• Abhinandan Das, Indranil Gupta and Ashish Motivala, "SWIM: Scalable Weakly-consistent Infection-style Process Group Membership Protocol", DSN 2002.
• Armon Dadgar, James Phillips and Jon Currey, "Lifeguard: Local Health Awareness for More Accurate Failure Detection", https://arxiv.org/abs/1707.00788
• Ricardo Gonçalves, Paulo Sérgio Almeida, Carlos Baquero and Victor Fonte, "Concise Server-Wide Causality Management for Eventually Consistent Data Stores", DAIS 2015.
• Naohiro Hayashibara et al., "The Phi Accrual Failure Detector", SRDS 2004.
• Apache Cassandra documentation on repair, https://cassandra.apache.org/doc/latest/cassandra/operating/repair.html
• Bartosz Sypytkowski's walkthrough of HyParView, https://www.bartoszsypytkowski.com/hyparview/