Distributed systems reasoning

advancedCommon atAnthropicGoogleAmazonMetaStripeRead firstHash maps Load balancing and consistent hashing Binary trees Practise in the Sandbox →

Sharding splits data so each machine holds a part; replication copies each part so a crash loses nothing. With N copies, writes acknowledged by W and reads that ask R, R + W > N forces every read to overlap the latest write. There is no shared clock, so order comes from logical clocks or ids with the time in the high bits, and replicas find their differences cheaply with Merkle trees: O(d log n) hash comparisons for d differing ranges instead of n.

The idea in plain words

Picture a library too big for one building. You split the books across branches by some rule (A–G here, H–M there): that’s sharding, also called partitioning. Then you keep copies of each branch’s books in two other branches, so a fire doesn’t lose anything: that’s replication.

Both create new problems that a single machine never has:

Which branch has the book? You need a rule that maps a key to a shard, and that rule must survive branches opening and closing.
Are the copies the same? Copies drift: one branch got the new edition, another hasn’t yet. Someone asking the wrong branch reads an old copy.
What happened first? Each branch has its own wall clock, and they disagree by milliseconds or more. “Latest” is not as simple as comparing timestamps.

Distributed-systems reasoning is mostly asking, for every operation: how many machines must agree, what if one is slow or dead, and what if two clocks disagree?

See it

Five writes go out over time; reads happen in between. Each replica applies a write after its own network delay. Start with N=3, W=1, R=1 and run the clock: watch reads come back with an old version. Then raise W or R until R + W > N and the stale reads disappear. Click a replica name to crash it and see which operations are now refused.

Replicas, quorums and stale reads

run the clock

N replicas = 3W (write acks) = 1R (read asks) = 1

writes

reads

R + W2 ≤ N = 3

newest written–

newest acked–

stale reads0

refused ops0

has newest write / fresh readreplica lagging behindstale or refusedreplicas this read asked

Writes go out at t = 2, 9, 16, 23, 30; reads happen in between. Press Run, or step with →. Click a replica name to crash it.

t = 0speed

What to notice:

With W=1 the client hears “OK” as soon as the fastest replica has the write. The others are still behind (amber). A read that happens to ask only lagging replicas returns old data.
With W=2, R=2 on 3 replicas, any two replicas a read asks must include at least one of the two that acknowledged. The read can still see a slow replica, but it takes the highest version, so it’s never stale.
Crashing one replica under W=3 refuses every write. That’s the trade: bigger quorums mean fresher reads and lower availability.

When to reach for it

The problem says…	What to think about
“data is split across many shards / nodes”	The partitioning rule, hot spots, and which queries must ask every shard.
“each shard can only answer simple questions”	Every message costs; count round trips and design the protocol around them.
“replicas”, “eventually consistent”, “N copies”	Quorums (R, W, N), what “acknowledged” means, and stale-read windows.
“unique ids without coordination”, “roughly time-ordered”	Pack time in the high bits, a machine id in the middle, a counter at the bottom.
“find what differs between two copies cheaply”	Hash trees (Merkle trees): compare top-down, descend only where hashes differ.
“which event happened first?” across machines	Don’t trust wall clocks; use logical clocks or a single sequencer.

How it works

1. Partitioning: pick the key, then the rule

There are two basic rules for mapping a key to a shard:

Rule	Good at	Bad at
Range (keys `a–g` on shard 0…)	Range scans (`users ada–dee`), ordered iteration	Hot spots when writes cluster (timestamps, auto-increment ids)
Hash (`hash(key)` picks the shard)	Spreading load evenly	Range queries, which must ask every shard (scatter-gather)

Rebalancing matters as much as the rule. hash(key) % N reshuffles most keys when N changes; consistent hashing or a fixed number of many small partitions (say 256 partitions on 8 nodes, moved whole) keeps moves small.

Try the three rules on the same traffic. Every new write is stamped “today”, which is exactly how real time-series workloads behave.

Partitioning: where do writes and queries land?

interactive

Shard 0 holds days 0–7, … shard 3 the most recent days.

one celebrity user's share = 0%salt the hot key (split "max" into 4 sub-keys)

Write load from 400 new events (all happen today):

shard 0

shard 1

shard 2

shard 3

400

one user, all days

