Distributed Systems
CHAPTER 1

Distributed Systems Building Blocks

The reusable primitives every highload service is assembled from. Get these right and horizontal scale, failover and graceful behaviour under stress come almost for free; get them wrong and no amount of hardware saves you. This chapter is the toolbox you draw from in every design round.

StatelessCachingDatabasesLoad balancing RetriesBackpressureIdempotency Circuit breakersDegradation

Map of this chapter: we move from how a request is served (stateless compute + caching + the database tier + load balancing) to how the system protects itself (retries, backpressure, idempotency, circuit breakers) and finally how it fails gracefully (degradation). Each block includes the theory, a concrete example, and a diagram you can redraw on a whiteboard.

🧩

Serve

Stateless instances, layered caches and a smart load balancer turn one box into a fleet.

🛡️

Protect

Retries, backpressure, idempotency and circuit breakers keep a bad minute from becoming an outage.

🪂

Degrade

When you can't do everything, do the important thing — serve stale, drop extras, fail static.

Stateless Service Design

A service is stateless when any instance can handle any request because it keeps no per‑client data in local memory between requests. All durable or shared state lives outside the process.

📈

Horizontal scaling

Add or remove identical instances behind a load balancer with no coordination — capacity scales linearly with the fleet.

♻️

Trivial failover

An instance can die mid‑fleet and traffic just reroutes. No session is lost because no session lived on that box.

🚀

Fast, safe deploys

Rolling and blue/green deploys, autoscaling and spot instances all assume interchangeable, disposable processes.

🧪

Simple reasoning

The 12‑factor rule — “processes are stateless and share‑nothing” — makes each request independent and easy to test.

Where does the state actually live?

“Stateless” never means state disappears — it means you externalize it to a purpose‑built store:

StateExternalized toExample
Session / identitySigned client tokenJWT or opaque cookie the client resends each call
Hot shared dataDistributed cacheRedis / Memcached for carts, counters, rate buckets
Durable recordsDatabasePostgres, DynamoDB — the system of record
Large blobsObject storeS3 / GCS for uploads, exports, model artifacts
In‑flight workMessage queueKafka / SQS holds jobs so a crash doesn't lose them
Key idea

Push state out of the compute tier. The app tier becomes a pool of interchangeable, disposable workers; scaling and healing happen at the load‑balancer level, not inside your code.

Sticky sessions are an anti‑pattern

Sticky sessions pin a client to one instance so its in‑memory session keeps working. It looks convenient and quietly re‑introduces every problem statelessness solved.

✅ Stateless + externalized

  • Any instance serves any request.
  • Instance death = a reroute, no data loss.
  • Even load distribution across the fleet.
  • Autoscale and deploy freely.

🚫 Stateful sticky sessions

  • Instance death = every pinned user logged out.
  • Hot instances while others sit idle.
  • Scale‑in drops live sessions; deploys are risky.
  • “Works on my node” debugging nightmares.
Clients token / JWT Load Balancer round-robin Instance A stateless Instance B stateless Instance N stateless Redis shared cache Database system of record
Diagram 1 · Identical stateless instances behind a load balancer, all reading shared state from Redis and the database. Any instance can serve any request.
Say this: “I keep the app tier stateless and externalize session to a signed token and hot data to Redis, so I can autoscale and lose an instance without dropping a single user.”

Caching

Caching trades a little staleness and memory for a large drop in latency and load. The skill is choosing the right pattern, the right eviction/TTL, and a plan for the two hard parts: invalidation and stampedes.

Caching patterns — when to use which

PatternHow it worksBest forWatch out
Cache‑aside (lazy)App checks cache; on miss loads DB and populates cache.Read‑heavy, general default.First hit is slow; risk of stampede on hot keys.
Read‑throughCache library loads from DB on miss transparently.Uniform read path, less app glue.Cache becomes a dependency of every read.
Write‑throughWrite goes to cache and DB synchronously.Read‑after‑write consistency needed.Adds write latency; caches data you may never read.
Write‑behind (write‑back)Write to cache, flush to DB asynchronously.Write‑heavy, absorb bursts.Data loss window if cache dies before flush.
Refresh‑aheadProactively refresh popular keys before they expire.Predictable hot keys, latency‑critical.Wasted work if prediction is wrong.

