Distributed Systems

The umbrella guide. This is the foundational layer under every pattern guide in this folder (Kafka/EDA, saga, CQRS, outbox, CDC, event sourcing, circuit-breaker/bulkhead, DDD, databases). Those cover what to build; this covers why the hard parts are hard and the vocabulary and trade-offs a Staff+ engineer is expected to reach for by reflex. Where a topic has its own guide, this one gives you the mental model and cross-references rather than repeating it.

The mental model (the one idea everything derives from)#

A distributed system is a set of independent computers that fail independently, communicate only by passing messages over an unreliable network, and share no clock. Therefore you can never observe the current global state — only stale, partial, local views of the past.

Every hard problem in the field is a direct consequence of three facts that you cannot engineer away, only cope with:

1. Partial failure. In a single process, it either works or it crashes. In a distributed system, some nodes are up while others are down, and — critically — a healthy node cannot distinguish a crashed peer from a slow peer from a partitioned network. A missing response tells you nothing about whether the work happened.
2. Asynchrony (no bound on delay). Messages can be arbitrarily delayed, reordered, duplicated, or dropped. There is no timeout short enough to be safe and long enough to be live. "It didn't reply in 2 seconds" ≠ "it's dead."
3. No global clock. Wall clocks on different machines disagree (skew) and jump (NTP corrections, leap seconds). You cannot use timestamps to reliably order events across machines.

From those three facts, everything follows: you cannot get consistency and availability during a partition (CAP); you cannot achieve guaranteed consensus in a purely asynchronous system with even one crash (FLP); you cannot get "exactly-once" delivery (only at-least-once + idempotency); you cannot trust wall-clock ordering (hence logical clocks). The entire discipline is building a reliable whole out of unreliable parts by choosing your trade-offs deliberately, per operation, instead of pretending the network is a local function call.

The single most senior instinct that flows from this: treat every remote interaction as a fallible message exchange with a timeout, a retry policy, an idempotency story, and a failure plan — never as a method call that "just works." The junior mistake is the opposite: modelling orderService.charge(card) as if the network were free, instant, reliable, and ordered. It is none of those.

The 8 Fallacies of Distributed Computing (Deutsch & Gosling, Sun, 1990s) are the canonical list of the lies we tell ourselves: (1) the network is reliable, (2) latency is zero, (3) bandwidth is infinite, (4) the network is secure, (5) topology doesn't change, (6) there is one administrator, (7) transport cost is zero, (8) the network is homogeneous. Every one of them is false, and every distributed bug is ultimately someone having believed one of them. Memorize these; naming a fallacy by number in an interview is a cheap senior signal.

1. The foundational reality: partial failure, asynchrony, failure models#

Why "just a network call" is the master trap#

