Distributed systems cheat sheet

What CAP actually says

CAP says that when a network partition splits replicas, a system must choose between consistency (every read sees the latest write, linearizably) and availability (every request to a live node gets a non-error answer). Partitions aren't optional in a real network, so the real choice is what to do during one: refuse some requests (CP) or answer with possibly stale data (AP). Without a partition you can have both.

Consistency models from strong to eventual

Linearizable: the system behaves like one copy; once a write completes, every later read sees it. Causal: operations that depend on each other are seen in that order by everyone, while unrelated ones may appear in different orders. Read-your-writes: a client always sees its own updates. Eventual: if writes stop, replicas converge, with no promise about when or what you read meanwhile.

Leader-follower replication and lag

All writes go to the leader, which streams its change log to followers; reads can be served by followers. With asynchronous replication a follower can be seconds behind, so a user may not see their own write, see newer data on one request and older on the next, or see an answer before its question. Synchronous replication to at least one follower avoids losing acknowledged writes on failover but adds latency to every commit.

Quorum reads and writes

If every write must be acknowledged by W replicas and every read asks R replicas, then W + R > N means the read set and the write set overlap in at least one replica, so a read normally sees the latest acknowledged write. It's not a full guarantee: concurrent writes, sloppy quorums that write to stand-in nodes, failed writes that partly applied, and clock-based conflict resolution can all still return stale or lost values.

Consensus and Raft

Consensus lets a group of nodes agree on one ordered log of decisions even when some crash, so every replica applies the same commands in the same order. Raft elects a leader for a numbered term by majority vote. The leader appends entries and replicates them; an entry is committed once a majority has stored it. A majority is needed for both, so two leaders can't commit conflicting entries in the same term.

Leader election, leases and fencing tokens

Use a coordination service (etcd, ZooKeeper, Consul) or a database row to grant a lease: a lock with an expiry that the holder must renew. The danger is a leader that stalls, for example in a long GC pause, while its lease expires and a new leader takes over; when it wakes it still thinks it leads. Fencing tokens fix this: each lease grant comes with a number that only goes up, and the storage rejects writes carrying an older number.

class FencedStorage:
    def __init__(self) -> None:
        self.highest = 0
        self.data = ""

    def write(self, token: int, value: str) -> bool:
        if token < self.highest:
            return False
        self.highest, self.data = token, value
        return True


s = FencedStorage()
print(s.write(34, "new"), s.write(33, "stale"), s.data)

Locks in Redis: when they're safe

It's a fine lock for efficiency, where an occasional double run only wastes work. Set it with a random value and release it only if the value still matches, using a small Lua script, so you never delete someone else's lock. It isn't safe for correctness: the holder can pause past the expiry, and a Redis failover can lose the key. If two holders would corrupt data, use fencing tokens checked by the storage or a database constraint.

Consistent hashing

With modulo, changing N changes almost every key's node: in our simulation of 1,000 keys going from 4 to 5 nodes, 807 keys moved, which empties most of a cache at once. Consistent hashing places nodes and keys on a ring and gives each key to the next node clockwise, so adding a node only takes keys from its neighbours: 176 of 1,000 moved with 100 virtual nodes each, close to the ideal 1/5.

Sharding: range, hash and directory

Pick a shard key that most queries include and that spreads load evenly, often the tenant or user id. Hash sharding spreads keys evenly but makes range scans across shards. Range sharding keeps ranges together, good for time series, but new writes can all hit the newest shard. A directory (lookup table) maps keys to shards explicitly and lets you move a big tenant alone. Queries without the shard key become scatter-gather.

Retries with exponential backoff and jitter

Retry only failures that might succeed later (timeouts, 503, 429, connection resets) and only for idempotent operations or ones protected by an idempotency key. Wait longer after each attempt (exponential backoff) with a cap, and randomize each wait (jitter) so clients that failed together don't all retry together. Limit attempts, respect Retry-After, and keep a retry budget so retries can't multiply the load.

Timeouts and deadline propagation

Every network call needs a timeout, or one slow dependency ties up threads until the caller falls over. Set connect and request timeouts from the dependency's measured latency (for example a little above its p99), not a round default. Across a chain, pass a deadline rather than fixed timeouts: each service uses the time left, so a downstream call never keeps working after the user's request has already timed out.

Circuit breakers and bulkheads

A circuit breaker watches calls to a dependency and, after too many failures, opens: calls fail fast for a cooldown period instead of waiting on a service that's down. Then it lets a trial request through (half-open); success closes it, failure opens it again. Retries try harder; a breaker stops trying, which protects both the caller's threads and the struggling dependency. Bulkheads add separate pools per dependency so one can't exhaust them all.

At-most-once, at-least-once and 'exactly once'

Not over an unreliable network: if an acknowledgement is lost, the sender can't tell whether the message arrived, so it either resends (at-least-once, possible duplicates) or doesn't (at-most-once, possible loss). What systems call exactly-once is effectively-once processing: at-least-once delivery plus idempotent handling or deduplication, or a transaction that commits the result and the consumer offset together.

The transactional outbox pattern

Don't write to the database and then publish to the broker as two steps; a crash between them loses the event or publishes one for a rolled-back change. Instead insert the event into an outbox table in the same transaction as the business change. A relay then reads the outbox and publishes, marking rows as sent. Delivery is at-least-once, so consumers must be idempotent.

Sagas: orchestration vs choreography

Use a saga: a sequence of local transactions, each with a compensating action that semantically undoes it (refund, release stock). If a step fails, the saga runs the compensations for the completed steps in reverse. Orchestration puts the sequence in one coordinator that tells each service what to do; choreography has each service react to the previous one's events. Orchestration is easier to follow and change; choreography has less central coupling.

CQRS and read models

Command Query Responsibility Segregation separates the model that handles writes from the models that serve reads. Writes go through one normalized model that enforces the rules; events or change data capture then update read models shaped for each screen or query, possibly in different stores such as a search index. It's worth it when reads and writes have very different shapes or scale. The cost is eventual consistency between the two sides and more moving parts.

Clocks, timestamps and ordering events

Server clocks drift and get stepped by NTP, so a write that happened later can carry an earlier timestamp and be silently discarded. Wall-clock timestamps can't tell you which event caused which. Logical clocks can: a Lamport clock gives an order consistent with causality, and version vectors detect when two writes were concurrent so you can merge them or ask the user. For one database, let it order writes, for example with a version column.

Detecting failures and avoiding split brain

A node is presumed dead when it misses heartbeats for a timeout, but a slow node, a long pause and a network cut all look the same. Short timeouts give false failovers; long ones give slow recovery. Split brain happens when the old primary is cut off but alive and keeps accepting writes. Prevent it by requiring a majority (quorum) to act as primary, fencing the old one (revoking its access or powering it off), and fencing tokens on writes.