TTL & eviction

LRU

Least Recently Used

Evict the entry untouched longest. Great default for temporal locality.

LFU

Least Frequently Used

Evict the least‑hit entry. Better when popularity, not recency, matters.

FIFO

First In First Out

Evict the oldest inserted regardless of use. Simple; ignores access pattern.

TTL (time‑to‑live) bounds staleness independent of memory pressure; eviction bounds memory independent of age. Use both: a TTL to expire, a policy to reclaim under pressure.

Application Cache Database READ (miss path) 1 · GET key → miss 2 · SELECT from DB 3 · SET key, TTL 4 · return value WRITE · UPDATE DB then DEL/INVALIDATE key
Diagram 2 · Cache‑aside: read populates on miss; write updates the DB and invalidates the key so the next read reloads fresh.

Invalidation — the hardest problem

“There are only two hard things in computer science: cache invalidation and naming things.” Strategies, roughly increasing in freshness and cost:

  • TTL expiry — simplest; accept up to TTL seconds of staleness.
  • Write‑invalidate — delete/overwrite the key on every write (shown above).
  • Versioned keys — bake a version or hash into the key (user:42:v7); bump on change, old key ages out.
  • Event‑driven — publish change events; subscribers purge affected keys (and CDN edges).

Cache stampede / thundering herd

A hot key expires and thousands of concurrent misses hammer the database at once. Mitigations, stackable:

🔗

Request coalescing (single‑flight)

Let one request recompute while the rest wait for its result — collapse N misses into 1 DB read.

🔒

Per‑key locks

A short mutex/lease per key so only one worker rebuilds it; others serve stale or block briefly.

🎲

Jittered TTL

Add randomness to expiry so keys don't all die on the same second and synchronize a herd.

⏱️

Early / probabilistic recompute

Refresh a key before it expires with rising probability as TTL approaches (XFetch).

🚫

Negative caching

Cache “not found” briefly so a flood of misses for a nonexistent key can't stampede the DB.

📦

Serve stale on rebuild

Return the last good value while a background task recomputes — availability over freshness.

Stampede in one line

Never let a popular key's expiry translate 1:1 into database load. Coalesce misses, jitter TTLs, and be willing to serve slightly stale data during a rebuild.

Layered caching

Each layer absorbs load so the next one sees far less. The closer to the user the hit, the cheaper and faster it is.

Browser HTTP cache ~0 ms CDN edge geo POPs ~5–20 ms App / API cache Redis ~1 ms DB buffer pool in‑memory pages ~5–50 ms Disk storage Miss cascades right → each layer shields the next Closer to the user = cheaper + faster. Aim to resolve most reads before they ever reach the database.
Diagram 3 · Layered caches: browser → CDN → application cache → DB buffer pool → disk. A miss cascades right, but most reads are answered early.

Hit‑ratio latency math

Average latency is a weighted blend of hits and misses. With a 95% hit ratio, hits at 1 ms and misses at 50 ms:

# avg = hit_ratio·hit_latency + miss_ratio·miss_latency
avg = 0.95 × 1ms + 0.05 × 50ms
    = 0.95ms + 2.5ms
    = 3.45 ms
3.45 ms
avg @ 95% hits
50 ms
uncached baseline
~14×
effective speedup
5%
of traffic hits DB
The tail dominates

Misses are what you feel. Pushing 95%→99% hits here drops avg from 3.45 ms to ~1.5 ms and cuts DB load 5×. The last few percent of hit ratio buys the most.

Databases & replication

The cache makes reads fast, but the database is the source of truth — the durable state everything else is derived from. Two questions dominate every data‑tier design: how does it survive a node dying (replication), and how does it hold more than one machine can (partitioning)?

Primary & replicas (leader–follower)

The default topology: one primary (a.k.a. leader) takes all writes; one or more replicas (followers) copy its changes and serve reads. Because most workloads are read‑heavy, fanning reads out to replicas is the cheapest way to add capacity — and if the primary dies, a replica can be promoted to take its place.

