distributed-systems-patterns

Files (47)

• SKILL.md
• aea-011-lessons-learned-from-10-years-of-dynamodb.md · references/
• aea-019-scaling-bitbucket-s-database.md · references/
• aea-055-dropbox-s-scalable-consistent-metadata-caching-solution.md · references/
• aea-060-how-ebay-exports-billion-scale-graphs-on-transactional-graph.md · references/
• aea-066-how-ebay-loads-and-updates-over-ten-billion-vertex-graphs.md · references/
• aea-067-ebay-s-real-time-and-performant-index-service-for-its-large-.md · references/
• aea-081-cache-made-consistent-how-meta-handles-cache-invalidation.md · references/
• aea-082-a-highly-available-strongly-consistent-storage-service-using.md · references/
• aea-084-how-we-built-a-general-purpose-key-value-store-for-facebook-.md · references/
• aea-095-how-figma-reduced-potential-instability-by-scaling-to-multip.md · references/
• aea-098-flipkart-s-mysql-highly-available-setup.md · references/
• aea-114-answering-billions-of-reporting-queries-each-day-with-low-la.md · references/
• aea-149-upscaling-linkedin-s-profile-datastore-while-reducing-costs.md · references/
• aea-175-netflix-s-key-value-data-abstraction-layer.md · references/
• aea-176-netflix-s-timeseries-data-abstraction-layer.md · references/
• aea-197-lessons-learned-from-sharding-postgres-at-notion.md · references/
• aea-244-horizontally-scaling-the-rails-backend-of-shop-app-with-vite.md · references/
• aea-263-scaling-datastores-at-slack-with-vitess.md · references/
• aea-312-mysql-at-uber.md · references/
• aea-321-how-uber-ensures-apache-cassandra-s-tolerance-for-single-zon.md · references/
• aea-322-how-ledgerstore-supports-trillions-of-indexes-at-uber.md · references/
• aea-324-how-uber-serves-over-40-million-reads-per-second-from-online.md · references/
• aea-325-how-uber-optimized-cassandra-operations-at-scale.md · references/
• aea-337-building-a-cost-effective-logging-platform-for-petabyte-scal.md · references/
• asdr-006-consistent-hashing.md · references/
• asdr-007-cap-theorem.md · references/
• asdr-008-failover.md · references/
• asdr-017-checksums.md · references/
• asdr-029-database-sharding.md · references/
• asdr-030-data-replication.md · references/
• asdr-031-database-scaling.md · references/
• asdr-033-bloom-filters.md · references/
• asdr-036-caching-strategies.md · references/
• asdr-037-cache-eviction-policies.md · references/
• asdr-038-distributed-caching.md · references/
• asdr-043-heartbeats.md · references/
• asdr-046-distributed-locking.md · references/
• asdr-047-gossip-protocol.md · references/
• asdr-055-peer-to-peer-p2p-architecture.md · references/
• asdr-062-strong-vs-eventual-consistency.md · references/
• htn-databases-in-depth-part-1.md · references/
• htn-databases-in-depth-part-3.md · references/
• htn-design-a-distributed-cache.md · references/
• htn-design-a-distributed-key-value-store.md · references/
• htn-design-consistent-hashing.md · references/
• htn-tricky-hld-patterns-for-interviews.md · references/

SKILL.md (first lines)

Design and review distributed data tiers — how keys find nodes, how data survives node loss, how consistency is tuned per workload, and what storage engine each node runs under the hood.

## When to use

- Designing or reviewing data placement for a horizontally scaled store (distributed cache, KV store, sharded database)
- Choosing a partitioning scheme (hash vs range vs directory) or diagnosing a hot partition
- Deciding whether to shard at all — climbing the scaling ladder in the right order
- Tuning replication and quorums (N/W/R) to a workload's actual correctness requirement instead of one global mode
- Debugging stale reads caused by replication lag (read-your-writes violations after a write)
- Designing a distributed cache cluster: placement, write strategy, eviction, invalidation, hot keys, herd protection
- Adding failure detection and membership (heartbeats, gossip) to a clustered service
- Choosing between B-tree and LSM storage for a read-heavy vs write-heavy workload
- Weighing cross-shard atomicity options (two-phase commit vs sagas) after a sharding decision
- Relieving a read or throughput bottleneck with materialized views or batching before reaching for a reshard
- Explaining why adding or removing one node reshuffled keys, wiped a cache tier, or overloaded a neighbor

Scope boundary — this skill owns **data placement and replication mechanics**. Adjacent territory is owned elsewhere:

- The partition trade-off itself (CAP/PACELC), clock/ordering correctness, retries, circuit breakers, and backpressure belong to `resilience-engineering` (always loaded) — sections below point there instead of restating it.
- System-scale composition (gateways, queues, fan-out, rate limiting) belongs to the `system-design-patterns` skill.
- Concrete cache implementation in a running service belongs to `redis-caching-patterns`; single-node transactions and locking belong to `python-dao-and-database`.

## Consistent hashing

