Distributed Systems: Consistency, Consensus, and Failure Handling

A distributed system is many machines working together as if they were one. They give you scale and fault tolerance — and in exchange, they give you network failures, partial outages, and clock problems. This tutorial covers the core concepts and patterns you need to design and reason about them.

Table of Contents

• What Is a Distributed System
• Why Distribution Is Hard: Fallacies and CAP
• Consistency Models
• Replication
• Partitioning (Sharding)
• Consensus and Leader Election
• Time, Clocks, and Ordering
• Failure Handling Patterns
• Distributed Transactions and Sagas

What Is a Distributed System

A distributed system is a set of independent computers that coordinate over a network to appear as a single coherent system to users.

Examples: a database cluster, Kubernetes, Kafka, a microservice fleet, even the internet itself.

Why distribute at all? Single machines hit hard limits on CPU, memory, and — most importantly — reliability. One box failing takes everything down.

Why Distribution Is Hard: Fallacies and CAP

The fallacies of distributed computing

Engineers new to distribution assume things that aren't true:

1. The network is reliable — it isn't.
2. Latency is zero — it isn't.
3. Bandwidth is infinite — it isn't.
4. The network is secure — it isn't.
5. Topology doesn't change — it does.
6. There is one administrator — there isn't.
7. Transport cost is zero — it isn't.
8. The network is homogeneous — it isn't.

The CAP theorem

When a network partition happens, a distributed system must choose:

• Consistency — every read sees the latest write
• Availability — every request gets a response
• Partition tolerance — the system survives network splits

You can only keep two of the three. Since partitions are unavoidable in practice, the real choice is between CP (consistent but may refuse requests) and AP (available but possibly stale).

Bank account service → CP (wrong balance is worse than downtime)
Social feed service   → AP (slightly stale feed is fine)

Consistency Models

Consistency is a spectrum, not a switch:

# Read-your-writes via session pinning
def update_profile(user_id, data):
    db_primary.execute(update_query(user_id, data))
    session.pin_to_primary(user_id, ttl_seconds=30)

def get_profile(user_id):
    if session.pinned(user_id):
        return db_primary.read(user_id)
    return db_replica.read(user_id)

The explanation: after a write, that user's reads go to the primary for a short window. Everyone else can use fast replicas. This one trick eliminates the most common "why doesn't my change show up" bug.

Replication

Replication keeps copies of data on multiple nodes for durability and read scaling.

• Single-leader: one primary accepts writes, replicas copy. Simple, standard.
• Multi-leader: multiple nodes accept writes. Powerful across datacenters, hard to resolve conflicts.
• Leaderless (quorum): clients write to several nodes; reads require a quorum. Used by Cassandra and DynamoDB.

Single-leader flow:
  Client → Leader (write) → async replicate → Replica 1, Replica 2
  Reads → any replica (may be slightly stale)

Quorum math

With N replicas, write to W and read from R. If W + R > N, every read overlaps a recent write — you get consistency. Example: N=3, W=2, R=2.

Partitioning (Sharding)

Partitioning splits data across nodes so no single machine holds everything.

# Simple approach: hash partitioning
def shard_for(user_id: str, shard_count: int) -> int:
    return int(hashlib.md5(user_id.encode()).hexdigest(), 16) % shard_count

The explanation: hashing spreads users evenly. Plain modulo breaks when you add shards (everything remaps), which is why production systems use consistent hashing — adding a node only moves a small fraction of keys.

Partitioning rules of thumb:

• Shard by your main access key (tenant ID, user ID) so queries hit one shard.
• Watch for hot shards — one celebrity customer on one shard ruins the balance.
• Resharding is painful; pick shard count with growth headroom or use consistent hashing.

Consensus and Leader Election

When nodes must agree (who is the leader? is this write committed?), naive approaches fail: two nodes can both think they're leader after a network hiccup ("split brain").

Consensus algorithms — Raft and Paxos — solve this safely.

Raft in one paragraph: nodes elect a leader by majority vote. The leader replicates a log of operations; an entry is "committed" once a majority stores it. If the leader dies, a new election happens with the latest log. Majorities guarantee only one leader can exist per term.

Cluster of 5: majority = 3
Leader receives write → sends to followers → 3 nodes have it → committed
If only 2 nodes are reachable → cluster halts (CP behavior) rather than diverge

Time, Clocks, and Ordering

You cannot use wall clocks to order events across machines — clocks drift and jump.

• Use logical clocks: Lamport timestamps give a partial order of events; vector clocks detect whether two events happened in order or concurrently.
• Use monotonic clocks for measuring durations.
• For ordering guarantees (like Kafka), rely on a single sequence per partition, not timestamps.

# Detecting concurrent updates with vector clocks
# Replica A: {a: 2, b: 1}   Replica B: {a: 1, b: 2}
# Neither dominates → conflict → resolve via merge or last-writer-wins
def happens_before(v1, v2):
    return all(v1.get(k, 0) <= v2.get(k, 0) for k in set(v1) | set(v2)) and v1 != v2

Failure Handling Patterns

Failure detection

Heartbeats + timeouts. A node is "suspect" after missing N heartbeats — but remember, a timeout only proves the network is slow or the node is dead; you can't always tell which.

Failover

Standby nodes take over when primaries die. The dangerous case: the old primary comes back while a new one is active. Prevent with fencing — a monotonically increasing epoch/token that old primaries are no longer allowed to write with.

Backpressure

When consumers can't keep up, producers must slow down rather than queue forever.

if queue.depth > MAX_DEPTH:
    producer.pause()          # or shed load with 503s

Distributed Transactions and Sagas

Classic ACID transactions don't span services. The Saga pattern breaks a workflow into local transactions with compensating actions for rollback.

# Saga: order placement across three services
steps = [
    (create_order,      cancel_order),      # Order service
    (reserve_inventory, release_inventory), # Inventory service
    (charge_payment,    refund_payment),    # Payment service
]

def run_saga(order):
    completed = []
    for action, compensate in steps:
        try:
            action(order)
            completed.append(compensate)
        except StepFailed:
            for undo in reversed(completed):   # roll back what succeeded
                undo(order)
            raise

The explanation: if payment fails, we release the reserved inventory and cancel the order — in reverse order. Each step must be idempotent, since retries may repeat them.