App fleet read/write split Primary (leader) all writes Replica · reads Replica · reads writes reads replicate (async)
Leader–follower: the primary takes every write and streams its change log to the replicas; reads fan out across the replicas.
📤

Writes → primary

Every insert/update/delete goes to the single primary, which orders them and streams its change log onward. One writer keeps the data consistent and avoids write‑write conflicts.

📥

Reads → replicas

Point read‑only queries at the replicas to scale reads horizontally and keep the primary free for writes. Read load grows? Add replicas.

Synchronous vs asynchronous replication

ModeHow it worksOn primary failureCost
SynchronousPrimary waits for the replica to confirm before acking the write.No data loss — the standby is fully up to date.Higher write latency; a slow replica stalls writes.
AsynchronousPrimary acks immediately; replicas catch up a moment later.May lose the last few seconds of writes.Fast writes, but replicas lag behind.

A common production setup blends both: one synchronous standby for zero‑data‑loss failover, plus several asynchronous read replicas for scaling reads.

Replica lag & “read‑your‑writes”

Async replicas trail the primary by milliseconds–seconds. A user who saves a change and immediately reads from a lagging replica sees the old value (“I just saved it — where did it go?”). Fixes: route that user’s reads to the primary for a short window after a write, read from the primary on critical paths, or use a store that offers read‑your‑writes session consistency.

Partitioning & sharding — scaling writes

Replicas scale reads, not writes: every replica still stores the whole dataset and replays every write. When a single primary can’t keep up with write throughput — or the data won’t fit on one machine — you shard: split the rows across N independent primaries, each owning a slice. The shard key decides which slice a row lands in, so pick one that spreads load evenly.

StrategyHowWatch out for
Hashshard = hash(key) % N.Even spread, but range scans hit every shard; naive % N reshuffles everything when you add shards — use consistent hashing.
RangeContiguous key ranges per shard (A–F, G–M…).Range queries stay on one shard, but sequential keys (timestamps, auto‑increment ids) create hot shards.
DirectoryA lookup table maps key → shard.Most flexible and easy to rebalance, but the lookup service is a new dependency and potential single point of failure.
Sharding is a one‑way door — delay it

Sharding breaks cross‑shard joins and transactions (they become slow scatter‑gather or application‑side work), and resharding later is painful. Exhaust the cheaper options first: read replicas, caching, a bigger box (vertical scaling), and archiving cold data. Also beware the hot shard (a “celebrity” key that draws disproportionate traffic) — it undoes the whole benefit, so choose a high‑cardinality, evenly‑distributed shard key.

SQL vs NoSQL

Not a rivalry — two toolboxes with different defaults. Start relational unless a specific access pattern or scale requirement pushes you off it.

SQL (relational)

  • Rows & columns with a fixed schema; related data split across tables and recombined with joins.
  • ACID transactions — strong consistency by default.
  • Flexible ad‑hoc queries; you don’t have to know them up front.
  • Scales up first; sharding is bolt‑on effort.
  • Postgres, MySQL, SQL Server.

NoSQL (non‑relational)

  • Flexible / schema‑less; data stored the shape it’s read in (denormalized), few or no joins.
  • Often eventual consistency, frequently tunable per operation.
  • Query patterns must be known up front — you design tables around them.
  • Scales out horizontally by design.
  • Document, key‑value, wide‑column, graph.

(Both columns are “good” — the winner depends on your access pattern, not on one being better.)

Reach for…When
SQL (relational)Relationships & joins, multi‑row transactions (money, orders, inventory), reporting/ad‑hoc queries, moderate scale. The safe default.
Document (MongoDB)Self‑contained JSON documents, flexible/evolving schema, one aggregate fetched per read.
Key‑value (Redis, DynamoDB)Simple get/put by key at massive scale — sessions, carts, feature flags, caches.
Wide‑column (Cassandra)Huge write volume & time‑series/event data, known query patterns, linear horizontal scale.
Graph (Neo4j)Deeply connected data where the relationships are the query — social graphs, fraud, recommendations.

Good practices

🔌

Connection pooling