4 shards (scatter-gather)

users ada–dee (a range)

4 shards (scatter-gather)

everyone, last 3 days

1 shard

Every new write is "today", so one shard takes all of them while three sit idle: a hot spot. Range-by-time is great for "last 3 days" queries and terrible for write spread.

Hashing doesn’t fix one hot key: all writes for max still hash to one shard. The standard fix is salting, appending a small suffix (max#1…max#4) so writes spread; the cost is that reads for that key must ask all four sub-keys and merge.

2. Replication and quorums

Every key lives on N replicas (often 3).
A write is sent to all N and counted as successful once W of them confirm.
A read asks R replicas and keeps the value with the highest version.
If R + W > N, the read set and the write set must share at least one replica, so the read sees the latest acknowledged write. If not, it may miss it.

The overlap rule is just counting (pigeonhole), and you can check it by brute force over every possible write set and read set:

from itertools import combinations


def quorums_overlap(n: int, w: int, r: int) -> bool:
    """Brute-force check: does EVERY write set of size w share a replica with EVERY read set of size r?"""
    replicas = range(n)
    for write_set in combinations(replicas, w):            # 1. every way a write can land
        for read_set in combinations(replicas, r):         # 2. every way a read can be served
            if not set(write_set) & set(read_set):         # 3. disjoint: this read can miss the write
                return False
    return True


def read_latest(versions: list[int], read_set: list[int]) -> int:
    """A quorum read asks r replicas and keeps the highest version it sees."""
    return max(versions[i] for i in read_set)              # 4. the newest copy wins


if __name__ == "__main__":
    # The rule of thumb r + w > n matches the brute force for every small configuration.
    for n in range(1, 7):
        for w in range(1, n + 1):
            for r in range(1, n + 1):
                assert quorums_overlap(n, w, r) == (r + w > n), (n, w, r)

    # n=3, w=2: version 7 reached replicas 0 and 1; replica 2 still has version 6.
    versions = [7, 7, 6]
    assert all(read_latest(versions, list(s)) == 7 for s in combinations(range(3), 2))  # r=2: always fresh
    assert read_latest(versions, [2]) == 6                                             # r=1: can be stale
    print("ok")

import assert from "node:assert/strict";

/** All k-element subsets of [0, n). Fine for the tiny n used here. */
function* combinations(n: number, k: number, start = 0, acc: number[] = []): Generator<number[]> {
  if (acc.length === k) { yield [...acc]; return; }
  for (let i = start; i < n; i++) yield* combinations(n, k, i + 1, [...acc, i]);
}

/** Brute-force check: does EVERY write set of size w share a replica with EVERY read set of size r? */
export function quorumsOverlap(n: number, w: number, r: number): boolean {
  for (const writeSet of combinations(n, w)) {          // 1. every way a write can land
    const wrote = new Set(writeSet);
    for (const readSet of combinations(n, r)) {         // 2. every way a read can be served
      if (!readSet.some((i) => wrote.has(i))) return false; // 3. disjoint: this read can miss the write
    }
  }
  return true;
}

/** A quorum read asks r replicas and keeps the highest version it sees. */
export function readLatest(versions: number[], readSet: number[]): number {
  return Math.max(...readSet.map((i) => versions[i])); // 4. the newest copy wins
}

// The rule of thumb r + w > n matches the brute force for every small configuration.
for (let n = 1; n <= 6; n++)
  for (let w = 1; w <= n; w++)
    for (let r = 1; r <= n; r++) assert.equal(quorumsOverlap(n, w, r), r + w > n, `${n},${w},${r}`);

// n=3, w=2: version 7 reached replicas 0 and 1; replica 2 still has version 6.
const versions = [7, 7, 6];
for (const s of combinations(3, 2)) assert.equal(readLatest(versions, s), 7); // r=2: always fresh
assert.equal(readLatest(versions, [2]), 6);                                    // r=1: can be stale
console.log("ok");

// The rule of thumb r + w > n matches the brute force for every small configuration.
for (int n = 1; n <= 6; n++)
    for (int w = 1; w <= n; w++)
        for (int r = 1; r <= n; r++)
            if (Quorum.Overlap(n, w, r) != (r + w > n)) throw new Exception($"mismatch at {n},{w},{r}");

// n=3, w=2: version 7 reached replicas 0 and 1; replica 2 still has version 6.
int[] versions = [7, 7, 6];
foreach (var s in Quorum.Combinations(3, 2))
    if (Quorum.ReadLatest(versions, s) != 7) throw new Exception("r=2 should always be fresh");
if (Quorum.ReadLatest(versions, [2]) != 6) throw new Exception("r=1 can be stale");
Console.WriteLine("ok");

static class Quorum
{
    // All k-element subsets of [0, n). Fine for the tiny n used here.
    public static IEnumerable<int[]> Combinations(int n, int k, int start = 0, int[]? acc = null)
    {
        acc ??= [];
        if (acc.Length == k) { yield return acc; yield break; }
        for (int i = start; i < n; i++)
            foreach (var c in Combinations(n, k, i + 1, [.. acc, i])) yield return c;
    }

    // Brute-force check: does EVERY write set of size w share a replica with EVERY read set of size r?
    public static bool Overlap(int n, int w, int r)
    {
        foreach (var writeSet in Combinations(n, w))           // 1. every way a write can land
            foreach (var readSet in Combinations(n, r))        // 2. every way a read can be served
                if (!writeSet.Intersect(readSet).Any()) return false; // 3. disjoint: this read can miss the write
        return true;
    }

    // A quorum read asks r replicas and keeps the highest version it sees.
    public static int ReadLatest(int[] versions, int[] readSet) => readSet.Max(i => versions[i]); // 4. newest wins
}

Common settings: N=3, W=2, R=2 (balanced); W=N, R=1 (fast reads, any dead node blocks writes); W=1, R=1 (fast and available, eventually consistent). Leader-based systems (one primary per shard, followers copy its log) get the same choice through synchronous vs asynchronous followers.

3. Clocks and ordering

Wall clocks on different machines drift and get corrected by NTP, sometimes backwards. Two rules follow:

Measure durations with a monotonic clock (time.monotonic(), performance.now(), Stopwatch), never with wall-clock subtraction.
Don’t decide “which write wins” by comparing timestamps from different machines unless you accept that a skewed clock can silently drop a write (last-write-wins).

Logical clocks order events without real time. A Lamport clock is one counter per node: bump it on every event, and on receive set it to max(local, received) + 1. If a caused b, then L(a) < L(b). A vector clock keeps one counter per node and can also tell you when two events are concurrent, meaning neither saw the other, which is a genuine conflict to resolve:

A local event bumps only your own entry.
On receive, take the element-wise max with the sender’s clock, then bump your own entry.
a happened before b exactly when every entry of a ≤ the same entry of b, and they’re not equal.
If neither is ≤ the other, the events are concurrent.

from typing import Literal

Clock = dict[str, int]  # node id -> number of events that node has seen from itself
Order = Literal["before", "after", "equal", "concurrent"]


def tick(clock: Clock, node: str) -> Clock:
    """1. A local event (or a send) bumps only this node's own entry."""
    return {**clock, node: clock.get(node, 0) + 1}


def merge(local: Clock, received: Clock, node: str) -> Clock:
    """2. On receive: take the element-wise max, then count the receive as a local event."""
    merged = {k: max(local.get(k, 0), received.get(k, 0)) for k in local.keys() | received.keys()}
    return tick(merged, node)


def compare(a: Clock, b: Clock) -> Order:
    """3. a happened before b iff a <= b in every entry and a != b. Neither way round: concurrent."""
    keys = a.keys() | b.keys()
    a_le_b = all(a.get(k, 0) <= b.get(k, 0) for k in keys)
    b_le_a = all(b.get(k, 0) <= a.get(k, 0) for k in keys)
    if a_le_b and b_le_a:
        return "equal"
    if a_le_b:
        return "before"
    if b_le_a:
        return "after"
    return "concurrent"  # 4. a real conflict: someone (app or user) must reconcile


if __name__ == "__main__":
    a1 = tick({}, "A")              # A writes          {A:1}
    b1 = merge({}, a1, "B")         # B receives it      {A:1, B:1}
    a2 = tick(a1, "A")              # A writes again, without hearing from B  {A:2}
    b2 = tick(b1, "B")              # B writes           {A:1, B:2}
    assert compare(a1, b1) == "before"
    assert compare(b1, a1) == "after"
    assert compare(a2, b2) == "concurrent"   # neither saw the other's latest write
    a3 = merge(a2, b2, "A")         # A hears B: now A has seen everything
    assert compare(b2, a3) == "before" and compare(a2, a3) == "before"
    print("ok")

import assert from "node:assert/strict";

type Clock = Readonly<Record<string, number>>; // node id -> number of events that node has seen from itself
type Order = "before" | "after" | "equal" | "concurrent";

/** 1. A local event (or a send) bumps only this node's own entry. */
export function tick(clock: Clock, node: string): Clock {
  return { ...clock, [node]: (clock[node] ?? 0) + 1 };
}

/** 2. On receive: take the element-wise max, then count the receive as a local event. */
export function merge(local: Clock, received: Clock, node: string): Clock {
  const merged: Record<string, number> = { ...local };
  for (const [k, v] of Object.entries(received)) merged[k] = Math.max(merged[k] ?? 0, v);
  return tick(merged, node);
}

/** 3. a happened before b iff a <= b in every entry and a != b. Neither way round: concurrent. */
export function compare(a: Clock, b: Clock): Order {
  const keys = new Set([...Object.keys(a), ...Object.keys(b)]);
  const aLeB = [...keys].every((k) => (a[k] ?? 0) <= (b[k] ?? 0));
  const bLeA = [...keys].every((k) => (b[k] ?? 0) <= (a[k] ?? 0));
  if (aLeB && bLeA) return "equal";
  if (aLeB) return "before";
  if (bLeA) return "after";
  return "concurrent"; // 4. a real conflict: someone (app or user) must reconcile
}

const a1 = tick({}, "A");        // A writes          {A:1}
const b1 = merge({}, a1, "B");   // B receives it      {A:1, B:1}
const a2 = tick(a1, "A");        // A writes again, without hearing from B  {A:2}
const b2 = tick(b1, "B");        // B writes           {A:1, B:2}
assert.equal(compare(a1, b1), "before");
assert.equal(compare(b1, a1), "after");
assert.equal(compare(a2, b2), "concurrent"); // neither saw the other's latest write
const a3 = merge(a2, b2, "A");   // A hears B: now A has seen everything
assert.equal(compare(b2, a3), "before");
assert.equal(compare(a2, a3), "before");
console.log("ok");

using Clock = System.Collections.Generic.IReadOnlyDictionary<string, int>;

var a1 = VC.Tick(new Dictionary<string, int>(), "A");  // A writes          {A:1}
var b1 = VC.Merge(new Dictionary<string, int>(), a1, "B"); // B receives it  {A:1, B:1}
var a2 = VC.Tick(a1, "A");                             // A writes again, without hearing from B  {A:2}
var b2 = VC.Tick(b1, "B");                             // B writes           {A:1, B:2}
if (VC.Compare(a1, b1) != "before") throw new Exception("a1 should precede b1");
if (VC.Compare(b1, a1) != "after") throw new Exception("b1 should follow a1");
if (VC.Compare(a2, b2) != "concurrent") throw new Exception("a2 and b2 are concurrent"); // neither saw the other
var a3 = VC.Merge(a2, b2, "A");                        // A hears B: now A has seen everything
if (VC.Compare(b2, a3) != "before" || VC.Compare(a2, a3) != "before") throw new Exception("a3 dominates");
Console.WriteLine("ok");

static class VC
{
    static int Get(Clock c, string k) => c.TryGetValue(k, out var v) ? v : 0;

    // 1. A local event (or a send) bumps only this node's own entry.
    public static Clock Tick(Clock clock, string node) =>
        new Dictionary<string, int>(clock) { [node] = Get(clock, node) + 1 };

    // 2. On receive: take the element-wise max, then count the receive as a local event.
    public static Clock Merge(Clock local, Clock received, string node)
    {
        var merged = new Dictionary<string, int>(local);
        foreach (var (k, v) in received) merged[k] = Math.Max(Get(merged, k), v);
        return Tick(merged, node);
    }

    // 3. a happened before b iff a <= b in every entry and a != b. Neither way round: concurrent.
    public static string Compare(Clock a, Clock b)
    {
        var keys = a.Keys.Union(b.Keys).ToList();
        bool aLeB = keys.All(k => Get(a, k) <= Get(b, k));
        bool bLeA = keys.All(k => Get(b, k) <= Get(a, k));
        if (aLeB && bLeA) return "equal";
        if (aLeB) return "before";
        if (bLeA) return "after";
        return "concurrent"; // 4. a real conflict: someone (app or user) must reconcile
    }
}

4. Ids without coordination

Auto-increment ids need one machine to hand them out. Fleet-wide ids are usually built so any machine can generate them alone:

Scheme	Layout	Sortable by time?	Notes
UUIDv4	122 random bits	No	Collisions practically impossible; random inserts scatter across a B-tree index.
UUIDv7 (RFC 9562)	48-bit Unix ms + random	Yes, to the ms	Same 128 bits, index-friendly.
ULID	48-bit ms + 80 random, base32 text	Yes	Popular text format before UUIDv7.
Snowflake (Twitter, 2010)	1 sign + 41 time + 10 machine + 12 sequence, in 64 bits	Roughly	Fits a `long`; up to 4096 ids per ms per machine; 41 bits of ms ≈ 69.7 years.

Drag the bit split and watch the trade: more worker bits mean more machines but fewer ids per millisecond, or a shorter lifetime. The lower half shows what clock skew does to “sorted by id”.

Anatomy of a time-ordered 64-bit id

drag the split

timestamp (ms)worker idsequence

worker bits = 10sequence bits = 12

bit 63 (top-left) … bit 0 (bottom-right)

timestamp: 41 bitsworker: 10 bitssequence: 12 bitssign bit (kept 0)

id517815304081413

hex0x0001d6f345425005

workers1,024

ids / ms / worker4,096

lifetime69.7 years

Sorted by id = sorted by time? Four events happen in order e1 → e4 on two workers.

worker 2 clock behind = 0 ms

e1 · w1

true 1000 · stamped 1000

e2 · w2

true 1002 · stamped 1002

e3 · w1

true 1003 · stamped 1003

e4 · w2

true 1005 · stamped 1005

Sorting by id gives the true order. The timestamp sits in the high bits, so it dominates the comparison; worker and sequence only break ties within one millisecond.

Because the timestamp sits in the top bits, comparing ids compares times first. That’s why these ids work well as primary keys: new rows append at the end of the index. But they’re only as ordered as the clocks that stamped them, and a generator must decide what to do when its clock jumps backwards or its sequence for one millisecond runs out.

5. Anti-entropy with Merkle trees

Replicas that miss writes must eventually repair. Comparing every key costs O(n) data over the network. A Merkle tree hashes each key range into a leaf, and each parent hashes its children. Two replicas compare root hashes; only where hashes differ do they descend. Amazon’s Dynamo and Apache Cassandra’s repair both use this.

Click keys to change replica B, then step through the comparison:

Merkle tree anti-entropy: find the diff, skip the rest

step through

change replica B's data:

replica A

replica B (capitalised = changed)

hash comparisons0

leaves (naive compare)8

ranges to repair–

hashes differ: descendhashes match: skip subtreediffering range foundnever looked at

Both replicas built the same tree shape over the same key ranges. Sync starts by exchanging just the root hashes.

1 / 9speed

What it costs

Thing	Cost	Why
Quorum write / read	latency of the W-th / R-th fastest replica	You wait for the slowest member of the quorum, not the slowest of all.
Range query on hash partitioning	ask all shards	Neighbouring keys are scattered on purpose.
`hash % N` rebalance	~(N−1)/N of keys move	Almost every key’s remainder changes.
Consistent-hash rebalance	~1/N of keys move	Only the arc next to the new node changes owner.
Merkle sync, d differing leaves of n	O(d log n) hash comparisons	Each difference costs one root-to-leaf path; matching subtrees are skipped whole.
Merkle sync, everything differs	O(n)	No subtree can be skipped; no worse than comparing everything.
Vector clock	O(number of nodes) per event and per message	One counter per participant; grows with cluster size.

A useful rule when shards are remote: count messages, not CPU. A round trip to another machine costs around a millisecond in one data centre, millions of times more than a local operation, so the number of messages (and how many happen in parallel) usually decides the design.

Variations you’ll meet

Leader-follower replication. One leader per shard orders all writes; followers replay its log. Simple, and reads from followers can be stale (replication lag).
Multi-leader / leaderless. Writes accepted in several places; needs conflict resolution: last-write-wins, vector clocks, or CRDTs (data types whose merges always converge).
Consensus (Raft, Paxos). A majority quorum agrees on a single ordered log; this is how you get one true order, at the cost of a round trip to a majority per write.
Read-your-writes / monotonic reads. Weaker guarantees you can provide by routing a user to the same replica, or by tracking the version they last saw.
Two-level keys. Partition by one key, sort by another inside the partition (Cassandra’s partition key + clustering key): one user’s events by time on one shard.
Gossip. Nodes periodically swap state with a random peer; information spreads to everyone in O(log n) rounds.

Pitfalls

Hot partitions. Timestamps or auto-increment ids as a range key send every write to the newest shard.
Treating “acknowledged” as “everywhere”. With W < N, some replicas don’t have the write yet.
Trusting wall clocks for order. NTP can step a clock backwards; last-write-wins then discards newer data.
Forgetting partial failure. A request can succeed on the server and time out on the client. Retries need idempotency.
Scatter-gather tail latency. A query that fans out to 100 shards is as slow as the slowest of the 100.

In the interview

Ask: How big is the data and how many machines? Read-heavy or write-heavy? Is stale data acceptable, and for how long? What happens on a network partition: refuse, or serve possibly stale data?
Say out loud what each message costs and how many you send. “A write waits for 2 of 3 replicas, so its latency is the second-fastest replica’s, and a read fans out to 2.”
Name the trade-off you’re choosing: consistency vs availability during a partition (CAP), and latency vs consistency the rest of the time (PACELC).
Expect follow-ups: “A node dies mid-write, what does the client see?”, “Clocks go backwards by 2 seconds”, “Add a shard: how much data moves?”, “Two replicas disagree; how do you find out where, cheaply?”

Check yourself

pick one

N = 5 replicas. You want reads to always see the latest acknowledged write, and writes to keep working with 2 replicas down. Which setting works?

Check yourself

pick one

You shard an event log by range on the event's timestamp across 8 nodes. What goes wrong first?

Check yourself

pick one

Two replicas each hold 1,024 key ranges in a binary Merkle tree. Exactly one range differs. About how many hash comparisons does a top-down sync need?

Practice on your own

These are classic exercises, not gym questions:

Simulate a Lamport clock across three processes exchanging messages and check that causally related events get increasing timestamps.
Given N, list every (R, W) pair that keeps reads fresh while tolerating f failed nodes for writes, then for reads.
Implement a G-Counter CRDT (one counter per node, merge by element-wise max) and show that merges in any order give the same total.
Design a URL shortener’s id scheme for 1,000 machines and 50,000 ids per second each: how many bits for each field, and how long until overflow?
Compute how many keys move when you go from 4 to 5 shards under hash % N versus a 256-slot fixed-partition table.

Go deeper

Martin Kleppmann, Designing Data-Intensive Applications, ch. 5 (Replication), 6 (Partitioning), 8 (The Trouble with Distributed Systems), 9 (Consistency and Consensus).
DeCandia et al., Dynamo: Amazon’s Highly Available Key-value Store (SOSP 2007): N/R/W quorums, vector clocks, Merkle-tree anti-entropy, hinted handoff.
Leslie Lamport, Time, Clocks, and the Ordering of Events in a Distributed System (1978).
RFC 9562: Universally Unique IDentifiers (UUIDv7 layout), and Twitter’s original Snowflake README (2010) for the 41/10/12 bit layout.

Load balancing and consistent hashingPlacing keys on servers so adding one moves only 1/N of them.

Binary treesThe shape behind a Merkle tree and its top-down comparison.

Time handlingWall clocks vs monotonic clocks, epochs and skew.

IdempotencyMaking retries safe when a request's outcome is unknown.

Distributed systems reasoning

The idea in plain words

See it

When to reach for it

How it works

1. Partitioning: pick the key, then the rule

2. Replication and quorums

3. Clocks and ordering

4. Ids without coordination

5. Anti-entropy with Merkle trees

What it costs

Variations you’ll meet

Pitfalls

In the interview

Check yourself

Practice on your own

Go deeper

Related