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.
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:
| State | Externalized to | Example |
|---|---|---|
| Session / identity | Signed client token | JWT or opaque cookie the client resends each call |
| Hot shared data | Distributed cache | Redis / Memcached for carts, counters, rate buckets |
| Durable records | Database | Postgres, DynamoDB — the system of record |
| Large blobs | Object store | S3 / GCS for uploads, exports, model artifacts |
| In‑flight work | Message queue | Kafka / SQS holds jobs so a crash doesn't lose them |
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.
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
| Pattern | How it works | Best for | Watch 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‑through | Cache library loads from DB on miss transparently. | Uniform read path, less app glue. | Cache becomes a dependency of every read. |
| Write‑through | Write 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‑ahead | Proactively refresh popular keys before they expire. | Predictable hot keys, latency‑critical. | Wasted work if prediction is wrong. |
TTL & eviction
Least Recently Used
Evict the entry untouched longest. Great default for temporal locality.
Least Frequently Used
Evict the least‑hit entry. Better when popularity, not recency, matters.
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.
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
TTLseconds 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.
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.
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
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.
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
| Mode | How it works | On primary failure | Cost |
|---|---|---|---|
| Synchronous | Primary 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. |
| Asynchronous | Primary 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.
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.
| Strategy | How | Watch out for |
|---|---|---|
| Hash | shard = hash(key) % N. | Even spread, but range scans hit every shard; naive % N reshuffles everything when you add shards — use consistent hashing. |
| Range | Contiguous key ranges per shard (A–F, G–M…). | Range queries stay on one shard, but sequential keys (timestamps, auto‑increment ids) create hot shards. |
| Directory | A 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 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.
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
| Algorithm | Picks backend by… | Use when |
|---|---|---|
| Round‑robin | Next in rotation. | Uniform requests & uniform backends. |
| Weighted RR | Rotation biased by capacity weight. | Heterogeneous instance sizes. |
| Least‑connections | Fewest active connections. | Long‑lived or variable‑length requests. |
| Least‑response‑time / EWMA | Lowest smoothed latency. | Backends with uneven or drifting latency. |
| Consistent hashing | Hash(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.
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.
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.
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.
| Verb | Idempotent? | Why |
|---|---|---|
| GET | yes | Read‑only, no side effects. |
| PUT | yes | Replaces the resource to a fixed state; repeats converge. |
| DELETE | yes | Resource ends deleted regardless of repeat count. |
| PATCH | depends | Idempotent if the patch is absolute (set x=5), not if relative (x += 1). |
| POST | no | Creates/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
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.
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.
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.
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.
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/Nkeys 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 withL = λ·W. - Idempotency:
GET/PUT/DELETEsafe;POSTneeds 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.