DB connections are expensive and capped. Reuse a bounded pool (e.g. PgBouncer) instead of opening one per request — otherwise a traffic spike exhausts connections and stalls every query.

🔎

Index the read paths

Add indexes on the columns you filter and sort by to turn full scans into seeks. But every index slows writes and costs storage — index deliberately, not everything.

🔁

Kill N+1 queries

Don’t loop issuing one query per row. Batch with a join or IN (…) — the N+1 pattern is the most common hidden latency bug on read paths.

🧬

Backward‑compatible migrations

Roll out schema changes in expand → migrate → contract steps so old and new code both work during a deploy. Never drop/rename a column the running version still uses.

💾

Backups you’ve restored

Automated backups + point‑in‑time recovery — and actually test the restore. An untested backup is a hope, not a recovery plan.

🔒

Transactions at boundaries

Wrap multi‑row invariants (debit + credit, order + inventory) in a transaction so they commit all‑or‑nothing. Keep them short to avoid lock contention.

Say this: “I default to a relational primary with read replicas and a cache in front. That covers most read‑heavy workloads; I reach for NoSQL only when a specific access pattern needs it, and I shard only once a single primary’s write throughput is genuinely the bottleneck.”

Load Balancing

A load balancer spreads traffic across the fleet and steers it away from unhealthy nodes. First decision: which OSI layer.

✅ L4 (transport)

  • Routes on IP + port; forwards TCP/UDP.
  • Cannot see URLs, headers or cookies.
  • Very fast, cheap, protocol‑agnostic.
  • Good for raw throughput and non‑HTTP.

✅ L7 (application)

  • Routes on path, host, headers, method, cookies.
  • Does TLS termination, path/header routing, rewrites.
  • Enables canary, A/B, per‑route pools.
  • Slightly more CPU, far more control.

(Both columns are “good” here — the point is capability, not a winner. Pick L7 when you need content‑aware routing; L4 when you need raw speed.)

Algorithms

AlgorithmPicks backend by…Use when
Round‑robinNext in rotation.Uniform requests & uniform backends.
Weighted RRRotation biased by capacity weight.Heterogeneous instance sizes.
Least‑connectionsFewest active connections.Long‑lived or variable‑length requests.
Least‑response‑time / EWMALowest smoothed latency.Backends with uneven or drifting latency.
Consistent hashingHash(key) mapped on a ring.Cache/shard affinity; minimize reshuffle on scaling.

Why consistent hashing? With plain hash(key) % N, changing N remaps almost every key. On a hash ring, adding or removing a node only moves the keys in that node's arc — about 1/N of them — so caches and shards stay mostly warm during scale events. Virtual nodes smooth out the distribution.

Health & lifecycle

🩺

Active health checks

LB probes /healthz on a schedule and ejects nodes that fail. Deterministic, proactive.

📉

Passive health checks

Infer health from live traffic — consecutive 5xx or timeouts eject a node. Reactive, zero extra probes.

🐢

Slow start

Ramp traffic to a freshly added node gradually so cold caches/JITs warm before full load.

🚰

Connection draining

On removal, stop new connections but let in‑flight requests finish — no abrupt cutoffs during deploys.

Global vs regional

Real systems layer them: global DNS/anycast/GeoDNS sends a user to the nearest healthy region; a regional L7 balancer then spreads within it.

Clients HTTPS L7 LB TLS · routes on path Backend 1 healthy ✓ Backend 2 healthy ✓ Backend 3 ejected ✕ Backend 4 slow-start ⟳ Consistent‑hash ring N1 N2 N3 N4 key key → next node clockwise; add/remove moves only ~1/N keys
Diagram 4 · L7 balancer over a health‑checked pool (ejected + slow‑start nodes shown), with a consistent‑hash ring for affinity that survives scaling.

Retries

Retries paper over transient blips — a dropped packet, a brief failover. They also amplify a struggling system into a full outage if done naively.

Only retry idempotent ops

Safe to repeat: GET, PUT, DELETE. A blind retry of a non‑idempotent POST can double‑charge — gate it with an idempotency key.

Only on transient failures