1. **Know why the obvious scheme fails before replacing it.** `hash(key) mod N` routes a key to one of N nodes with no lookup table and no coordination — every client computes the owner independently, in constant time. Its hidden assumption is that N never changes. It is acceptable only when membership is genuinely fixed or remapping is free (stateless routing).

2. **The mod-N reshuffle is a cluster wipe.** The moment a node joins or dies, the modulo arithmetic reassigns roughly (N−1)/N of all keys — almost the entire keyspace changes owner in one instant. For a cache tier that is equivalent to flushing the cluster: every request misses simultaneously and falls through to the database, and the resulting cold-cache stampede can cascade into a full outage. Any placement scheme that bakes the cluster size into the hash turns node failure — an inevitable event — into a remap-everything event. Design for membership churn as a normal operation, not an exception.

3. **The ring.** Consistent hashing maps servers and keys into the same circular hash space (conceptually 0 to 2^32−1, wrapping around). Each server hashes an identifier (hostname/IP/ID) to a ring position; each key hashes to its own position; a key's owner is the first server encountered walking clockwise.

    ```
            0/2^32
              |
        S3 ---+--- k1 → owner: S1 (first server clockwise)
       /              \
      k4               S1
      |                 |
      S2 ------ k3 ---- k2 → owner: S2
                (k3 → owner: S2)
    ```

    Implement lookup as a sorted array (or balanced tree) of server positions plus a binary search for the first position ≥ hash(key), wrapping to the smallest position past the top — O(log N) per lookup, no central directory.

4. **Bounded key movement is the whole point.** Adding a server claims only the arc between it and its counterclockwise predecessor; removing a server hands its arc to the next clockwise node. With K keys on N servers, a membership change moves roughly K/N keys — every other node is untouched. Compare that to the near-total remap of mod-N: this property is what makes scale-out and failure recovery routine operations instead of incidents.

5. **The basic ring has three predictable defects.** Server positions come from hashing a handful of names, so with few points on the circle:
    - **Uneven arcs** — one node can own half the ring while another owns a sliver; load is wildly unbalanced.
    - **Neighbor-dump on failure** — a dead node hands its *entire* arc to exactly one clockwise neighbor, which can overload that neighbor and start a domino of failures.
    - **No capacity weighting** — there is no way to give a bigger machine a bigger share.

    These are geometric consequences of placing few points on a ring, not rare edge cases.

6. **Virtual nodes fix all three.** Place each physical server on the ring many times by hashing derived identifiers (`node-a#1`, `node-a#2`, …); 100–200 vnodes per physical machine is a common operating point. Each machine then owns many small scattered arcs whose sizes average out:
    - **Even load** — aggregate ownership per machine converges to near-equal shares even in small clusters.
    - **Graceful failure** — a dead node's ~150 scattered arcs are absorbed by many different physical neighbors, spreading its load thin instead of doubling one victim's traffic.
    - **Smooth scale-out** — a joining node takes small slices from many existing nodes rather than one big arc from a single neighbor.
    - **Capacity weighting for free** — give a machine with twice the RAM/CPU twice the vnodes and it attracts roughly twice the keys; no separate weighting logic needed.

7. **Tune the vnode count, don't max it.** The ring's position index grows by the vnode multiplier — more entries to binary-search and more membership metadata to keep in sync across the cluster. That is why vnode counts are an operating point to tune, not a number to crank arbitrarily high.

8. **Where the pattern shows up.** Cassandra's token ring (vnodes included) and DynamoDB's partitioning are direct implementations; CDNs route requests to edge nodes this way with no central lookup table; load balancers use it so sticky sessions survive backend scale-up/down with only the affected arcs remapping. The progression mod-N → basic ring → vnodes is worth internalizing as a method: each design is adequate until scale breaks one of its assumptions, and the successor targets exactly that failure mode.

## Partitioning and sharding

1. **Climb the scaling ladder in order — sharding is the last rung.** A single primary with good indexes and query discipline carries far more traffic than intuition suggests. Exhaust the cheaper rungs first:

    | Rung | Move | What it buys |
    |------|------|--------------|
    | 1 | Optimize the single node | Indexes, query shape, pooling — often 10× for free |
    | 2 | Cache the hot read set | Absorbs the read multiplier before it reaches the DB |
    | 3 | Add read replicas | Horizontal read scale; primary handles writes only |
    | 4 | Scale vertically | A bigger box is operationally free vs resharding |
    | 5 | Shard | Only when write volume or working set exceeds all of the above |

    The classic era of URL shorteners and early social platforms served enormous traffic for years on rungs 2–3 alone. Sharding buys write scale at the permanent price of losing cross-shard transactions and joins, complicating every operational procedure, and making key-distribution mistakes expensive — take that deal only when forced.

2. **Read replicas: the workhorse of rung 3.** One primary handles all mutations; asynchronous replication fans data out to replicas that serve selects, search, reporting, and analytics. Read-heavy products (feeds, profiles, content platforms, public APIs) scale a long way on this rung. Two costs come with it: replication lag means read-your-writes is not guaranteed (see point 5), and primary failover/promotion is the operationally hard part — rehearse it before you rely on it.

