Gossip Protocol and Failure Detection

Large clusters need two things without a central coordinator: agreement on which nodes are members, and a way to tell which of them have failed. A gossip protocol spreads information by having each node periodically exchange state with a few randomly chosen peers. A failure detector decides, based on observed messages, whether a node is alive. Together they provide decentralized membership in systems such as Cassandra, Consul (through Serf), Riak, and Amazon's Dynamo design.

Gossip dissemination

A gossip protocol, also called an epidemic protocol, works in rounds:

1. Each node, every fixed interval (for example one second), picks one or a few peers at random.
2. It sends them some of its state, or a summary of it.
3. The peer merges the received state with its own. It may reply with its own state.

graph LR
    A["A: has the update at round 0"] -->|"round 1"| B["B"]
    A -->|"round 2"| C["C"]
    B -->|"round 2"| D["D"]

The set of nodes that know an update roughly doubles each round early on. Information reaches all N nodes in O(log N) rounds with high probability. Each node sends a constant number of messages per round, so per-node load does not grow with cluster size. For a cluster of 1,000 nodes, doubling alone needs about 10 rounds (log2(1000) is close to 10), and a few more rounds cover the remaining uninformed nodes, so at a one-second interval an update reaches the cluster in roughly a dozen seconds.

Exchange styles

Properties

• Scalable. Load per node is constant and independent of cluster size.
• Fault tolerant. There is no single point of failure, and messages lost in one round are covered by later rounds.
• Eventually consistent. Nodes converge, but at any moment different nodes may hold different views.
• Redundant. Nodes may receive the same update several times. Digests (a version number or hash per item) reduce redundant transfers.

Gossip is used for membership lists, cluster metadata such as token ownership, and aggregate values such as load. It does not fit data that must be delivered in a strict order or exactly once.

Membership

Each node keeps a membership table: a list of known nodes, each with an address, a status (alive, suspected, down), and a heartbeat counter or incarnation number. Gossip merges tables by keeping the entry with the highest counter for each node.

• A node that joins contacts a seed node to obtain the current table, then announces itself through gossip.
• A node that leaves announces its departure. A node that fails is detected by the failure detector and its status is spread by gossip.
• Cassandra's gossip exchanges each node's endpoint state (heartbeat generation and version, status, load, and token information) once a second with up to three other nodes.

Failure detection

A failure detector cannot know whether a remote node is dead, slow, or unreachable, because a message that has not arrived yet may arrive later. Any detector therefore makes trade-offs between:

• Completeness. Every failed node is eventually detected by all healthy nodes.
• Accuracy. Healthy nodes are not wrongly declared failed (false positives).
• Speed. Detection happens quickly.

A wrong declaration triggers costly actions such as re-replicating data or electing a new leader (see leader election).

Heartbeats and fixed timeouts

Each node sends a heartbeat every interval T. A node is suspected when no heartbeat has arrived for a timeout D. Weaknesses:

• Choosing D is difficult. Too low gives false positives when the network or the node is briefly slow. Too high delays detection.
• The right value differs between networks and changes over time.
• All-to-all heartbeating costs O(N^2) messages.

Phi-accrual failure detector

The phi-accrual failure detector (Hayashibara et al.) replaces a binary alive/dead output with a continuous suspicion level, phi. The detector keeps a sliding window of recent heartbeat inter-arrival times and estimates their distribution. When a heartbeat is overdue by t since the last arrival, it computes the probability P(t) that a heartbeat would still arrive after that delay. Then:

phi = -log10(P(t))

Each unit of phi is a factor of ten in confidence. phi = 1 means about a 10% chance that the node is alive and merely late, phi = 2 about 1%, and phi = 3 about 0.1%. The application chooses a threshold: suspect the node when phi exceeds it. Cassandra uses this detector, with a configurable threshold (phi_convict_threshold), and Akka Cluster does too.

Because the estimate is derived from observed arrival times, the detector adapts automatically. On a network with jittery heartbeats it tolerates larger delays before suspecting a node, and on a stable network it detects failure sooner.

A simplified version assuming exponentially distributed inter-arrival times, where P(t) = exp(-t / mean):