Retry timeouts, 503s, connection resets. Never retry a 400/422 — the request is wrong; retrying just wastes capacity.

💰

Retry budgets & caps

Cap attempts (e.g. 2–3) and enforce a fleet‑wide budget (e.g. retries ≤ 10% of requests) so retries can't dominate load.

🌀

Backoff + jitter

Space attempts with exponential backoff and randomized jitter so clients don't resynchronize into a wave.

✅ Good retry

  • Idempotent op or idempotency key attached.
  • Retry only transient (timeout, 503, reset).
  • Max 2–3 attempts, exponential backoff + jitter.
  • Respect Retry‑After; obey a global budget.

🚫 Bad retry

  • Retry a POST with no dedup → duplicates.
  • Retry a 400/422 → guaranteed to fail again.
  • Immediate, unbounded retries in a tight loop.
  • Every layer retries → multiplicative amplification.
Retry storm

When a backend slows, naive clients retry, multiplying load exactly when it's weakest — a self‑inflicted DDoS that turns a blip into an outage. Nested retries compound: 3 layers × 3 attempts = 27× amplification. Bound attempts, add jittered backoff, and pair retries with a circuit breaker. See exponential backoff & jitter in Chapter 3 · Reliability Engineering.

Backpressure & Load Shedding

You cannot process more than your slowest resource allows. Backpressure signals “slow down” upstream; load shedding drops work you can't serve so the work you do serve stays fast. Failing fast beats failing slow.

📥

Bounded queues

Every buffer has a hard limit. An unbounded queue just converts a throughput problem into an out‑of‑memory crash with terrible latency.

🎫

Credit / flow control

Consumers grant credits; producers send only what there's capacity for (TCP windows, gRPC/HTTP‑2 flow control, reactive streams).

✂️

Shed low‑priority work

Under pressure, drop or defer non‑critical requests (analytics, prefetch) to protect the critical path.

🚦

Reject fast: 429 / 503

When the queue is full, return 429/503 with Retry‑After immediately. A fast rejection is kinder than a slow timeout.

Little's Law — sizing the queue

The relationship between concurrency, arrival rate and latency:

// L = concurrent items in system,  λ = arrival rate,  W = time in system
L = λ × W

If requests arrive at λ = 500/s and each spends W = 0.2 s in the system, then L = 100 in flight. If your pool only handles 100 concurrently, a queue longer than that only adds latency without adding throughput — so bound the queue near L and shed the rest. Rearranged, W = L / λ: a deeper queue directly inflates response time.

Producer λ = 500 req/s Bounded buffer · capacity 100 Consumer μ = 400 req/s shed → 429 / 503 when λ > μ and buffer full, reject fast instead of queuing forever
Diagram 5 · A bounded buffer between producer and consumer. When arrivals outpace the consumer and the buffer fills, excess work is shed with a fast 429/503 rather than queued into a latency spiral.
Say this: “I'd bound the queue with Little's Law, shed low‑priority load with a fast 429 plus Retry‑After, and protect the critical path — a slow timeout is the worst possible failure mode.”

Idempotency

An operation is idempotent if performing it once or many times yields the same result and side effects. It's what makes retries and at‑least‑once delivery safe.

VerbIdempotent?Why
GETyesRead‑only, no side effects.
PUTyesReplaces the resource to a fixed state; repeats converge.
DELETEyesResource ends deleted regardless of repeat count.
PATCHdependsIdempotent if the patch is absolute (set x=5), not if relative (x += 1).
POSTnoCreates/append by default — needs an idempotency key to be safe.

Idempotency keys + dedup store

Make a non‑idempotent write safe: the client sends a unique Idempotency-Key; the server records it with the result and returns the stored response on any replay.

POST /v1/payments
Idempotency-Key: 4b1e9f6a-2c...   # client-generated, unique per intent

# server logic
if store.exists(key):
    return store.get(key).response   # replay → same result, no double charge
result = charge_card(...)
store.put(key, result, ttl=24h)      # first time → do work, remember it
At‑least‑once ⇒ idempotent consumers

