Distributed consensus and reliability

A distributed system is one in which a machine you have never heard of can fail and stop your work. The network loses packets, clocks drift, processes pause, and every message may arrive twice or never. This article is about the handful of ideas that let systems agree, retry, back off and fail over without making things worse.

The map

Read this first when short on time. Every branch is a section below.

CAP: what a partition forces you to choose

The CAP theorem says that when the network splits a system into groups that cannot talk to each other, each group must either refuse to answer (stay consistent) or answer with what it has (stay available). It is not a menu of three from which you pick two; partitions are not optional, so the only choice is what to do during one.

flowchart TB
  subgraph Normal["Normal operation"]
    direction LR
    A1[node A] <--> B1[node B]
  end
  subgraph Split["Network partition"]
    direction LR
    A2[node A] x--x B2[node B]
  end
  Split --> CP["CP choice: the minority side refuses writes, maybe reads, until the partition heals"]
  Split --> AP["AP choice: both sides keep answering, data diverges, reconcile later"]
  Normal ~~~ Split

Consistency (C)

Linearizability: every read returns the most recent write, as if there were one copy. Not the C in ACID.

Availability (A)

Every request to a working node gets a non-error response, even if it is stale.

Partition tolerance (P)

The system keeps operating when messages between nodes are lost. Any real network needs this.

Three misreadings to avoid. A single-node database is not "CA": it is not distributed, so CAP does not apply. Most systems are not one letter or the other but tunable per operation: DynamoDB offers strongly consistent reads at double the cost, Cassandra lets each query choose its consistency level. And "available" in CAP is a yes-or-no property under partition, not the 99.99 percent you put in an SLO.

PACELC: the trade-off when nothing is broken

PACELC extends CAP with the question that matters every day, not just during outages: when there is no partition, do you pay latency for consistency, or accept stale reads for speed? A replicated write is either acknowledged after every copy has it (slow, consistent) or after one has it (fast, may lag).

flowchart LR
  P{Partition?} -- yes --> PA[Availability]
  P -- yes --> PC[Consistency]
  P -- no, the normal case --> EL[Low latency: answer from the nearest copy]
  P -- no, the normal case --> EC[Consistency: wait for the copies to agree]

Consensus: getting machines to agree

Consensus is how several machines agree on one value, or on one ordered log of values, even when some of them crash or messages are lost. It is the foundation under leader election, distributed locks, configuration stores and replicated state machines. Raft is the algorithm most systems use today; Paxos is its older, harder-to-explain ancestor that solves the same problem.

Raft in plain words

1. One leader at a time. Time is divided into numbered terms. In each term, at most one node is leader; the others are followers. All writes go through the leader.
2. Election. A follower that hears nothing from the leader for a randomised timeout (150 to 300 ms is typical) becomes a candidate, increments the term and asks for votes. A majority of votes makes it leader. Random timeouts are what stop two candidates from splitting the vote forever.
3. Log replication. The leader appends each write to its log and sends it to the followers. Once a majority has written the entry, it is committed: applied to the state machine and acknowledged to the client.
4. Safety. A node can only win an election if its log is at least as up to date as the majority's, so a committed entry is never lost, and any two logs that agree on an entry agree on everything before it.

sequenceDiagram
  autonumber
  participant C as Client
  participant L as Leader
  participant F1 as Follower 1
  participant F2 as Follower 2
  C->>L: SET x = 5
  L->>L: append entry (term 3, index 41)
  L->>F1: AppendEntries(41)
  L->>F2: AppendEntries(41)
  F1-->>L: ok
  Note over L: 2 of 3 have it: majority, entry 41 is committed
  L->>L: apply to state machine
  L-->>C: ok
  F2-->>L: ok (late, does not matter)
  Note over L,F2: next heartbeat tells followers that 41 is committed

stateDiagram-v2
  [*] --> Follower
  Follower --> Candidate: election timeout, no heartbeat
  Candidate --> Leader: majority of votes
  Candidate --> Follower: another leader seen, higher term
  Candidate --> Candidate: split vote, new random timeout
  Leader --> Follower: discovers a higher term

Why a majority

Any two majorities of the same set overlap in at least one node. That single fact is what carries knowledge across elections: the new leader's majority must include someone who saw the old leader's committed entries. It also sets the arithmetic: a cluster of 2F + 1 nodes survives F failures, so 3 nodes tolerate 1 and 5 tolerate 2. Even numbers add cost without adding tolerance, which is why clusters are 3 or 5.

Quorums without a leader

Leaderless stores use a related overlap trick without full consensus. With N replicas, a write waits for W acknowledgements and a read asks R replicas; if W + R is greater than N, every read overlaps every write and sees the latest version. Cassandra's QUORUM is W = R = ⌈(N + 1) / 2⌉. This gives read-your-writes but not the ordering guarantees of a consensus log, because two concurrent writes can still race; version vectors or last-writer-wins settle those.