import java.util.Locale;

public class Main {
    static double phi(double elapsedMs, double meanIntervalMs) {
        // P(t) = exp(-t / mean), so -log10(P) = t / (mean * ln 10)
        return elapsedMs / (meanIntervalMs * Math.log(10));
    }

    public static void main(String[] args) {
        double mean = 1000.0;
        for (double elapsed : new double[] {500.0, 1000.0, 3000.0, 8000.0, 12000.0}) {
            double p = phi(elapsed, mean);
            String verdict = p > 3.0 ? "suspect" : "alive";
            System.out.println(String.format(Locale.ROOT, "elapsed %7.0f ms  phi %.2f  %s", elapsed, p, verdict));
        }
    }
}

elapsed     500 ms  phi 0.22  alive
elapsed    1000 ms  phi 0.43  alive
elapsed    3000 ms  phi 1.30  alive
elapsed    8000 ms  phi 3.47  suspect
elapsed   12000 ms  phi 5.21  suspect

Implementations differ in the distribution model. Cassandra's detector uses an exponential model like the one above, based on the mean of the window. The original paper and Akka Cluster's detector use a normal-distribution approximation, which also accounts for the variance of inter-arrival times.

SWIM

SWIM (Scalable Weakly-consistent Infection-style process group Membership) separates failure detection from dissemination and keeps per-node message load constant. It underlies the membership layer of Serf, Consul, and HashiCorp's memberlist.

Failure detection. Every protocol period, node A:

1. Picks a random member B and sends it a ping.
2. If no ack arrives in time, picks k other random members and asks each to send an indirect ping to B (ping-req).
3. If neither the direct nor any indirect probe yields an ack within the period, marks B as suspect.

sequenceDiagram
    participant A
    participant B
    participant C
    participant D
    A->>B: ping
    Note over A: no ack within timeout
    A->>C: ping-req(B)
    A->>D: ping-req(B)
    C->>B: ping
    D->>B: ping
    B-->>C: ack
    C-->>A: ack via C, B is alive

Indirect probes avoid declaring a node failed because of a single faulty network path between A and B.

Suspicion. A suspected node is not removed immediately. Its suspect status is gossiped, and the node has a timeout to refute it by broadcasting an alive message with a higher incarnation number. Only after the timeout expires is it declared confirmed dead and removed. This reduces false positives caused by slow nodes.

Dissemination. Membership updates (join, leave, suspect, alive, dead) are piggybacked on the ping and ack messages already being sent, so no separate gossip traffic is needed. Each update is included in a bounded number of messages, on the order of log N, before it is dropped.

Properties: each node sends and receives a constant number of messages per protocol period, expected detection time is constant and independent of cluster size, and a failed node is probed by some member within a bounded number of periods with high probability.

Comparison

Limits

• Gossip provides eventual, not immediate, agreement on membership. Different nodes can briefly disagree on who is alive.
• Failure detection is unreliable in principle. A long garbage collection pause or network congestion looks like a crash.
• Membership decided by gossip must not be used as the sole basis for actions needing exclusivity, such as electing a single writer. Those need consensus (Raft) or fencing tokens.
• Gossip of large state per round consumes bandwidth. Systems send digests and transfer only differences.

Interview questions

• Why does gossip converge in O(log N) rounds? The number of informed nodes roughly doubles each round while few are informed, then the remaining uninformed nodes are reached quickly, so total rounds grow logarithmically with cluster size.
• What problem does the phi-accrual detector solve compared to fixed timeouts? It adapts the suspicion threshold to the observed heartbeat distribution and outputs a graded suspicion value, so one threshold setting works across networks with different jitter.
• Why does SWIM use indirect probes? The failed ping may be caused by a bad path between two nodes only. Asking other members to probe the target reduces false failure declarations.
• What is the purpose of the suspect state in SWIM? It gives the target time to prove it is alive with a higher incarnation number before being removed, reducing false positives.
• Can gossip-based membership be used for leader election? Not safely by itself, because views differ between nodes at any moment and can produce two leaders. A consensus protocol is required.