A local method call has properties you get for free and stop noticing: it always executes, it executes exactly once, it's fast and cost-free, arguments don't get corrupted, and either it returns or the whole process dies together. A remote call has none of these. It can be lost on the way out, executed and then have its response lost on the way back (so the caller retries and it runs twice), delayed past your timeout (so you give up while it's still running), or partially applied. The famous distillation (Waldo et al., "A Note on Distributed Computing," 1994) is that trying to make remote calls look local — classic RPC transparency — is a category error that leaks at exactly the worst moments.

The Two Generals Problem#

Two armies must attack at the same time to win; they coordinate by messengers who can be captured crossing enemy territory. There is no protocol — no finite number of acknowledgements — that guarantees both armies know the other will attack. Every message needs an ack, and that ack needs an ack, forever. This is the bedrock impossibility result for reliable communication over an unreliable channel, and it's why TCP (which sits on this same reality) gives you reliable-enough delivery with retransmission, not a mathematical guarantee. Practical takeaway: you can never be certain a message was received; you can only make the probability of loss acceptably low and design so that duplicates and losses are survivable.

Failure models — know the hierarchy#

You design against a failure model: the set of ways you assume components can misbehave. From easiest to hardest to tolerate:

The interview-relevant line: almost all commercial distributed systems (databases, queues, microservices) assume crash-recovery with omission, not Byzantine faults. Byzantine fault tolerance (BFT) is the domain of blockchains and aerospace/defense — bringing it up unprompted for a normal backend design signals you don't know where the cost lives. But knowing the term and that "3f+1 to tolerate f" is the price shows range.

The "aha": failure detection is fundamentally imperfect#

Because of asynchrony, a perfect failure detector is impossible — you cannot reliably tell "dead" from "slow." Everything practical is a timeout-based (unreliable) failure detector: you wait some bound, and if no heartbeat arrives, you declare the node dead. This is a decision, not an observation, and it can be wrong in both directions — which is the root cause of split-brain (you declared a live leader dead and elected a second one) and of premature failover. The sophisticated version is the phi-accrual failure detector (Hayashibara et al., 2004; used by Cassandra and Akka), which outputs a continuous suspicion level from the recent distribution of heartbeat inter-arrival times instead of a hard up/down bit, letting each consumer pick its own threshold.

2. Time, clocks, and the ordering of events#

This section is the one most engineers are fuzzy on, and it's a reliable senior/staff differentiator.

Two kinds of clock, and never confuse them#

• Time-of-day (wall) clock — "what time is it": System.currentTimeMillis(), Instant.now(). Synced to NTP, so it can jump backwards, pause, or skip. Never use it to measure elapsed time or to order events. NTP over a WAN is typically accurate to only a few to tens of milliseconds, and a bad sync can be off by far more.
• Monotonic clock — "how much time has passed": System.nanoTime(). Only ever moves forward, unaffected by NTP. Use this for timeouts, latency measurement, backoff, and anything comparing durations. Its absolute value is meaningless; only differences matter.

Interview trap: measuring a timeout or a request duration with currentTimeMillis(). If NTP steps the clock during the measurement you get a negative or absurd duration — and in the wild this has caused everything from spurious timeouts to broken rate limiters. Use nanoTime() for durations.

Why you cannot order events by wall-clock timestamp#

Two events on two machines with clocks skewed by 50 ms: the one that "really" happened second can carry the earlier timestamp. If you resolve conflicts by "highest timestamp wins" (Last-Write-Wins), you will silently drop the write that a correct causal order should have kept. LWW is not a correctness strategy; it's a data-loss strategy you accept when the data is low-value (e.g., a cache entry). Cassandra's default LWW conflict resolution has bitten many teams for exactly this reason.

Logical clocks — order without a shared clock#

Since physical time fails, we order events by causality instead.

• Lamport timestamps (Leslie Lamport, "Time, Clocks, and the Ordering of Events in a Distributed System," 1978 — one of the most cited papers in CS). Each node keeps a counter; increment on each event; attach it to every message; on receive, set counter = max(local, received) + 1. This gives a total order consistent with causality: if A happened-before B (A could have influenced B), then L(A) < L(B). But the converse is false — L(A) < L(B) does not mean A influenced B; they might be concurrent. Lamport clocks give you a consistent tie-breakable order; they cannot tell you whether two events were causally related or merely concurrent.
• Vector clocks (Fidge and Mattern, independently, 1988) fix that. Each node keeps a vector of counters, one per node. Now you can detect concurrency: compare vectors element-wise — if neither dominates the other, the events are concurrent (a genuine conflict that needs resolving). This is how Dynamo-style stores detect conflicting writes instead of silently losing one. Cost: the vector grows with the number of writers.
• Hybrid Logical Clocks (HLC) combine a physical timestamp with a logical counter, so timestamps stay close to wall-clock time (useful for humans and TTLs) while preserving causal ordering. Used by CockroachDB and others.

The physical-clock end of the spectrum: Google Spanner & TrueTime#

Spanner (Google, OSDI 2012) is the famous counter-move: instead of giving up on physical time, engineer the uncertainty down and expose it. TrueTime uses GPS receivers and atomic clocks in every datacenter to return an interval [earliest, latest] with a bounded error (single-digit milliseconds). To guarantee external consistency (linearizability) for globally-distributed transactions, Spanner does commit-wait: it waits out the uncertainty interval before releasing a commit, so no other transaction can observe a timestamp inversion. The lesson worth quoting: you can buy stronger consistency with better clocks, but you pay for it in latency (the commit-wait) and in exotic hardware.

How to sound senior here: "I never order cross-node events by wall-clock time — I use logical/version clocks for causality and monotonic clocks for durations. If I truly need globally-consistent timestamps I'm in Spanner/TrueTime territory, and I'm paying commit-wait latency for it."

3. Consistency models — the spectrum you actually choose from#

"Consistency" is overloaded. In ACID it means "the DB moves between valid states" (integrity constraints). In CAP / distributed replication it means "what ordering and recency guarantees does a read see across replicas." This section is the second meaning. (Your databases-and-data-storage-study-guide.md covers ACID isolation levels and their anomalies; here we focus on the replication/distributed consistency spectrum.)

Think of it as a ladder from strongest (most intuitive, most expensive) to weakest (cheapest, most surprising):

† ZooKeeper writes are linearizable, but reads are sequentially consistent and may be stale unless you issue a sync() first — a classic interview gotcha. etcd, by contrast, serves linearizable reads by default.

The client-centric "session" guarantees (know these by name)#

Eventual consistency is often too weak to build on, so systems add session guarantees that make the weirdness tolerable without paying for linearizability:

• Read-your-own-writes (read-after-write): you always see your own updates. (Classic failure: user edits their profile, gets routed to a lagging replica, sees the old value, thinks it didn't save.)
• Monotonic reads: you never see time go backwards — once you've read a value, you won't later read an older one. (Failure: refreshing a comment thread and watching comments disappear and reappear as you hit different replicas.)
• Monotonic writes: your writes are applied in the order you issued them.
• Consistent prefix reads: you see writes in an order that respects causality — never the answer before the question.

Staff-level framing: consistency is a per-operation choice, not a per-system one. The "add to cart" path can be eventually consistent; the "charge the card / decrement inventory to zero" path needs linearizability or a compensating saga. A senior answer assigns a consistency requirement to each operation and picks storage/routing accordingly, rather than declaring "the system is strongly consistent." Naming the exact model ("I need read-your-writes here, full linearizability there") is the differentiator.

4. CAP, PACELC, and the trade-off you are actually making#

CAP, stated precisely (and the way most people get it wrong)#

Brewer's conjecture (2000 keynote) proved by Gilbert & Lynch (2002). Define the three terms precisely, because the loose version ("pick 2 of 3") is wrong:

• C (Consistency) here specifically means linearizability — every read sees the latest write.
• A (Availability) means every request to a non-failing node returns a (non-error) response — eventually, in bounded time.
• P (Partition tolerance) means the system keeps operating when the network drops/delays messages between nodes.

The theorem: when a partition occurs, you must give up either C or A. That's it. The common phrasing "pick 2 of 3" is misleading for two reasons: (1) P is not optional — networks will partition, so a real distributed system must tolerate partitions; the actual choice is only C vs A, and only during a partition. A "CA system" is a single-node system (or one that simply stops being distributed under partition). (2) When there is no partition, you get both C and A — CAP says nothing about the normal case.

So the honest statement is: "When the network partitions, does this data path stay available and risk stale/conflicting data (AP), or refuse to serve to preserve correctness (CP)?" Examples: a bank ledger balance → CP (refuse rather than double-spend). A social feed / shopping cart / product catalog → AP (serve stale, reconcile later). ZooKeeper/etcd → CP by design (they'd rather stop than diverge). Cassandra/Dynamo → AP by design (tunable toward CP with quorums).

PACELC — the extension that matters more day-to-day#

CAP only describes the partition case, which is rare. PACELC (Daniel Abadi, 2010/2012) completes it:

If Partition (P) → choose Availability or Consistency (A/C); Else (E, normal operation) → choose Latency or Consistency (L/C).

The "else" branch is the one you live in 99.9% of the time: even with a healthy network, stronger consistency costs latency, because a linearizable write/read must coordinate across replicas (wait for a quorum, or round-trip to the leader) before returning. This is why PACELC is the more useful lens for real design:

How to sound senior: don't say "it's a CP system." Say "for this data it's PC/EC — I'm paying cross-AZ latency on every write to keep it linearizable, and I've decided that's worth it because it's money; for the catalog data it's PA/EL and I'll serve stale under partition." That sentence is a Staff-level tell.

5. Replication — copies of data, and the trade-offs between them#

Replication = keeping copies of the same data on multiple nodes. You do it for four reasons, often at once: availability (survive node loss), read scaling (serve reads from many replicas), latency (put a copy near the user), durability (survive disk loss). The whole difficulty is keeping the copies in agreement while writes happen.

The three replication architectures#

1. Single-leader (primary/replica, master/slave). All writes go to one leader; it streams a replication log to followers; reads can be served by any replica. Simple, no write conflicts, linearizable if you read from the leader. This is the default for relational databases (Postgres, MySQL). Weaknesses: the leader is a write bottleneck and a failover risk. Failover is the dangerous part — detecting the leader is dead (imperfect failure detection again), promoting a follower, and avoiding two leaders (split-brain).

2. Multi-leader. Multiple nodes accept writes (e.g., one leader per region). Better write availability and local-write latency, but now you have write conflicts (two regions edit the same row concurrently) and need conflict resolution. Used for multi-datacenter and offline-capable apps. Conflict resolution is the hard part (see below).

3. Leaderless (Dynamo-style). Any replica takes writes; the client (or a coordinator) writes to several and reads from several. Cassandra, DynamoDB, Riak. Consistency is tuned via quorums.

Synchronous vs asynchronous replication — the durability/latency knife-edge#

• Synchronous: the leader waits for the follower(s) to acknowledge before confirming the write. No data loss on leader failure, but every write pays the slowest follower's latency, and if the follower is down, writes block.
• Asynchronous: the leader confirms immediately and ships the log in the background. Fast and available, but if the leader dies before shipping, those acknowledged writes are lost — the classic "we told the user it saved, then lost it" failure.
• Semi-synchronous (the usual compromise): one synchronous follower (guaranteed durable copy) + the rest async. Bounded data-loss risk without the full latency hit.

Quorums — the leaderless consistency dial#

With N replicas, require W nodes to ack a write and R nodes to respond to a read. The key inequality:

R + W > N ⟹ the read set and the write set overlap by at least one node ⟹ a read is guaranteed to see the latest acknowledged write.

Classic setting: N=3, W=2, R=2 (overlap of 1, tolerates one node down for both reads and writes). Tuning: W=N, R=1 makes reads fast and writes fragile; W=1, R=N the reverse. W + R ≤ N gives you higher availability and lower latency but no overlap guarantee → you can read stale. Note the subtleties that trip people up: quorums don't make the write atomic across replicas (a write can reach some and not others and still be "successful"), and concurrent writes to the same key still need conflict resolution — the quorum is about recency overlap, not conflict-freedom.

Supporting machinery in leaderless stores: read repair (fix stale replicas noticed during a read), anti-entropy (background sync, often via Merkle trees to find diffs cheaply), hinted handoff (a live node temporarily holds writes for a down node and forwards them on recovery).

Conflict resolution (when concurrent writes collide)#

• Last-Write-Wins (LWW): pick the write with the highest timestamp. Simple, and lossy — discards a concurrent write, and depends on unreliable clocks. Acceptable only for low-value data.
• Version vectors / vector clocks: detect that two writes were concurrent and surface both (siblings) for the application to merge. Correct but pushes work to the app.
• CRDTs (Conflict-free Replicated Data Types) (Shapiro et al., 2011): data structures (counters, sets, maps) with merge functions that are commutative, associative, and idempotent, so replicas always converge without coordination. The principled answer for collaborative/offline data (used by Redis, Riak, and collaborative editors). Great when your data fits a CRDT shape; not everything does.

Replication lag is the everyday consequence: the read-after-write problem (route the user's own reads to the leader or a caught-up replica), monotonic reads (pin a user to one replica), and replication-lag monitoring as an SLI. (Your databases guide has the storage-side detail; the distributed-systems point is that lag is not a bug to eliminate but a property to design around.)

6. Partitioning (sharding) — splitting data across nodes#

Replication copies the same data; partitioning splits different data across nodes so you can scale writes and storage beyond one machine. Real systems combine both: each partition is replicated. (Your databases guide covers sharding mechanics; here's the distributed-systems framing and the traps.)

How to assign data to partitions#

• Range partitioning: contiguous key ranges per node (e.g., users A–F on node 1). Great for range scans; prone to hot spots if the key is sequential (all "today's" timestamped writes hit one node — the classic time-series/auto-increment hotspot).
• Hash partitioning: hash the key, assign by hash. Spreads load evenly; kills range queries (adjacent keys scatter). The common default.
• Consistent hashing (Karger et al., 1997): map both keys and nodes onto a hash ring; a key belongs to the next node clockwise. Adding/removing a node only remaps the keys in one segment (~1/N of data), not everything — this is why it underpins Dynamo, Cassandra, and CDNs. Virtual nodes (each physical node owns many small ring segments) smooth out load imbalance and make rebalancing granular.

The problems that actually come up#

• Hot partitions / the "celebrity problem": one key (a viral tweet, a mega-tenant) gets disproportionate traffic and hashing can't help because it's a single key. Fixes: add a random salt/suffix to split the key across partitions (and scatter-gather on read), cache the hot key hard, or give the whale its own dedicated capacity.
• Rebalancing: moving partitions as nodes join/leave. Do it incrementally to avoid a rebalancing storm that saturates the network and cascades into failure. Avoid hash-mod-N schemes (hash % N) where changing N remaps almost everything — this is exactly the failure consistent hashing prevents.
• Request routing (service discovery for data): how does a client find the node holding key K? Three approaches — client-side awareness (client knows the partition map), a routing tier/proxy, or a coordination service (ZooKeeper/etcd) holding the map. This is a consensus/coordination problem, which is the next section.
• Secondary indexes: partitioning by primary key is easy; querying by a non-key attribute is not. Local (document-partitioned) indexes keep the index with its data → writes are cheap, but reads must scatter-gather across all partitions. Global (term-partitioned) indexes partition the index by the indexed term → reads hit one partition, but writes must update a remote index partition (and now you have a distributed-write consistency problem). This local-vs-global trade-off is a great thing to name in a design round.

7. Consensus and coordination — agreement despite failure#

This is the theoretical heart of the field and a Staff+ favorite. Consensus = getting a set of nodes to agree on a single value (or an ordered log of values) even though nodes crash and messages are delayed. Almost every "hard" coordination task reduces to consensus: electing a leader, agreeing on cluster membership, committing a distributed transaction, ordering a replicated log, implementing a distributed lock.

What "solving consensus" formally requires#

A correct consensus protocol guarantees: Agreement (no two correct nodes decide differently), Integrity/Validity (the decided value was actually proposed), and Termination (every correct node eventually decides). Agreement is easy alone (decide nothing); termination is easy alone (decide immediately); getting both under failure is the hard part.

FLP impossibility — the result every Staff+ candidate should be able to state#

Fischer, Lynch, and Paterson, 1985: in a purely asynchronous system, no deterministic protocol can guarantee consensus if even a single node may crash. The intuition: because you can't distinguish a crashed node from a slow one (asynchrony!), any protocol can be forced to wait forever for a message that may or may not come, so it can't guarantee termination.

This sounds fatal but isn't, because real systems escape the assumptions in one of three ways: (1) partial synchrony — assume the network is eventually well-behaved (bounded delay most of the time), which is enough to make progress when things are calm and merely stall (not violate safety) when they're not. This is what Paxos and Raft do. (2) Randomization — use coin-flips so no adversarial schedule can stall you forever (Ben-Or; used in BFT). (3) Failure detectors — assume an oracle (in practice, timeouts) that eventually identifies crashed nodes. The senior takeaway: FLP means you can never have safety and guaranteed liveness in an async system — production consensus keeps safety always and sacrifices liveness during bad network periods (it stalls rather than diverges).

Paxos and Raft#

• Paxos (Lamport, "The Part-Time Parliament," published 1998; "Paxos Made Simple," 2001). One of the first proven-correct fault-tolerant consensus protocols; roles of proposers/acceptors/learners; a two-phase (prepare/promise, accept/accepted) majority-quorum protocol. Famous for being correct and notoriously hard to understand and implement correctly. Multi-Paxos extends it to a log.
• Raft (Ongaro & Ousterhout, "In Search of an Understandable Consensus Algorithm," 2014) was designed for understandability and won practice. It decomposes consensus into three sub-problems: leader election (elect one leader per term using randomized election timeouts to avoid split votes), log replication (leader appends entries, replicates to followers, commits once a majority has them), and safety (only up-to-date nodes can be elected, so committed entries survive). It's the algorithm behind etcd, Consul, TiKV, and CockroachDB's replication.

Both need a majority quorum (⌊N/2⌋+1) to make progress, which is why consensus clusters are odd-sized (3 or 5). 3 nodes tolerate 1 failure, 5 tolerate 2. Even sizes waste a node (4 also only tolerates 1). This "why 3 or 5, never 4" question is a common quick screen.

The most important practical rule: don't build your own#

Consensus is a landmine — easy to get subtly wrong (lose committed data, admit two leaders). The Staff+ answer is almost never "I'll implement Raft"; it's "I'll delegate coordination to a battle-tested consensus system — ZooKeeper (ZAB protocol), etcd, or Consul — and use it for leader election, distributed locks, config, and membership." Managed equivalents: anything built on these.

Distributed locks need fencing tokens (a classic trap)#

If you use a lock service for mutual exclusion, a naive lock is unsafe: process A acquires the lock, then GC-pauses for 30s; the lease expires; the service grants the lock to B; A wakes up believing it still holds the lock and writes → two writers. The fix (Martin Kleppmann's well-known example) is a fencing token: the lock service hands out a monotonically increasing number with each grant, and the protected resource rejects any write carrying a token lower than the highest it has seen. A's stale token gets rejected. Naming "fencing token" when you propose a distributed lock is a strong senior signal — it shows you know that "I'll just use a distributed lock" is not, by itself, safe.

8. Distributed transactions and cross-service consistency#

You want to change data in two places (two services, two databases, or a database and a message broker) and have both succeed or both fail. In a single database this is a local ACID transaction. Across a boundary, it's one of the hardest problems in the field, and the modern answer is to avoid the distributed transaction, not to perfect it.

Two-Phase Commit (2PC) — and why you avoid it across services#

2PC uses a coordinator: Phase 1 (prepare) — coordinator asks every participant "can you commit?"; each does the work, durably prepares, and votes yes/no while holding locks. Phase 2 (commit/abort) — if all voted yes, coordinator tells everyone to commit; otherwise abort. It gives atomicity, but the cost is brutal for services:

• It's blocking. Between voting "yes" and hearing the decision, a participant holds locks and cannot proceed. If the coordinator crashes at that moment, participants are stuck holding locks indefinitely ("in-doubt" transactions) — a single point of failure that freezes resources across services. (3PC tries to fix the blocking but adds latency and still fails under partitions; it's not used in practice.)
• It couples availability. The transaction only commits if every participant and the coordinator are up. Availability multiplies down: five 99.9% services in a 2PC ≈ 99.5%. It scales terribly and enshrines temporal coupling.

2PC is fine inside one database or via XA across a couple of resources you control with low concurrency. Across microservices it's an anti-pattern. The line to say: "I don't do 2PC across services — it's blocking on coordinator failure and couples availability. I use a saga with compensations and make each step idempotent."

The alternatives you actually reach for (all have their own guides here)#

• Saga pattern — model the cross-service change as a sequence of local transactions, each publishing an event that triggers the next; on failure, run compensating transactions to semantically undo the prior steps. Trades atomicity/isolation for availability; you must handle intermediate visibility. → saga-pattern-study-guide.md.
• Outbox pattern — solve the dual-write problem (below) by writing the business row and an "outbox" row in the same local transaction, then relaying the outbox to the broker. → inbox-outbox-pattern.md.
• Change Data Capture (CDC) — tail the database's replication log (Debezium) to publish changes reliably, no dual write. → cdc-change-data-capture-study-guide.md.
• Event sourcing / CQRS — make the event log the source of truth, derive read models from it. → event-sourcing-study-guide.md, CQRS-comprehensive-guide.md.

The dual-write problem (the single most common distributed-data bug)#

You need to update the database and publish a message ("order created"). If you do them as two separate operations, there is no ordering that is safe: DB commits then the app crashes before publishing → the rest of the system never hears about the order. Or publish first then the DB write fails → you announced an order that doesn't exist. There is no retry or try/catch that closes this gap, because the process can die between the two. The only correct solutions funnel both facts through one atomic write: the outbox (one DB transaction, relay later) or CDC (the DB log is the event). Recognizing a proposed design is a dual write ("you're writing to Postgres and then calling kafkaTemplate.send() — that's a dual write, it'll drop events on crash") is one of the highest-value pattern-matches in a design interview.

"Exactly-once" is a myth — internalize this#

You cannot have exactly-once delivery over an unreliable network (Two Generals). What you can have is at-least-once delivery + idempotent processing = "effectively-once." Even Kafka's "exactly-once semantics" (idempotent producer + transactions, since 0.11) is exactly-once within Kafka's own boundary; the moment an effect leaves that boundary (you charge a card, send an email, call another service) you are back to at-least-once and you must make the effect idempotent yourself. Anyone who says "I'll use exactly-once delivery" as a design primitive is signalling a gap. Which brings us to the most important practical skill in the whole guide:

9. Delivery semantics and idempotency — the everyday core skill#

The three delivery semantics#

• At-most-once: send and forget; no retries. Zero duplicates, but messages can be lost. Fine for high-volume, loss-tolerant telemetry (metrics samples).
• At-least-once: retry until acknowledged. No loss, but duplicates happen (the ack got lost, so you retry something that already succeeded). This is the practical default for anything that matters.
• Exactly-once: the unicorn. Approximated as at-least-once + idempotency (see above).

Because at-least-once is the default, idempotency is not optional — it's the tax you pay for reliability. An operation is idempotent if applying it twice has the same effect as applying it once. "Set status = SHIPPED" is naturally idempotent; "add $10 to balance" is not; "charge the card" is catastrophically not.

How to make operations idempotent (the toolbox)#

1. Idempotency key + dedup store. The client generates a unique key per logical operation (e.g., a UUID per "place order" click); the server records processed keys and, on a repeat, returns the original result instead of re-executing. This is exactly how Stripe's API works and how you should design any create/charge endpoint.
2. Natural idempotency. Design the operation to be a set (absolute) rather than a delta (relative): "set quantity to 5," not "add 3." Upserts by a business key rather than blind inserts.
3. Dedup on a consumer. Track processed message IDs in a table (the inbox pattern) and skip duplicates. → inbox-outbox-pattern.md.

Concrete: an idempotent HTTP endpoint (Idempotency-Key header)#

Kotlin

Java

The subtlety this simplified version glosses: to be correct under concurrency you insert the key first with a UNIQUE constraint (so two simultaneous retries race and exactly one wins), mark it IN_PROGRESS, and only the winner executes the charge — otherwise two near-simultaneous retries both miss the find() and both charge. In production you'd wrap the check-execute-store in one DB transaction with the unique-key insert as the guard.

Concrete: an idempotent Kafka consumer (inbox dedup)#

Kotlin

Java

Ordering note: Kafka only guarantees order within a partition, and only preserves it if you don't let retries reorder (max.in.flight.requests.per.connection=1 with retries, or use the idempotent producer). Order-sensitive data must share a partition key. → kafka-eda-study-guide.md, event-broker-vs-message-bus-study-guide.md.

10. Failure handling and resilience — building reliable from unreliable#

The single organizing principle: assume every dependency will be slow or down, and make sure that failure does not become your failure. The enemy is the cascading failure, where one slow dependency exhausts your threads/connections, which makes you slow, which topples your callers, until the whole system is down from one leaf. The patterns below are the standard toolkit; the circuit-breaker/bulkhead pair has its own guide (circuit-breaker-bulkhead.md) — here's how they fit together.

The core patterns (and the order to apply them)#

1. Timeouts on every remote call. A call with no timeout is a resource leak waiting to happen — one hung dependency and your thread pool fills with threads blocked forever. There is no such thing as a remote call without a timeout; the only question is the value. Interview trap: forgetting that the default timeout in many HTTP clients is infinite.
2. Retries — but only with backoff, jitter, and idempotency. Retrying a transient failure is good; retrying immediately and in lockstep creates a retry storm / thundering herd that DDoSes a recovering service right as it's trying to come back. Use exponential backoff (1s, 2s, 4s…) plus jitter (randomize the delay) so retries spread out instead of synchronizing. And never retry a non-idempotent operation without an idempotency key — you'll double-charge. Cap total attempts; give up to a fallback.
3. Circuit breaker. After a threshold of failures, "open" the circuit and fail fast (don't even try) for a cooldown, then "half-open" to test recovery. Stops you from hammering a dead dependency and lets it recover. → circuit-breaker-bulkhead.md.
4. Bulkhead. Isolate resources (separate thread pools / connection pools per dependency) so one saturated dependency can't consume all your capacity — like watertight compartments in a ship. → circuit-breaker-bulkhead.md.
5. Load shedding & backpressure. When overloaded, reject early (return 429/503) rather than accepting work you can't finish — a queue that grows without bound converts a latency problem into a total outage. Backpressure = telling upstream to slow down (bounded queues, reactive streams, consumer lag as the signal). Shedding some load keeps the system up for the rest; accepting all load kills it for everyone.
6. Graceful degradation / fallback. Serve a stale cache, a default, or a reduced feature instead of an error (recommendations service down → show generic bestsellers, not a 500).

Concrete: Resilience4j on a Spring Boot call#

Resilience4j is the standard modern library (Hystrix is EOL). The recommended nesting order of decorators is Bulkhead → TimeLimiter → CircuitBreaker → Retry (retry outermost so it retries the whole protected call).

Kotlin

Java

Set an explicit request timeout on the client too (Resilience4j TimeLimiter for async, or the HTTP client's own connect/read timeouts for sync) — the annotations above don't impose one by themselves on a blocking RestClient.

Two more failure phenomena worth naming#

• Cache stampede / thundering herd on expiry: a hot cache key expires and thousands of concurrent requests all miss and hit the origin at once, knocking it over. Fixes: jittered TTLs (don't expire everything at the same instant), request coalescing / single-flight (one request recomputes while others wait), and early/probabilistic recomputation (refresh just before expiry).
• Metastable failures: the system gets stuck in a bad state that sustains itself even after the trigger is gone — e.g., retries caused overload, and now the retries from the overload keep it overloaded. Recovery requires removing load (shedding, disabling retries), not just fixing the original trigger. This is a sophisticated failure class to name.

11. Observability — you cannot operate what you cannot see#

In a monolith a stack trace tells you what happened. In a distributed system a single user request fans out across a dozen services, and no one machine has the whole story. Observability is therefore not optional tooling; it's a design requirement. The distinction to draw: monitoring answers known questions ("is CPU high?"); observability lets you ask new questions about unknown failures after the fact.

The three pillars — and the one that matters most in distributed systems#

• Metrics — cheap numeric time-series (rates, counts, gauges). Aggregate, high-level. Frameworks: Micrometer → Prometheus. Use RED (Rate, Errors, Duration) per service and USE (Utilization, Saturation, Errors) per resource.
• Logs — discrete events. Structured (JSON with fields), not free text, so they're queryable. Must carry a correlation/trace id.
• Distributed tracing — the pillar unique to distributed systems. A trace follows one request across all services; each hop is a span (with timing and metadata). This is how you find which of twelve services added the 800 ms, and how you see the call graph. Context propagates via headers — the W3C Trace Context standard uses the traceparent header. Tools: OpenTelemetry (the vendor-neutral standard), Jaeger, Zipkin, Tempo.

Correlation IDs — the cheapest high-value thing you can add#

Generate/propagate a request id at the edge and thread it through every log line and outbound call. In Spring, Micrometer Tracing (successor to Spring Cloud Sleuth) does this automatically and puts trace/span ids in the logging MDC. Minimal manual version of the idea:

Kotlin

Java

Tail latency — the metric juniors miss#

Averages lie. In a fan-out request that calls 10 services in parallel and waits for all, the slowest one determines your latency — so p99 of the whole is driven by p99 of the parts, and tail latency amplifies with fan-out (Dean & Barroso, "The Tail at Scale," 2013). If each service is slow 1% of the time, a request touching 100 of them is slow ~63% of the time. Always reason in percentiles (p50/p95/p99/p99.9), never averages, and know the mitigations from that paper: hedged requests (send to two replicas, take the first answer, cancel the other) and tied requests. Talking in p99s and mentioning hedging is a Staff-level tell.

SLIs, SLOs, and error budgets#

• SLI — the measured indicator (e.g., % of requests < 200 ms).
• SLO — the target (e.g., 99.9% of requests < 200 ms over 30 days).
• Error budget — 1 − SLO (0.1% here). This is the permission to fail: as long as you're within budget you ship features fast; when you burn it, you freeze features and fix reliability. It turns "how reliable?" into an explicit, negotiated business decision rather than "as reliable as possible" (which is infinitely expensive). Liveness vs readiness probes are the operational cousins: liveness failing → restart me; readiness failing → stop sending me traffic but don't kill me (I'm warming up or a dependency is down).

12. Scalability and performance — and the estimation every interview wants#

Scale up vs scale out#

Vertical (scale up) = bigger machine. Simple, no distribution needed, but a hard ceiling and a single failure domain. Horizontal (scale out) = more machines. Effectively unlimited and fault-tolerant, but forces you into everything in this guide. The enabler for horizontal scaling is statelessness: if a service holds no per-client state in memory, any instance can serve any request and you scale by adding boxes behind a load balancer. Push state to the edges — databases, caches, the client (a token/cookie). Sticky sessions are a scaling smell.

Load balancing#

• L4 (transport): routes by IP/port, no payload awareness — fast, dumb.
• L7 (application): routes by HTTP path/header/cookie — enables path-based routing, canaries, and content-based decisions.
• Algorithms: round-robin, least-connections, consistent-hashing (sticky by key), power-of-two-choices (pick two at random, send to the less loaded — surprisingly effective and cheap).

Caching (the biggest single performance lever) — and its perils#

Covered in depth in databases-and-data-storage-study-guide.md; the distributed-systems essentials: cache-aside (app checks cache, on miss loads DB and populates) is the common pattern; the two hard problems are invalidation (stale data — pick a TTL and/or invalidate on write) and stampede (§10). Multi-layer caching (client → CDN → API gateway → app-local (Caffeine) → distributed (Redis) → DB) is normal. The senior caution: a cache is a consistency decision, not just a speed decision — you are choosing to serve possibly-stale data, so it belongs on the AP/EL side of your CAP/PACELC reasoning.

Back-of-the-envelope estimation — practice this out loud#

Interviewers want to see you quantify. Two tools:

Little's Law: concurrency = throughput × latency (L = λW). If you serve 2,000 req/s at 50 ms average, you have 2000 × 0.05 = 100 requests in flight → you need ~100 concurrent workers/threads. This one formula sizes thread pools, connection pools, and queue depths.

Latency numbers (orders of magnitude — memorize the ratios, not the digits):

The point isn't the exact figures (they shift with hardware); it's the ratios: memory is ~1000× faster than SSD, cross-region is ~300× a same-DC hop, and the speed of light sets a floor cross-region latency can never beat. This is why you cache, why you co-locate data with compute, and why "just add a cross-region synchronous replica" is not free.

Capacity math example (say this structure out loud): "100 M daily active users × 10 requests/day ≈ 1 B req/day ÷ 86,400 s ≈ ~11.6 K req/s average, and I'll plan for a ~5× peak → ~58 K req/s. Each app instance handles, say, 500 req/s → ~120 instances at peak plus headroom. Storage: 1 B events/day × 1 KB ≈ 1 TB/day → I need a retention/archival policy." Interviewers care far more about the method and the peak multiplier than the precise number.

13. How the pieces fit — the architecture layer (map to your other guides)#

The foundations above are what the applied patterns in this folder are built on. The umbrella view:

• Microservices decompose a system along bounded contexts (→ ddd-bounded-contexts-study-guide.md, domain-driven-design-invariants-and-core-concepts.md). The distributed-systems tax: every in-process call that becomes a network call inherits §1's failure modes — so don't split services finely without a reason. "Distributed monolith" (services so chatty and coupled they must deploy together) is the worst of both worlds.
• Event-driven architecture trades temporal coupling for eventual consistency: services communicate via events on a broker (→ kafka-eda-study-guide.md, event-broker-vs-message-bus-study-guide.md), which is how you get availability and loose coupling — at the cost of every consistency wrinkle in §3, §8, §9.
• CQRS / event sourcing separate write and read models and make the event log the source of truth (→ CQRS-comprehensive-guide.md, event-sourcing-study-guide.md, eda-vs-event-sourcing-study-guide.md).
• Saga / outbox / CDC are the concrete answers to §8's cross-service consistency problem (→ saga-pattern-study-guide.md, inbox-outbox-pattern.md, cdc-change-data-capture-study-guide.md).
• Resilience patterns (§10) keep the whole thing standing (→ circuit-breaker-bulkhead.md).

The most senior meta-point of all: distributed systems are a cost, not a goal. Every one of these patterns exists to cope with distribution. If a single well-provisioned Postgres box behind one stateless service meets your scale and availability needs, that is the correct architecture, and reaching for Kafka + microservices + sagas + eventual consistency to solve a problem you don't have is the most common Staff-interview red flag. Distribute when a real constraint (scale beyond one box, independent team/deploy boundaries, fault isolation, geo-latency) forces it — and then pay the tax deliberately.

Problems & how to solve them#

The failure catalog. In an interview, recognizing which of these a design is walking into — before it's pointed out — is the whole game.

Interview framing — how to sound senior#

The opening move on a system-design question. Don't jump to boxes. (1) Clarify functional requirements and the critical flows. (2) State non-functional requirements as numbers: expected scale (DAU, req/s, read:write ratio), latency targets (p99), consistency and durability needs, availability target. (3) Do the back-of-envelope math (§12) and note the peak multiplier. (4) Then design, and drive the conversation yourself.

The one habit that separates senior from junior: narrate every choice as a trade-off. Never "I'd use Cassandra." Always "I'd use Cassandra because the workload is write-heavy and append-only and I can tolerate eventual consistency on this path — at the cost of no ad-hoc queries and app-side conflict handling. If I needed strong consistency here I'd pay the latency for a single-leader store instead." Every technology sentence gets a because and an at the cost of.

Reflexes that read as Staff+ (drop these unprompted):

• Assign a consistency model per operation, not per system ("cart is eventually consistent; the inventory decrement is linearizable").
• Name the CAP/PACELC posture of each datastore ("this path is PC/EC, that one PA/EL").
• Say "idempotent", "at-least-once", "timeout + backoff + jitter", and "fencing token" at the moments they apply.
• Talk in p99, not averages; mention error budgets and blast radius.
• Call out a dual write the instant a design writes to a DB and a broker separately.
• Discuss rollout: canary, feature flags, and how you'd detect and roll back a bad deploy.
• Know when not to distribute — propose the single-box solution first if it fits, and justify each added piece.

Common wrong answers / traps to never fall into:

• "I'll use exactly-once delivery." (Doesn't exist end-to-end; say at-least-once + idempotency.)
• "CAP means pick 2 of 3." (P isn't optional; the choice is C-vs-A during a partition, and PACELC covers the normal case.)
• "I'll just use a distributed lock." (Unsafe without fencing tokens; and often a coordination smell.)
• Ordering events by wall-clock timestamp across machines. (Clock skew; use logical clocks.)
• Reaching for microservices / Kafka / sagas by default on a problem a single service and Postgres would solve.
• Measuring durations with currentTimeMillis() instead of nanoTime().
• Retrying a non-idempotent call, or retrying without jitter.
• Forgetting that 2PC across services blocks on coordinator failure.
• Ignoring the network entirely — treating a remote call like a local one.

When you don't know something, reason from the mental model out loud: "I don't remember that system's exact defaults, but it's leaderless, so it's AP/EL and I'd expect tunable quorums — I'd verify R + W > N before relying on read-your-writes." Reasoning from first principles beats a memorized but shaky fact.

Cheat-sheet#

The three unavoidable facts → partial failure · asynchrony (unbounded delay) · no global clock. Everything derives from these.

Consistency models (strong → weak): linearizable → sequential → causal → eventual. Session guarantees on top of eventual: read-your-writes, monotonic reads, monotonic writes, consistent prefix.

CAP: under a partition, choose C (linearizable) or A (every request answered). P is mandatory. PACELC: if Partition → A/C; Else → Latency/Consistency (the everyday case).

Quorum: R + W > N ⟹ reads see latest write. Default N=3, W=2, R=2. Consensus needs a majority ⟹ clusters are odd (3 or 5); a cluster of N tolerates ⌊(N−1)/2⌋ failures (3 → 1, 5 → 2).

Delivery semantics: at-most-once (lossy) · at-least-once (default → needs idempotency) · exactly-once (myth; = at-least-once + idempotent = effectively-once).

Replication: single-leader (no conflicts, failover risk) · multi-leader (write conflicts) · leaderless (quorum-tuned). Sync = durable+slow; async = fast+can-lose-data; semi-sync = the compromise.

Partitioning: range (scans, hot spots) vs hash (even, no scans) vs consistent hashing (only ~1/N remaps on resize) + virtual nodes.

Consensus: FLP → no guaranteed async consensus with 1 crash → keep safety, sacrifice liveness. Raft/Paxos, delegate to etcd/ZooKeeper/Consul. Locks need fencing tokens.

Resilience stack (inner→outer): Bulkhead → TimeLimiter → CircuitBreaker → Retry(backoff+jitter). Plus timeouts everywhere, load shedding, graceful degradation.

Cross-service consistency: avoid 2PC; use saga (+ compensation) · outbox / CDC for the dual-write problem.

Clocks: nanoTime() for durations/timeouts; never wall-clock for cross-node ordering → logical (Lamport/vector) clocks.

Performance: think in p99, not averages; tail amplifies with fan-out → hedged requests. Little's Law L = λW sizes pools. Memory ≈ 1000× faster than SSD; cross-region RTT ≈ 150 ms floor.

Golden rule: distribution is a cost. Single box until a real constraint forces otherwise.

Self-test (say the answers out loud; that's the interview)#

1. Name the three unavoidable facts about distributed systems and one hard problem each one causes.
2. Why can't a healthy node tell a crashed peer from a slow one, and what one design consequence follows?
3. State CAP precisely. Why is "pick 2 of 3" wrong? What does PACELC add, and which branch dominates real life?
4. You read from a follower right after writing to the leader and see stale data. Name the problem and two fixes.
5. Why is Last-Write-Wins a data-loss strategy? What detects the conflict instead?
6. Given N=5 replicas, pick R and W for read-your-writes with maximum write availability. What's the overlap rule?
7. Why must a consensus cluster be odd-sized? How many failures does a 5-node cluster tolerate?
8. State FLP in one sentence. How do Raft/Paxos make progress despite it?
9. Walk through the dual-write problem and why no try/catch fixes it. Give the two correct solutions.
10. "I'll use exactly-once delivery." What's wrong, and what's the correct framing?
11. Make POST /transfer safe to retry. What exactly do you store, and how do you handle two simultaneous retries?
12. A distributed lock protects a resource, but you still get two writers. What happened and what's the fix?
13. One slow dependency takes down your whole service. Name the failure mode and the four patterns that contain it.
14. Why measure p99 not average, and why does tail latency get worse as you add services to a fan-out?
15. Estimate infra for 50 M DAU × 20 requests/day at 5× peak: req/s average, req/s peak, rough instance count at 500 req/s each.
16. When should you not build a distributed system, and how do you justify that in an interview?
17. Which of your other guides solves each of: cross-service atomicity, reliable event publishing, read/write model separation, cascading-failure isolation?

Lineage & sources: the canonical text is Martin Kleppmann's Designing Data-Intensive Applications (2017). Key papers referenced: Lamport, "Time, Clocks, and the Ordering of Events" (1978); Fischer–Lynch–Paterson, impossibility of async consensus (1985); Gilbert & Lynch's proof of Brewer's CAP conjecture (2002); Abadi, PACELC (2010/2012); DeCandia et al., Dynamo (2007); Ongaro & Ousterhout, Raft (2014); Corbett et al., Spanner (2012); Dean & Barroso, "The Tail at Scale" (2013). Framework specifics (Spring Boot, Resilience4j, Spring Kafka, Micrometer Tracing) are current as of 2026 and version-sensitive — verify annotation/property names against the version on your classpath. Cross-references point to the companion guides in this InterviewPrep folder.