Replicated state machine

Every node applies the same log in the same order, so every node ends in the same state. Consensus is about agreeing on the log.

Term

Raft's logical clock. Every message carries the sender's term; a higher term always wins.

Quorum

The minimum number of nodes that must agree for an operation to count. Usually a majority.

Idempotency: safe to do twice

An operation is idempotent when performing it several times has the same effect as performing it once. In a distributed system you never know whether a timed-out request was processed, and brokers deliver at least once, so every write path that matters must be safe to repeat.

sequenceDiagram
  autonumber
  participant C as Client
  participant S as Payment service
  participant D as Store
  C->>S: POST /charge, Idempotency-Key: 7f3a, $40
  S->>D: is 7f3a known? no
  S->>D: charge $40, save (7f3a, result)
  S--xC: response lost in the network
  C->>S: retry: POST /charge, Idempotency-Key: 7f3a, $40
  S->>D: is 7f3a known? yes
  S-->>C: same response as before, nothing charged again

Ways to get there

• Naturally idempotent operations. "Set status to shipped" and "add user to set" can run any number of times. "Increment balance by 40" cannot. Prefer the first shape where the domain allows.
• Idempotency keys. The client generates a unique id per logical operation and sends it with every attempt. The server stores the key with the result and replays the stored response on a repeat. The store must be checked and written atomically (a unique constraint or a conditional put) and keys expire after a day or so.
• Conditional writes. Include a version: UPDATE ... WHERE version = 7, HTTP If-Match, DynamoDB condition expressions. A repeat fails the condition instead of applying twice.
• Deduplication at the consumer. Message consumers keep a table of processed message ids and skip repeats, ideally in the same transaction as the work.
• The outbox pattern. To publish an event and update the database atomically without distributed transactions, write the event into an outbox table in the same local transaction as the change. A relay then publishes outbox rows to the broker, at least once. Consumers dedupe. Together this gives "effectively once".

suspend fun charge(req: ChargeRequest, key: String): ChargeResult {
    idempotency.get(key)?.let { return it }               // seen before: replay the stored result
    return db.transaction {
        // the unique index on idempotency_keys(key) makes the check-and-insert atomic:
        // a concurrent duplicate fails here and is answered from the store on its retry
        idempotencyKeys.insert(key, status = "in_progress")
        val result = payments.charge(req)
        outbox.insert(PaymentCharged(result.id))            // same transaction as the change
        idempotencyKeys.complete(key, result)
        result
    }
}

Circuit breakers, timeouts and bulkheads

A circuit breaker watches calls to a dependency and, once enough of them fail, stops making calls for a while so that the caller fails fast and the dependency gets room to recover. Timeouts and bulkheads are its two companions: a timeout bounds how long any call may wait, and a bulkhead bounds how many callers can wait on one dependency at once.

stateDiagram-v2
  [*] --> Closed
  Closed --> Open: failure rate over threshold in the sliding window
  Open --> HalfOpen: wait duration elapsed
  HalfOpen --> Closed: trial calls succeed
  HalfOpen --> Open: a trial call fails
  note right of Closed: calls flow, outcomes are counted
  note right of Open: calls fail immediately, fallback runs
  note right of HalfOpen: a few probe calls are let through

• Configuration. A sliding window (last 100 calls or last 10 seconds), a failure threshold (50 percent), a minimum number of calls before judging, and an open duration (30 seconds). Slow calls count as failures if they exceed a threshold.
• Fallbacks. When open, return a cached value, a default, or a degraded response: show the product page without recommendations rather than no page.
• Timeouts everywhere. Every network call needs a deadline, and the deadline should travel with the request so that a downstream call does not wait longer than its caller will. gRPC deadlines and Go contexts do this by default.
• Bulkheads. A separate, bounded pool or semaphore per dependency. If the recommendation service hangs, it exhausts its own ten permits, not the whole server.
• Retries only behind a breaker. Retrying into a failing dependency multiplies its load; the breaker is what turns retries from a hazard into a tool.

Throttling: rate limiting and load shedding

Throttling caps how many requests a client, a tenant or the whole system may make per unit of time, and rejects the rest with a clear signal. It protects the system from abuse, from accidental floods and from its own popularity, and it makes capacity a number you can reason about.

flowchart LR
  R[refill: 100 tokens per second] --> B[("bucket<br/>capacity 500")]
  Q[request arrives] --> T{token available?}
  B --- T
  T -- yes: take one --> A[allow]
  T -- no --> D["reject: 429, Retry-After"]

Where and how