Most queues (Kafka, SQS) deliver at‑least‑once — duplicates happen on redelivery. Rather than chase exactly‑once, design consumers to be idempotent (dedup by message id / natural key). See idempotency keys in depth in Chapter 2 · API Design.

Circuit Breakers

A circuit breaker stops calling a dependency that's clearly failing, so you fail fast instead of piling requests onto a sick service (and stacking up your own threads waiting on it).

🟢

Closed

Normal operation. Requests pass through; the breaker tracks the error rate over a rolling window.

🔴

Open

Error threshold tripped → reject immediately (fail fast / fall back) for a cool‑down period. No calls hit the sick dependency.

🟡

Half‑open

After the timeout, allow a few trial requests. Success → close; failure → open again.

CLOSED requests pass OPEN fail fast HALF‑OPEN trial requests error % > threshold (over min volume) open timeout elapsed trials succeed → close trial fails → re‑open
Diagram 6 · Circuit breaker state machine. Thresholds trip Closed→Open; a cool‑down opens a Half‑open trial that either closes on success or re‑opens on failure.

Tuning knobs: error‑rate threshold (e.g. >50%), a minimum request volume before tripping (don't trip on 1 of 1), the open timeout (cool‑down), and how many half‑open trials to admit.

Bulkheads pair with breakers

Isolate resources into separate pools (threads, connections, queues) per dependency — like a ship's watertight compartments. One flooded dependency can't drown the rest of the service, and a breaker keeps you from refilling it.

Graceful / Service Degradation

When you can't serve the full experience, serve a reduced one instead of an error. Design the fallbacks up front so degradation is a deliberate mode, not a crash.

🥫

Serve stale cache

Return the last known‑good value when the source is down. Stale but useful beats fresh but unavailable.

🔀

Fallbacks

A default recommendation set, a cached price, a secondary provider — a sensible answer when the primary path fails.

🚩

Drop non‑critical features

Feature‑flag off recommendations, avatars, related items under load. Keep checkout working.

📉

Reduce quality

Smaller/faster model, fewer results, lower‑res media, coarser data — trade fidelity for availability.

📄

Fail static

Fall back to a static/cached page or a maintenance response instead of a hard 500.

🎯

Protect the core

Prioritize the critical path (login, pay, read) and shed everything else first.

Degradation is a UX decision

Users forgive “recommendations are temporarily unavailable” far more than a blank error page. Decide in advance which features are droppable and wire the fallbacks so the product stays usable when a dependency is on fire.

Say this: “Under partial failure I degrade rather than error — serve stale from cache, flag off non‑critical features, and keep the critical path alive. Availability of the core beats perfection of the whole.”

Chapter recap

Cheat sheet — Distributed Systems

  • Stateless: externalize state (token, Redis, DB, object store); sticky sessions are an anti‑pattern.
  • Caching: cache‑aside is the default; pick TTL + LRU/LFU; the hard parts are invalidation and stampedes.
  • Stampede: single‑flight, per‑key lock, jittered TTL, early recompute, negative caching, serve stale.
  • Hit math: avg = h·hit + (1−h)·miss → 95% @1ms/50ms = 3.45 ms; chase the last few % of hits.
  • Databases: primary takes writes, replicas serve reads; sync = no data loss, async = fast but lags (read‑your‑writes).
  • Scale writes: shard on an even, high‑cardinality key; sharding breaks joins/transactions — delay it. SQL by default, NoSQL for a specific access pattern.
  • Load balancing: L4 = fast/IP‑port, L7 = content‑aware; consistent hashing moves only ~1/N keys on scale.
  • LB lifecycle: active + passive health checks, slow‑start, connection draining.
  • Retries: idempotent + transient only, capped, budgeted, backoff + jitter — beware the 27× retry storm.
  • Backpressure: bounded queues, shed low‑priority, reject fast with 429/503; size with L = λ·W.
  • Idempotency: GET/PUT/DELETE safe; POST needs an idempotency key; at‑least‑once ⇒ idempotent consumers.
  • Circuit breaker: Closed → Open → Half‑open on error %/volume; pair with bulkheads.
  • Degrade: serve stale, fall back, drop extras, reduce quality, fail static — protect the core.