Distributed Systems¶
A distributed system has components that communicate over a network and can fail independently. The hard parts are partial failure, uncertain time, concurrent state, and coordination—not the number of machines.
The Working Model¶
Assume that:
- messages can be delayed, duplicated, reordered, or lost;
- a process can pause, restart, or fail while others continue;
- a timeout reveals uncertainty, not whether work occurred;
- clocks drift and can move relative to one another;
- topology, capacity, configuration, and ownership change;
- retries and recovery can overlap with original work.
Design every remote operation around these facts. The classic eight fallacies—reliable network, zero latency, infinite bandwidth, secure network, fixed topology, one administrator, zero transport cost, and homogeneous network—are reminders, not laws.
Delivery and Idempotency¶
Common delivery descriptions are end-to-end properties, not magic broker switches:
- at-most-once: an operation may be lost but is not deliberately retried;
- at-least-once: retries reduce loss but can create duplicates;
- effectively-once: duplicates occur but idempotent processing makes their business effect equivalent to one execution.
Exactly-once effects across arbitrary external systems require those systems to participate in the same protocol or support deduplication. Usually, use stable operation IDs, durable state transitions, unique constraints, and reconciliation.
An idempotency record should bind a key to the caller, operation, request fingerprint, status, and result. Reusing a key for different input must fail.
Timeouts, Retries, and Back-Pressure¶
A timeout is a local decision to stop waiting. The remote side may still complete. Therefore:
- attach an end-to-end deadline;
- retry only transient failures and safe operations;
- use bounded attempts, exponential backoff, and jitter;
- avoid retries at multiple layers;
- cap queues and concurrency;
- shed load before saturation;
- expose queue age and rejected work.
Back-pressure propagates limited capacity toward producers. Without it, buffers postpone failure while increasing latency and memory use.
Consistency Models¶
A consistency model states what results concurrent clients may observe.
Linearizability¶
Each operation appears to take effect atomically between invocation and response, respecting real-time order. It is strong and intuitive but coordination can increase latency or reduce availability during partitions.
Sequential Consistency¶
All clients observe one ordering consistent with each client's program order, but that order need not respect wall-clock order across clients.
Causal Consistency¶
Causally related operations are observed in order; concurrent unrelated operations may appear in different orders.
Eventual Consistency¶
If updates stop and communication continues, replicas converge. This says little about convergence time, conflicts, or what a session observes. Define those separately.
Session Guarantees¶
Read-your-writes, monotonic reads, and monotonic writes often provide useful user guarantees without global linearizability.
Choose consistency per invariant. Inventory display may tolerate staleness; preventing two confirmed owners for one seat may require stronger coordination.
CAP and PACELC¶
During a network partition, a replicated system cannot guarantee both linearizable consistency and availability for every request. “CA” is not a meaningful partition response for a system that must continue operating across a real network; it means avoiding or failing under that partition rather than escaping the trade-off.
PACELC adds that even when there is no partition, systems often trade latency against consistency. These models guide questions; they do not classify an entire product with one permanent label.
Replication¶
Single Leader¶
One leader orders writes and replicas follow. This simplifies conflict handling, but failover must prevent two leaders from accepting incompatible histories. Reads from followers can be stale.
Multi-Leader¶
Several leaders accept writes, often across regions or disconnected sites. It improves local write availability but requires conflict detection and resolution. Last-write-wins can lose valid updates and depends on clocks.
Leaderless¶
Clients or coordinators write to and read from several replicas. Quorums can improve overlap, but R + W > N alone does not guarantee linearizability under sloppy quorums, concurrent writes, clock-based resolution, or failed repair. Read repair and anti-entropy are part of operating the design.
Replication improves availability and read capacity. It is not a backup: corruption or deletion can replicate too.
Partitioning and Sharding¶
Partitioning divides data across nodes.
- range partitioning supports ordered scans but can create hot ranges;
- hash partitioning distributes many workloads but weakens range access;
- directory-based partitioning provides flexible placement but adds lookup state;
- consistent hashing reduces remapping when membership changes but still needs virtual nodes, capacity weighting, and rebalancing.
Choose keys from access and load patterns. Plan for hot keys, cross-partition queries, transactions, resharding, and tenant growth. Rebalancing consumes production bandwidth and should be observable and throttled.
For storage foundations, see Sharding, Replication, and Partitioning.
Consensus¶
Consensus lets non-faulty participants agree on a value or ordered log despite some failures. It commonly supports leader election, replicated state machines, membership, and metadata—not arbitrary business transactions.
Raft organizes consensus around leader election, log replication, and safety. A majority is required to commit entries, so a cluster of three tolerates one unavailable member and a cluster of five tolerates two. More members also add coordination cost. The Raft paper is the primary reference.
Paxos is a family of protocols for reaching agreement; practical replicated logs generally build on repeated or multi-decree forms rather than a single isolated decision.
Consensus does not eliminate application-level duplicates, stale reads, schema compatibility, or disaster recovery.
Distributed Transactions¶
Two-Phase Commit¶
A coordinator asks participants to prepare, then commit or abort. It can provide atomicity across participating resources, but prepared participants may hold locks or remain uncertain while the coordinator is unavailable. Production implementations rely on durable logs and recovery protocols.
Sagas¶
A saga commits a sequence of local transactions and uses compensating actions after failure. Compensation is domain logic, not a general rollback: sending a second email does not unsend the first, and refunds can fail. Persist workflow state and make every step idempotent.
Prefer keeping invariants within one transactional boundary. Use cross-system coordination only when the business requirement demands it.
Events, Ordering, and Schema Evolution¶
Global total order is expensive and usually unnecessary. Define the smallest ordering scope, commonly per aggregate or partition. Include stable event ID, entity ID, schema version, occurrence time, and causation or correlation identifiers where useful.
Consumers must tolerate duplicates and compatible schema changes. Keep events immutable; correct mistakes with new events. A transactional outbox can atomically store a domain change and the intent to publish, followed by asynchronous relay.
Clocks and Causality¶
Physical clocks are useful for user-facing time, expiration, and operations, but unsafe as the sole ordering mechanism for correctness-sensitive concurrent writes.
- Lamport clocks preserve “happened-before” implication but cannot identify concurrency;
- vector clocks can represent causality and concurrency but metadata grows with participants;
- hybrid logical clocks combine approximate physical time with logical ordering.
Use monotonic clocks for elapsed durations on one process. Use protocol state, versions, fencing tokens, or consensus for correctness across processes.
Failure Detection and Membership¶
Failure detectors infer suspicion from missing evidence. A slow node and failed node are indistinguishable within a bounded observation window. Heartbeats, gossip, and adaptive detectors trade detection speed against false positives.
Membership changes need a protocol. Removing a suspected node, electing a replacement, and restoring it later must not permit two owners of the same lease or leadership term. Fencing tokens let downstream resources reject stale leaders.
Repair and Reconciliation¶
Distributed state drifts. Design repair as a normal operation:
- checksums, anti-entropy, or comparison jobs find differences;
- replay and backfill restore derived state;
- dead-letter handling isolates poison data without hiding it;
- reconciliation compares desired and observed business outcomes;
- runbooks define ownership, rate limits, and safe restart points.
Every repair tool needs idempotency and auditability.
Observability and Testing¶
Carry correlation context across calls and messages. Measure latency, timeouts, retries, duplicate rate, queue age, replication lag, leader changes, conflict rate, and repair backlog.
Test more than process crashes:
- latency and packet loss;
- one-way partitions;
- clock skew and pauses;
- duplicate and reordered messages;
- full disks and exhausted connection pools;
- rolling version skew;
- failover followed by recovery of the old leader.
Verify safety invariants as well as eventual recovery. Chaos experiments are useful only with a hypothesis, bounded blast radius, observability, and an abort condition.
Design Checklist¶
- Which invariants require coordination?
- What can be stale, and for how long?
- What does a timeout mean to the caller?
- Are all retried operations safe to repeat?
- Where are ordering guarantees required?
- What happens during a partition and after healing?
- How are conflicts, duplicates, and poison messages handled?
- How is state repaired and reconciled?
- Are recovery and failover tested under realistic version skew?
- Is the added distribution justified by ownership, scale, isolation, or geography?
The simplest reliable distributed design is often to keep related data and invariants together and distribute only where measured constraints require it.