• At the edge (gateway, CDN) for per-client limits, so rejected traffic never reaches the services.
• Distributed state in Redis: one key per client, an atomic script that checks and decrements, a TTL that refills. Local approximate limiters per node are cheaper and good enough for coarse protection.
• Tell the client. 429 Too Many Requests, a Retry-After header, and RateLimit-Remaining style headers so well-behaved clients slow down before being cut off.
• Load shedding is throttling's sibling: when the system as a whole is over capacity, drop the cheapest-to-lose work (background refreshes, low-priority tenants) before the important work, and drop early rather than after doing half the work.

Backoff and jitter: retrying without a stampede

Retrying a failed request is the right instinct and the wrong reflex. If every client retries at the same fixed delay, a brief outage turns into a synchronised wave that hits the recovering service at exactly the same instant. Exponential backoff spreads retries over time; jitter spreads them across clients.

• Exponential backoff. Wait base × 2^attempt, capped at a maximum: 100 ms, 200, 400, 800, up to say 20 s.
• Full jitter. Wait a random value between 0 and that backoff. In AWS's measurements it produced the fewest total calls and the fastest recovery of the variants tried; "equal jitter" (half fixed, half random) is the common alternative.
• Retry budgets. Cap retries as a fraction of traffic (for example at most 10 percent of requests may be retries). When the budget is spent, fail fast. This is what stops retries from turning a 20 percent overload into a 100 percent one.
• Retry the right things. Timeouts, 503s, connection resets and 429s (after Retry-After) are retriable. A 400 or a validation error never is. Only idempotent operations may be retried blindly; everything else needs an idempotency key first.
• Retry at one layer. If the client, the gateway and the service each retry three times, one failure becomes 27 calls. Decide where retries live and make the other layers pass failures through.

suspend fun <T> withRetry(maxAttempts: Int = 5, base: Duration = 100.milliseconds, cap: Duration = 20.seconds,
                          retriable: (Throwable) -> Boolean, block: suspend () -> T): T {
    var attempt = 0
    while (true) {
        try {
            return block()
        } catch (e: Throwable) {
            attempt++
            if (attempt >= maxAttempts || !retriable(e)) throw e
            val backoff = minOf(cap, base * (1L shl attempt))          // exponential, capped
            delay(Random.nextLong(0, backoff.inWholeMilliseconds))     // full jitter
        }
    }
}

Heartbeats, leases and failure detection

A heartbeat is a periodic "I am alive" message, and failure detection is the art of deciding that its absence means death rather than delay. Nothing in a distributed system can tell the two apart with certainty, so every detector is a timeout, and every timeout is a bet.

sequenceDiagram
  autonumber
  participant L as Leader (old)
  participant S as Lock service
  participant N as Leader (new)
  participant D as Storage
  L->>S: renew lease (every 5 s)
  S-->>L: lease valid until t+15, token 33
  Note over L: long GC pause, no renewals
  S->>S: lease expired at t+15
  N->>S: acquire lease
  S-->>N: granted, token 34
  N->>D: write with token 34
  Note over L: pause ends, still believes it is leader
  L->>D: write with token 33
  D--xL: rejected: token 33 is older than 34

The pieces

• Heartbeats and timeouts. Send every T; declare a node suspect after K missed beats. A short timeout detects failure fast but flags every garbage-collection pause and network blip; a long one is safe but slow. Typical settings are seconds for services and tens of seconds for nodes.
• Adaptive detectors. Instead of a fixed timeout, track the distribution of heartbeat gaps and compute how suspicious a silence is. Cassandra's phi accrual detector is the classic version.
• Gossip. Each node periodically shares what it knows about membership with a few random peers. No central monitor, and knowledge spreads in logarithmic time. Cassandra, Consul (Serf) and DynamoDB's internals use it.
• Leases. A lock or leadership that is valid only until a time. The holder must renew; if it cannot, the lease expires and someone else may take it. Leases turn "is it dead?" into "did it renew?", which is answerable.
• Leader election. A consensus store (etcd, ZooKeeper) grants the lease to one candidate; everybody else watches and waits. Kubernetes controllers, Kafka's controller and many schedulers work this way.
• Fencing tokens. Every lease grant carries an increasing number. Anything the leader writes to must check that the token is not older than the last one it saw. This is what defeats the paused-but-alive old leader in Figure 10.

Split brain

Two nodes both believing they are leader is split brain, and it is the failure mode every mechanism above exists to prevent. It happens when a detector declares a live node dead: the node keeps working, a replacement is started, and both write. Prevention is layered: elect through a majority so two leaders cannot both hold a quorum; expire leadership with leases so a partitioned leader steps down on its own; and fence at the storage so a stale leader's writes are refused even if it never notices.

Lease

Ownership with an expiry. Must be renewed; expires on its own if the holder goes silent.

Fencing token

A monotonically increasing number attached to each lease grant, checked by the resource being protected.

Phi accrual

A failure detector that outputs a suspicion level from the history of heartbeat intervals instead of a fixed yes or no.

Recap