3. **Three ways to split, with distinct failure modes:**

    | Scheme | How | Strength | Weakness |
    |--------|-----|----------|----------|
    | Hash | Route by hash of partition key | Even key spread | Destroys range scans; adjacent keys scatter |
    | Range | Each shard owns a contiguous key range | Cheap range scans, ordered access | Sequential keys hammer the tail shard |
    | Directory | Lookup service maps key → shard | Surgical placement and rebalancing | One more critical, consistent, HA component on every request path |

    Hash sharding should be combined with consistent hashing so membership churn moves only ~1/N of keys. Range sharding on timestamps or auto-increment IDs aims every insert at the newest shard — a hot partition by construction. Directory sharding is the flexible option for tenant-level placement, but the directory's availability now bounds the whole tier's availability.

4. **Hot partitions come from key choice, not bad luck.** A celebrity user, a monotonic key, or a coarse partition key (partition-by-country puts one populous country on one shard) concentrates traffic while other shards idle. Defenses:
    - Pick higher-cardinality, traffic-uniform partition keys in the first place.
    - Salt or split known-hot keys across sub-partitions and merge on read.
    - Isolate whale tenants onto dedicated partitions rather than letting them degrade shared ones.
    - Watch per-partition metrics, not per-cluster averages — a healthy mean hides a burning shard.

5. **Replication lag is a product decision, not a footnote.** With async replicas, a user can write to the primary and immediately read stale data from a replica — the "my edit disappeared" bug. Route lag-sensitive reads (a user reading back their own just-written data) to the primary, or to a replica whose reported replication position has passed the client's write; let analytics and search tolerate staleness on any replica. Name the guarantee explicitly ("your own writes are visible immediately; others' writes within X seconds") instead of letting it emerge by accident.

6. **Materialized views are a read-scaling lever that often postpones sharding.** When the read bottleneck is an expensive aggregation (dashboards, reports, recommendation summaries) rather than raw row lookups, precompute the query result into its own table and serve reads from that. Three refresh strategies, by staleness budget:
    - **Scheduled** — recompute every N minutes/hours; simplest, staleness bounded by the interval.
    - **Incremental** — recompute only the deltas since the last refresh; cheaper at scale.
    - **Event-driven** — update the view when a relevant write occurs; freshest, most machinery.

    The trade is explicit staleness plus refresh orchestration in exchange for reads that no longer touch the expensive query path. A creator dashboard aggregating views and watch-time periodically — instead of live per page load — is the canonical shape.

7. **Know the cross-shard atomicity cost before you shard.** Once related rows live on different shards, "both updates apply or neither" needs a protocol. Two-phase commit is the classic one: a coordinator asks every participant to *prepare* (validate, lock resources, vote), and only a unanimous yes triggers the global *commit* — any no triggers global rollback. It delivers real atomicity, and its costs are exactly why modern systems avoid it: participants sit blocked holding locks while the protocol runs, the coordinator is both a bottleneck and a single point of failure, latency stacks with participant count, and partial-failure recovery is genuinely messy. Reserve 2PC for the narrow cases where partial success is unacceptable (the debit-and-credit pair of a funds transfer); prefer sagas, event-driven flows, and compensating transactions everywhere else — and let this cost push you back up the scaling ladder: data that transacts together should live on one shard in the first place.

## Replication and quorum

1. **Replicate at write time, not at failure time.** Copying data after a machine dies is a recovery scramble racing the next failure. Instead, every write is sent to the key's primary owner plus its replicas *at the moment it happens* — a replication factor of 3 (one primary, two replicas) is the standard production setting, so losing the primary leaves two live copies and no emergency. The cost is 3× storage and write fan-out, paid deliberately for durability and read availability during failures.

2. **Choose a topology: leader-follower or leaderless.**
    - **Leader-follower** — one node orders all writes for a partition; followers replay its stream. Simple to reason about, and read-your-writes is solvable by reading the leader. Costs: failover/promotion is a real operational event, and the leader caps write throughput for its partition. The right default for transactional stores.
    - **Leaderless (Dynamo-style)** — any replica accepts writes; quorums reconcile divergence. No failover event and better write availability under node loss, but concurrent writes now genuinely conflict, and the system needs the conflict machinery in the KV-store section below. Suits high-availability KV workloads.

3. **N/W/R — the consistency knob, tuned per workload.** In a quorum system: N is the replica count, W how many replicas must acknowledge a write, R how many are consulted on a read. When **W + R > N**, every read set overlaps every write set in at least one replica, so a read is guaranteed to see the latest acknowledged write — read-your-writes by arithmetic, no timestamps involved.

    ```
    N=3, W=2, R=2:
    replicas:   [R1]   [R2]   [R3]
    write set:  {R1, R2}          — acked by 2 of 3

From ajyadav013/claude-kit, MIT.

More from ajyadav013/claude-kit