96 Distributed Computing Interview Questions and Answers (2026)

The Fallacies of Distributed Computing are eight false assumptions (identified by Deutsch and Gosling at Sun) that developers habitually make about networks. Believing them produces systems that work in a demo but fail in production.

• The network is reliable: Packets are dropped and connections break; code with no retries, timeouts, or idempotency hangs or corrupts state on the first blip.

• Latency is zero: Remote calls take milliseconds, not nanoseconds; treating a chatty loop of remote calls like local calls kills performance.

• Bandwidth is infinite: Shipping large payloads saturates links and causes queueing and timeouts under load.

• The network is secure: Assuming a trusted network skips encryption/authn, exposing data to interception and spoofing.

• Topology doesn't change: Nodes come and go; hardcoded IPs/hosts break, so you need service discovery.

• There is one administrator, transport cost is zero, and the network is homogeneous: Multiple owners, real serialization/transfer costs, and mixed protocols all break naive designs.

• Why ignoring them fails: Each fallacy removes a defense (timeouts, retries, encryption, backpressure); the system passes tests then falls over the moment real network conditions appear.

Vertical scaling makes one machine bigger (more CPU/RAM), so state stays local and simple; horizontal scaling adds more machines, which forces you to decide where shared state lives because requests can land on any node.

• Vertical scaling (scale up):

  • One node holds all state in local memory/disk, so no coordination is needed.

  • Simpler but capped by hardware limits and a single point of failure.

• Horizontal scaling (scale out):

  • Many nodes serve requests, so local state on one node isn't visible to others.

  • Requires externalizing state (shared DB, cache like Redis) or partitioning/replicating it, which introduces consistency and coordination challenges.

• The state-management crux: Vertical scaling avoids the distributed-state problem; horizontal scaling forces you to solve it (sticky sessions, shared stores, or replication).

An RPC (Remote Procedure Call) lets a program invoke a function that executes on another machine, with the network hidden behind what looks like an ordinary call. The danger is that this abstraction pretends a remote call is a local one, hiding failure, latency, and partial-failure semantics that don't exist locally.

• How it works: A client stub serializes (marshals) arguments, sends them over the network, the server executes and returns a result, and the stub unmarshals it.

• Partial failure: A local call either runs or doesn't; a remote call can fail after execution but before the reply arrives, so you can't tell if it succeeded.

• Latency and cost: Remote calls are orders of magnitude slower; treating them as free encourages chatty designs that perform badly.

• Semantics of retries: Retrying after uncertainty can cause duplicate execution unless the operation is idempotent; you must choose at-least-once vs at-most-once delivery.

• Takeaway: Design RPC APIs acknowledging the network exists: use timeouts, idempotency, and coarse-grained calls rather than pretending it's a local function.

Request/response is one-to-one and pull-oriented: a client asks a known service and waits for its reply. Publish/subscribe is one-to-many and push-oriented: publishers emit events to a topic and any number of subscribers receive them, without either side knowing the other.

• Request/response:

  • Tight coupling: the client must know and address the specific service.

  • Best when the caller needs a direct, immediate answer (queries, commands).

• Publish/subscribe:

  • Loose coupling via a broker/topic: publishers and subscribers don't know each other and scale independently.

  • One event can fan out to many consumers, and you add subscribers without changing the producer.

• Trade-off: Request/response is simpler to trace and debug; pub/sub is more scalable and extensible but adds a broker, eventual delivery, and harder end-to-end visibility.

We build distributed systems because a single machine eventually can't meet real-world demands for capacity, availability, speed, and reach. Spreading work across many machines lets systems grow, survive failures, respond quickly worldwide, and satisfy data locality laws, at the cost of added complexity.

• Scalability: A single machine has hard CPU/RAM/disk ceilings; scaling out across nodes handles workloads no single box can.

• Fault tolerance / availability: One machine is a single point of failure; replicating across nodes lets the system keep serving when components die.

• Low latency: Placing servers near users cuts network round-trip time, so responses feel fast regardless of where users are.

• Geographic distribution: Serving multiple regions gives resilience to regional outages and satisfies data-residency/compliance requirements.

• The trade-off: These benefits come with partial failure, coordination, and consistency challenges (CAP), so you distribute only when the gains justify the complexity.

Partial failure is when some components of a distributed system fail or become unreachable while others keep running, leaving the system in a mixed, indeterminate state. On a single machine an operation typically succeeds or fails atomically and deterministically; across a network you often cannot even tell which happened.

• Definition: Part of the system works and part doesn't, at the same time, often invisibly to callers.

• The core problem: ambiguity:

  • A request that times out could mean the message was lost, the server crashed before processing, or it processed and the reply was lost.

  • The sender cannot distinguish "not done" from "done but unacknowledged."

• Why single-machine reasoning breaks:

  • A local function call fails deterministically and you get a clear exception; a network call adds independent failure of the peer plus the link.

  • No shared clock or shared memory, so there's no global view of who is up.

• Consequences for design:

  • Forces retries, which forces idempotency to avoid double-applying side effects.

  • Needs timeouts, health checks, and failure detectors that are themselves imperfect (can't tell slow from dead).

  • Drives patterns like acknowledgements, deduplication, and reconciliation to converge state.

Strong consistency means every read returns the most recent write, as if there were a single copy; eventual consistency means replicas may temporarily disagree but will converge to the same value if writes stop. The choice is about whether correctness demands seeing the latest value immediately or whether availability and low latency matter more.

• Strong consistency:

  • Reads never see stale data; simple to reason about.

  • Requires coordination, so higher latency and reduced availability during failures.

• Eventual consistency:

  • Reads may be stale for a while; replicas converge in the background.

  • Low latency, stays available under partitions, but application must tolerate/resolve conflicts.

• When to choose strong: Money, inventory, unique constraints, anything where a stale read causes real harm (double-spend, overselling).

• When to choose eventual:

  • High-scale, read-heavy, tolerant data: like counts, feeds, caches, presence indicators.

  • When global availability and latency outweigh momentary staleness.

Redundancy means keeping multiple copies of data or components so no single failure loses data or stops service. Replication (the data form of redundancy) directly boosts availability and durability, but each extra copy that must be kept in sync introduces a consistency cost, which is the heart of the CAP/PACELC tradeoff.

• What replication buys you:

  • Availability: if one replica fails, others serve reads/writes.

  • Durability: data survives disk/node loss.

  • Performance: reads can spread across replicas and sit closer to users.

• The consistency tension:

  • With multiple copies, a write must reach them all; until it does, replicas disagree.

  • Synchronous replication keeps copies consistent but adds latency and can block on a slow/failed replica (favoring C).

  • Asynchronous replication stays fast and available but exposes stale reads and possible data loss on failover (favoring A).

• Quorums as a tuning knob: Choosing R and W over N replicas (e.g. R + W > N) lets you trade consistency against availability/latency explicitly.

• Takeaway: redundancy is what makes fault tolerance possible, but the price is the ongoing work (and tradeoffs) of keeping copies consistent.

Batch processing runs computation over a large, bounded dataset all at once and produces results after the whole set is processed. Stream processing operates on an unbounded, continuous flow of events, processing each as it arrives to give results with low latency.

• Bounded vs unbounded data:

  • Batch: a finite input with a defined start and end (yesterday's logs).

  • Stream: an infinite sequence with no natural end (a live click feed).

• Latency vs throughput:

  • Batch optimizes throughput and is high-latency (minutes to hours).

  • Stream optimizes latency (milliseconds to seconds) per event.

• Time and completeness:

  • Batch sees the whole dataset, so results are complete and easy to reason about.

  • Stream must handle late/out-of-order events using windows and watermarks, trading completeness for freshness.

• Convergence: A batch is really a bounded stream; frameworks like Flink and Spark unify both, and Kappa/Lambda architectures blend them.

A load balancer sits between clients and a pool of backend nodes and distributes incoming requests across them, so no single node is overwhelmed and the system scales horizontally. It also improves availability by routing only to healthy nodes, detected via health checks.

• What it provides: Scalability (spread load), availability (skip failed nodes), and a stable virtual endpoint hiding the fleet behind it.

• Layer of operation:

  • L4 (transport): routes by IP/port, fast and protocol-agnostic.

  • L7 (application): inspects HTTP paths/headers/cookies for smarter, content-aware routing.

• Common strategies:

  1. Round robin: cycle through nodes evenly; simple but ignores actual load.

  2. Weighted round robin: bigger nodes get proportionally more traffic.

  3. Least connections / least load: send to the node with the fewest active requests, better for uneven request costs.

  4. Consistent hashing: map a key (user, session) to a node with minimal reshuffling when nodes join/leave, good for cache/session affinity.

  5. Power of two choices: pick two nodes at random, route to the less loaded; near-optimal with little coordination.

• Supporting mechanics:

  • Health checks eject unhealthy nodes; sticky sessions pin a client to a node when state isn't shared.

  • The balancer itself must be redundant (multiple instances, DNS or anycast) so it isn't a single point of failure.

The Two Generals Problem shows that two parties cannot reach guaranteed agreement over an unreliable channel where messages can be lost. Two generals must attack at the same time to win, but any message (including acknowledgments) can be lost, so neither can ever be certain the other is committed.

• The setup: General A sends "attack at dawn"; A can't act until it knows B received it, so B must acknowledge.

• The infinite acknowledgment regress: B's ack can be lost, so B needs an ack of its ack, which can also be lost. No finite exchange ever makes the last message safe.

• Why 100% certainty is impossible: The sender of the final message can never know it arrived, so common knowledge (both know that both know) is unattainable over a lossy channel.

• Practical takeaway: Real systems don't seek perfect certainty; they use retries, timeouts, and probabilistic confidence, and lean on idempotency so a redelivered message is harmless.

In synchronous communication the caller blocks and waits for a response, coupling the two parties in time; in asynchronous communication the caller sends a message and continues, with the response (if any) arriving later, decoupling them. The trade-off is simplicity and immediacy versus resilience and throughput.

• Synchronous (e.g. HTTP request/response, blocking RPC):

  • Simple to reason about: you get the result inline and handle errors immediately.

  • Temporal coupling: both services must be up at once, and a slow callee blocks the caller, so failures cascade.

• Asynchronous (e.g. message queues, events):

  • Sender and receiver are decoupled in time; a queue buffers load and absorbs downstream outages (better fault tolerance and throughput).

  • More complex: you need correlation IDs, callbacks/eventual results, and handling of out-of-order or duplicate messages.

• How to choose: Use synchronous when the caller genuinely needs the answer now; use asynchronous for decoupling, load leveling, and long-running or fire-and-forget work.

Stateless services scale easily because every instance is interchangeable: any request can go to any node, so you just add more nodes behind a load balancer. Stateful services are harder because a node holds data that other nodes lack, so requests can't be freely routed and state must be kept consistent across instances.

• Why stateless scales:

  • No per-node memory of past requests, so instances are identical and can be added, removed, or replaced freely.

  • Load balancing is trivial and failover is painless (a dead node loses no unique data).

• Why stateful is hard: Data lives on specific nodes, so you must handle replication, consistency, and rebalancing when nodes change.

• Handling state when scaling out:

  • Externalize state: push it into a shared database or cache (e.g. Redis) so app tiers stay stateless.

  • Partition (shard) state so each node owns a slice, and replicate for availability.

  • Sticky sessions as a last resort: route a user to the same node, but this hurts elasticity and failover.

Read-Your-Writes consistency guarantees that once a client has written a value, its own subsequent reads will always reflect that write (or something newer), never an older state. It's a per-client (session) guarantee, not a global one: other clients may still see stale data.

• What it promises:

  • A client never "loses" its own update by reading a replica that hasn't caught up.

  • Common example: you edit your profile, then reload and expect to see the edit.

• Why async replication threatens it: Writes go to a leader, replicas lag; a follow-up read routed to a stale replica shows the old value.

• Implementation strategies:

  1. Read from the leader for data the user recently modified (e.g. within a time window after a write).

  2. Track a write version/timestamp per client and only read from a replica that has applied at least that version; otherwise wait or fall back to the leader.

  3. Sticky routing: pin a session to the same replica so it always sees its own writes.

• Practical notes:

  • The version-token approach is the most robust; store the token in the session or a cookie.

  • Watch cross-device cases: a user on phone then laptop needs the token shared server-side, not just client-local.

Even with a healthy network (no partition), stronger consistency costs latency, because a strongly consistent operation must coordinate with other replicas before responding. This is the ELC (else) half of the PACELC theorem: without a Partition, you still trade between Latency and Consistency.

• Why the trade-off exists without a partition:

  • To guarantee a read sees the latest write, a node must contact a quorum/leader, adding round-trips (often cross-datacenter).

  • Serving from the nearest replica is fast but may return stale data.

• PACELC framing:

  • If Partition: choose Availability or Consistency (the classic CAP case).

  • Else (normal operation): choose Latency or Consistency.

• System examples:

  • Dynamo/Cassandra-style (PA/EL): favor low latency, allow staleness.

  • Spanner-style (PC/EC): favor consistency, accept higher latency from coordination.

• Design lever: Tunable quorums let you slide the dial per operation: strong reads via full quorum, fast reads via one replica.

Linearizability is the strongest single-object model: every read reflects the latest completed write in real-time order, so the system behaves like one up-to-date copy. Eventual consistency is among the weakest: it only promises that, absent new writes, all replicas eventually converge, with no bound on how stale reads can be in the meantime. Eventual consistency is "not enough" whenever a stale or out-of-order read can violate correctness.

• Linearizability:

  • Real-time recency guarantee on individual objects; needs coordination and costs latency/availability.

  • Enables things like distributed locks, leader election, uniqueness checks.

• Eventual consistency:

  • Only convergence in the limit; reads may be arbitrarily stale and see writes out of order.

  • Cheap, highly available, needs conflict resolution (e.g. last-writer-wins, CRDTs).

• When eventual is not enough:

  • Correctness constraints: bank balances, inventory counts, "reserve exactly one seat."

  • Coordination primitives: locks, uniqueness, leader election need a single agreed truth.

  • Read-after-write expectations where users must see their own effect immediately and stale reads are confusing or unsafe.

Both are session guarantees (client-centric consistency) that constrain what a single client observes over time, not what the whole system agrees on. Monotonic reads means you never see time move backwards; monotonic writes means your own writes are applied in the order you issued them.

• Monotonic reads:

  • Once you read a value, later reads never return an older version.

  • Prevents the anomaly of "going back in time": reading an updated comment, then a refresh shows it missing because you hit a stale replica.

• Monotonic writes:

  • Writes from the same client are applied in issue order on all replicas.

  • Prevents a later write landing before an earlier one (e.g. edit-then-delete arriving out of order and resurrecting data).

• Scope:

  • They are per-client (per-session) promises, weaker than linearizability but cheap and very useful under eventual consistency.

  • Commonly implemented by pinning a client to a replica or tracking version/timestamp tokens so a client only reads replicas at least as fresh as what it last saw.

CAP states that a distributed system can guarantee at most two of Consistency, Availability, and Partition tolerance simultaneously. Since partitions are unavoidable in real networks, the real choice during a partition is between staying consistent or staying available.

• The three properties:

  • Consistency: every read sees the latest write (linearizability).

  • Availability: every request to a non-failed node gets a (non-error) response.

  • Partition tolerance: the system keeps operating despite dropped/delayed messages between nodes.

• Why you can't have both C and A during a partition:

  • Two nodes are cut off from each other. A write hits one side.

  • If the other side answers reads, it must either serve stale data (sacrificing C) or refuse to answer until it can sync (sacrificing A). It physically cannot know the latest value.

• CP example: ZooKeeper, etcd, HBase: the minority side rejects requests to avoid stale reads.

• AP example: Cassandra, DynamoDB (in default modes): all sides keep serving and reconcile later (e.g. last-write-wins, read repair).

• Nuance: CAP is only about the partition moment; when healthy, a well-designed system can offer both. It's not a permanent "pick 2" label.

A quorum is the minimum number of replicas that must acknowledge an operation for it to count as successful. With N replicas, W is the write quorum and R is the read quorum; tuning them shifts the system along the consistency vs. availability/latency spectrum.

• The three parameters:

  • N: number of replicas holding each piece of data.

  • W: replicas that must ack a write before it's considered committed.

  • R: replicas that must respond to a read before returning a value.

• Effect on consistency:

  • If R + W > N, the read and write sets overlap by at least one node, so a read is guaranteed to see the latest committed write (strong-ish consistency).

  • If R + W <= N, reads may miss recent writes: eventual consistency.

• Effect on availability/latency:

  • Smaller quorums tolerate more node failures and respond faster (fewer nodes to wait for).

  • Larger quorums are safer but block if too many replicas are down.

• Common choice: N=3, W=2, R=2 gives overlap and tolerates one node failure: a popular balanced default.

R + W > N is the quorum overlap rule: it forces every read set to intersect every write set on at least one replica, guaranteeing reads observe the most recent write. To boost read throughput you lower R (and raise W to keep the inequality), but if you break the inequality you trade away strong consistency.

• Why overlap matters: At least one node in the read quorum was part of the last write quorum, so it holds the newest value.

• To increase read throughput/lower read latency:

  • Set R small (e.g. R=1): reads contact fewer nodes and finish faster.

  • To preserve consistency, raise W (e.g. W=N) so R + W > N still holds.

• The trade-off:

  • Fast reads with R=1 make writes slower and less available (must ack on more replicas).

  • If you drop W too so that R + W <= N, you get fast reads and writes but only eventual consistency (possible stale reads).

• Rule of thumb: Read-heavy + must stay consistent: low R, high W. Read-heavy + tolerant of staleness: low R and low W.

Because network partitions are not a design choice but an inevitable fact of any real network: packets get dropped, links fail, nodes hang. A system that isn't partition tolerant simply breaks (or corrupts data) when this happens, so in practice you must handle partitions, which leaves the real trade-off as C vs. A.

• Partitions are unavoidable:

  • Switch failures, GC pauses, overloaded links, and slow nodes all look like partitions to peers.

  • You can't buy a network that never drops or delays messages.

• "Not partition tolerant" only makes sense on a single node: The moment data spans two machines, the link between them can fail, so P must be handled.

• So CAP is really CA-during-partition:

  • Dropping P would mean freezing or corrupting the system on the first blip: not an acceptable production behavior.

  • Hence engineers keep P and consciously choose C or A for the partition window.

The Byzantine Generals Problem asks how distributed nodes can reach agreement when some participants may be not just dead but actively lying, sending different or malicious messages to different peers. It's strictly harder than crash failures, where a node simply stops and never sends wrong information.

• The problem:

  • Generals must agree to attack or retreat, but traitors send conflicting messages to split the loyal ones.

  • Classic result: with arbitrary (Byzantine) faults you need at least 3f + 1 nodes to tolerate f faulty ones.

• Byzantine vs. fail-stop/crash:

  • Crash: a node stops responding but never gives wrong answers. Detectable by timeouts; tolerated with 2f + 1 nodes (e.g. Raft, Paxos).

  • Byzantine: a node may equivocate, corrupt, or forge data, so honest nodes can't trust any single message.

• When an engineer actually needs BFT:

  • Trustless/adversarial settings: blockchains, cross-organization consensus where participants may cheat.

  • Extreme reliability domains: avionics, aerospace, where hardware may produce arbitrary garbage.

  • Usually NOT needed inside your own trusted datacenter: crash-tolerant consensus is cheaper and enough there.

Graceful degradation is a system's ability to keep serving reduced but useful functionality when some components fail, instead of failing completely. The idea is to isolate failures and drop non-essential features while protecting the core path.

• Serve a degraded response: Return cached, stale, or default data when a downstream dependency is down (e.g. show a generic product page without personalized recommendations).

• Fail non-critical features first: Prioritize core flows (checkout, login) and shed optional ones (analytics, suggestions) under stress.

• Isolation mechanisms:

  • Circuit breakers stop calling a failing dependency and return a fallback immediately.

  • Bulkheads partition resources (thread pools, connections) so one failing dependency can't exhaust everything.

  • Timeouts and fallbacks prevent slow dependencies from blocking callers.

• Design implication: Dependencies must be classified as critical vs optional up front, so the system knows what it can safely drop.

Consistent hashing maps both keys and nodes onto the same circular hash space (a ring); a key is owned by the first node found clockwise from the key's position. Because only the keys in one segment of the ring move when a node joins or leaves, it avoids the near-total reshuffle that hash(key) mod N causes.

• The mod N problem: With hash(key) % N, changing N changes almost every key's assignment, forcing massive data movement and cache invalidation.

• The ring:

  • Hash the node IDs and the keys into the same space (e.g. 0 to 2^32-1) arranged as a circle.

  • Each key belongs to the next node clockwise; that node owns the arc between it and its predecessor.

• Adding a node: It slots into one point on the ring and only steals keys from its immediate clockwise successor: roughly K/N keys move, not all of them.

• Removing a node: Its keys pass to the next node clockwise; other nodes are untouched.

• Result: Movement is bounded to O(K/N), which is why it underpins systems like Dynamo, Cassandra, and distributed caches.

Virtual nodes (vnodes) mean each physical node is placed at many points on the ring instead of one, by hashing several derived identifiers per node. This solves the uneven-load and lumpy-distribution problem that plagues basic consistent hashing.

• Problem with one point per node:

  • With few nodes, random placement leaves arcs of very unequal size, so some nodes get far more keys than others.

  • When a node leaves, its entire load lands on a single successor rather than being spread.

• How vnodes help:

  • Many small arcs per physical node average out, giving smoother, more uniform distribution.

  • On failure, a departing node's many vnode segments are absorbed by many different peers, spreading the extra load.

• Heterogeneity: Assign more vnodes to bigger machines to give them a proportionally larger share.

• Cost: More ring entries to track as metadata; a tuning trade-off between balance and overhead.

Synchronous replication waits for replicas to acknowledge a write before confirming to the client (durable but slow); asynchronous replies immediately and replicates in the background (fast but can lose recent writes on failure). Semi-synchronous is the middle ground: acknowledge after at least one replica has the write, protecting against data loss if the primary dies.

• Synchronous: Strong durability and no data loss on primary failure, but higher write latency and stalls if a replica is slow or down.

• Asynchronous: Lowest latency and stays available if replicas lag, but a crash of the primary before replication loses acknowledged writes (a durability/consistency gap).

• Semi-synchronous:

  • Wait for one (or a quorum) of replicas to confirm, letting the rest catch up asynchronously.

  • Balances latency against durability.

• What semi-sync protects against: The scenario where the primary acknowledges a write and then crashes before any replica has it, leaving the write lost forever; semi-sync guarantees at least one other copy exists before acknowledging.

These are three replication topologies differing in where writes are accepted. Leader-based routes all writes to one node; multi-leader accepts writes at several leaders that sync with each other; leaderless lets clients write to any/many replicas directly, resolving disagreements via quorums.

• Leader-based (single-leader / primary-backup):

  • One leader takes all writes and streams them to read-only followers.

  • Simple, avoids write conflicts, but the leader is a bottleneck and needs failover; used by most SQL databases.

• Multi-leader:

  • Multiple leaders (often one per datacenter) each accept writes and replicate to the others.

  • Better write availability and locality, but concurrent writes to the same key create conflicts needing resolution (e.g. last-write-wins, CRDTs).

• Leaderless (Dynamo-style):

  • No designated leader; clients (or a coordinator) send reads and writes to many replicas.

  • Uses quorums (W + R > N) plus read repair and anti-entropy to converge; highly available but weaker consistency. Used by Cassandra and DynamoDB.

• Key trade-off axis: Moving from single to multi to leaderless increases write availability and complexity of conflict handling.

Range partitioning assigns contiguous key ranges to partitions (sorted), while hash partitioning maps a key through a hash function and assigns it by the hash value. Range favors efficient range scans but risks hotspots; hash spreads load evenly but destroys key ordering.

• Range partitioning:

  • Keys stay sorted, so scans like WHERE ts BETWEEN a AND b hit few contiguous partitions.

  • Downside: sequential keys (timestamps, auto-increment IDs) all land on one partition, creating write hotspots.

• Hash partitioning:

  • A hash of the key distributes writes/reads uniformly, so load is balanced even with sequential keys.

  • Downside: adjacent keys scatter across partitions, so range queries must fan out to all partitions (scatter/gather).

• The core tradeoff:

  • Range = good locality/range queries, weak load balancing.

  • Hash = good load balancing, weak range queries.

• Hybrid: compound keys hash a prefix (e.g. user_id) for spread and keep a sorted suffix (e.g. timestamp) for ranges within that prefix.

Rebalancing is the redistribution of data (and load) across nodes when partitions/nodes are added or removed. The goal is to move as little data as possible while keeping the distribution even and the system serving traffic throughout.

• The anti-pattern: hash mod N: Using hash(key) % N means changing N remaps almost every key: massive data movement.

• Fixed number of partitions: Create many more partitions than nodes up front; adding a node just reassigns whole partitions, moving only their share of data.

• Consistent hashing: Places nodes and keys on a ring so adding/removing a node only relocates keys between adjacent segments; virtual nodes smooth the distribution.

• Dynamic partitioning: Split a partition when it grows too large and merge when small (like B-tree/HBase regions), so partition count adapts to data volume.

• Operational guardrails: Throttle/rate-limit migration and prefer manual or supervised rebalancing to avoid overloading the network and cascading failures.

Routing is deciding which node holds a given key so a request reaches it. The choices differ in where the partition-to-node mapping lives: in a client, in a routing tier, or in the nodes themselves. This is the service-discovery problem applied to partitions.

• Client-side routing: The client knows the mapping and contacts the right node directly: one fewer hop, but every client must track topology changes.

• Routing tier / proxy: A dedicated layer (coordinator/proxy) looks up the partition and forwards: clients stay simple, but the tier is an extra hop and must be highly available.

• Node-based (gossip/redirect): Any node can receive a request and forward or redirect it to the owner (Cassandra-style gossip, Redis Cluster MOVED redirects): no separate tier, but nodes carry routing complexity.

• The shared hard problem: Keeping the mapping consistent as partitions rebalance; often solved with a coordination service like ZooKeeper or etcd, or via gossip for eventual agreement.

A Lamport clock is a single monotonic counter each process keeps to timestamp events and impose a consistent total-ish ordering; a vector clock keeps one counter per process, capturing causality precisely. The problem vector clocks solve that Lamport clocks cannot is detecting concurrency: Lamport clocks can order causally related events but cannot tell whether two events are causally related or truly concurrent.

• Lamport clock:

  • Each process holds one integer counter; increment on local events, and on receive set counter = max(local, received) + 1.

  • Guarantees: if a happens-before b then C(a) < C(b). The converse does NOT hold.

• Vector clock:

  • Each process holds a vector of counters, one entry per process; on send it attaches the vector, on receive it takes the element-wise max then increments its own entry.

  • Comparison: V(a) < V(b) means a causally precedes b; if neither dominates, the events are concurrent.

• The key difference:

  • Lamport gives a one-way implication (causal implies smaller counter) so it cannot distinguish causality from coincidence.

  • Vector clocks give an if-and-only-if, so they can definitively identify concurrent events (needed for conflict detection in systems like Dynamo).

• Cost tradeoff: Lamport is O(1) space; vector clocks are O(N) in the number of processes.

Happens-before (written a → b) is a partial ordering of events defined by Lamport that captures potential causality: a could have influenced b. Two events are concurrent when neither happens-before the other.

• Rules that define a → b:

  1. Same process: if a occurs before b in the same process, then a → b.

  2. Message passing: if a is a send and b is the matching receive, then a → b.

  3. Transitivity: if a → b and b → c, then a → c.

• Concurrency:

  • Events a and b are concurrent (a ∥ b) if NOT (a → b) and NOT (b → a): no chain of process order or messages links them.

  • Concurrent means causally independent, not simultaneous in wall-clock time.

• How to detect it in practice:

  • Vector clocks make it decidable: compare vectors element-wise; if neither dominates the other, the events are concurrent.

  • Plain Lamport timestamps cannot decide concurrency because unrelated events can still have ordered counters.

Physical clocks like those synced by NTP are never perfectly aligned across machines and can even jump backward, so comparing two timestamps from different nodes can produce a false ordering. Clock skew is the instantaneous difference between two clocks, and it means a log entry written 'later' can carry an earlier timestamp than one written before it.

• Why NTP is not enough for ordering:

  • NTP synchronizes clocks only to within a bounded error (typically milliseconds over WAN), not exactly.

  • NTP can step the clock backward to correct drift, so timestamps are not monotonic.

  • Physical time carries no causal meaning: equal or close timestamps say nothing about which event caused which.

• Clock skew defined: The difference in reading between two clocks at a given instant (contrast with drift, the rate at which they diverge).

• Impact on distributed logs:

  • Merging logs by wall-clock timestamp can reorder cause and effect (a reply appearing before its request).

  • Last-Write-Wins conflict resolution can silently drop the actually-newer write.

  • Fix: use logical ordering (Lamport or vector clocks) for causality, or bounded-uncertainty schemes like TrueTime / hybrid logical clocks when real time matters.

A Lamport clock is a simple monotonically increasing integer counter maintained by each process to timestamp events. It lets us build a partial ordering consistent with causality without any synchronized physical clock, by ensuring that if one event can causally affect another, its timestamp is strictly smaller.

• The algorithm:

  1. Before any local event, increment the counter.

  2. When sending a message, attach the current counter value.

  3. On receiving, set counter = max(local, received) + 1.

• Why it captures partial order:

  • The rules guarantee the clock condition: a → b implies C(a) < C(b).

  • It is partial, not total: two concurrent events may share equal or arbitrarily-ordered counters.

• Making it a total order: Break ties with a process ID: order by (counter, processId). Useful for deterministic tie-breaking (e.g. mutual exclusion).

• Limitation: C(a) < C(b) does NOT imply a → b, so Lamport clocks cannot detect concurrency (that needs vector clocks).

Last-Write-Wins is a conflict-resolution strategy where, when two replicas hold conflicting versions of a value, the one with the highest timestamp wins and the other is discarded. It is dangerous with unsynchronized physical clocks because the 'latest' timestamp may not correspond to the write that actually happened last, so a genuine update can be silently lost.

• How LWW works:

  • Each write carries a timestamp; on conflict the higher timestamp is kept. Simple, stateless, no coordination.

  • Used in Cassandra and Dynamo-style stores for its simplicity.

• Why clock skew makes it dangerous:

  • A node with a fast clock stamps future timestamps; its stale writes can永 override newer writes from a correct node.

  • Lost updates: an acknowledged write silently vanishes because a peer's clock was ahead.

  • Backward NTP corrections can make a later write look older.

• Mitigations:

  • Use logical or hybrid logical clocks so timestamps respect causality.

  • Preserve conflicts as sibling versions (vector clocks) and resolve at the application layer instead of blindly discarding.

Clock skew is the instantaneous difference between the readings of two clocks on different machines at the same real instant. In distributed transactions it corrupts ordering: timestamps assigned by different nodes cannot be compared reliably, so transactions may appear to commit in an order that violates their real causal or temporal sequence.

• What skew is:

  • Two clocks read different times right now; even with NTP the offset is bounded but nonzero (milliseconds).

  • Distinct from drift, which is the ongoing rate of divergence that produces skew over time.

• Impact on transaction ordering:

  • Timestamp-based ordering can assign a later transaction a smaller timestamp, breaking serializability.

  • Consistency anomalies: a read may miss a write that already committed because its timestamp appears in the future ('causal reverse').

  • Snapshot isolation and MVCC that key off timestamps can return stale or inconsistent snapshots.

• How systems cope:

  • Bound the uncertainty and wait it out: Spanner's TrueTime commit-wait delays until the skew interval passes.

  • Hybrid logical clocks combine physical time with a logical counter to enforce causal monotonicity despite skew.

A Lamport clock is a logical counter each process keeps to order events causally: it captures "happens-before" but cannot resolve concurrency, so equal timestamps tell you nothing about real order.

• The algorithm (three rules):

  1. Each process holds a counter C, starting at 0.

  2. Before any local event or send, increment C = C + 1.

  3. On receiving a message with timestamp t, set C = max(C, t) + 1.

• What it guarantees:

  • If event A happened-before B (A → B), then C(A) < C(B).

  • The implication is one-directional only.

• Why equal (or even ordered) timestamps don't decide order:

  • The converse fails: C(A) < C(B) does NOT imply A → B; they may be concurrent.

  • Two events on different processes can hold the same value with no causal link, so "which happened first" is undefined.

• Practical fix for a total order: Break ties with a process ID, giving a deterministic but arbitrary total order (not a causal one).

A vector clock keeps one counter per process, so comparing two vectors reveals whether one event causally precedes another or whether they happened independently (concurrent), which a single Lamport number cannot.

• Structure and update rules:

  • Each process i holds a vector V of size N (one entry per process).

  • On a local event, increment its own slot: V[i] += 1.

  • On receive, take the element-wise max of both vectors, then increment its own slot.

• Comparing two vectors:

  • A → B (causal) if every element A[k] <= B[k] and at least one is strictly less.

  • Concurrent if neither vector dominates the other: each has some slot greater than the other.

• Concurrent / conflicting scenario:

  • Two clients read the same value, then each writes based on it without seeing the other's write.

  • Example: from [1,0] one node produces [2,0] while another produces [1,1]; neither dominates, so the system flags a conflict.

  • Systems like Dynamo keep both siblings and defer resolution (last-write-wins or app-level merge).

Replication lag is the delay between a write committing on the leader and it appearing on read replicas. During that window replicas serve stale data, and the most jarring anomaly is a user not seeing their own just-made update. You fix it by routing reads that must reflect the user's writes to fresh data (read-your-writes consistency).

• Anomalies caused by lag:

  • Read-your-writes violation: you post a comment, refresh, and it's gone.

  • Monotonic-read violation: successive reads hit different replicas and appear to move backward in time.

  • Causal violation: you see an answer before the question it replies to.

• Designing for read-your-writes:

  1. After a user writes, read their own data from the leader for a short window (e.g. a few seconds).

  2. Track a per-user write timestamp/version and only read from a replica that has caught up to it.

  3. Or pin a session to one replica and wait for it to reach the write's log position (session consistency).

• Trade-off: Routing more reads to the leader reduces its scaling benefit, so scope the guarantee to the user's own recent writes rather than all reads.

Eventual consistency only promises replicas converge "eventually" if writes stop, so in practice users can read stale or non-monotonic data. Session guarantees like read-your-writes and monotonic reads narrow those windows to make the experience feel consistent even though the store underneath is not.

• Practical risks:

  • Stale reads: users act on outdated data (double-book, resend).

  • Time appearing to go backward when consecutive reads hit different replicas.

  • Lost-update / conflicting writes needing later reconciliation.

  • No bound on "eventually": under partitions convergence can take a long time.

• Read-your-writes: Guarantees a user always sees the effect of their own prior writes, so their edits never seem to vanish.

• Monotonic reads:

  • Guarantees that once you've seen a value, you never see an older one: reads only move forward in time.

  • Usually implemented by pinning a session to a replica or tracking a minimum version.

• Key point: These are per-session (client-centric) guarantees; they improve perceived consistency without the cost of global strong consistency.

Eventual consistency trades immediate correctness for high availability and low latency: any replica can accept reads/writes without coordination, but replicas may temporarily diverge and need a merge strategy to converge once updates propagate.

• What you gain:

  • Availability during partitions (the AP side of CAP) and low-latency local reads/writes.

  • Horizontal scale, since no cross-node lock or quorum is needed on the hot path.

• What you give up: Stale reads and no linearizable global order; application must tolerate temporary divergence.

• How convergence happens:

  • Anti-entropy: background sync (gossip, Merkle-tree comparison) reconciles replica differences.

  • Conflict resolution when two writes clash: last-write-wins by timestamp, vector-clock sibling detection, or CRDT merge.

  • Read repair: divergences detected on read are corrected inline.

• The catch: LWW can silently drop a concurrent write, so the resolution policy must match business needs (order matters).

Raft elects a single leader that owns the replicated log; followers passively replicate, and a candidate is a follower campaigning for votes after a timeout. It prevents two leaders per term because a term is a monotonically increasing number and winning requires a majority of votes, and each node grants only one vote per term.

• The three roles:

  • Leader: handles all client writes, appends to its log, and replicates via AppendEntries (also used as heartbeats).

  • Follower: passive; responds to leader and candidate RPCs, resets its election timer on valid contact.

  • Candidate: a follower whose election timeout fired; increments the term and requests votes.

• Why never two leaders in one term:

  1. Terms act as a logical clock; each election is tied to a unique, increasing term number.

  2. To become leader a candidate needs votes from a majority (quorum) of nodes.

  3. Each node votes at most once per term, and two majorities of the same cluster must overlap, so at most one candidate can reach a majority.

• Handling split votes and stale leaders:

  • Randomized election timeouts make simultaneous candidacies rare; a tie just triggers a new term and retry.

  • Any node seeing a higher term immediately steps down, so a partitioned old leader can't keep committing.

Leader election is the process by which nodes agree on a single coordinator responsible for ordering writes and driving the protocol. When the leader is partitioned from its followers, the majority side detects missing heartbeats, elects a new leader in a higher term, and continues; the isolated old leader can no longer commit anything because it lacks a quorum.

• What leader election achieves:

  • Picks one node to serialize operations, avoiding the complexity of every node proposing.

  • Requires a majority (quorum) vote so two leaders can't both win the same term.

• On partition, the majority side:

  • Followers time out waiting for heartbeats, bump the term, and elect a new leader.

  • Since it holds a quorum, it can keep committing writes: the system stays available.

• The isolated old leader:

  • May still think it is leader but cannot reach a majority, so it cannot commit new entries.

  • On rejoining it sees the higher term and steps down, discarding uncommitted entries.

• Split-brain avoidance: The quorum requirement guarantees the minority side never commits, so there are never two authoritative histories.

Heartbeats are periodic messages the leader sends to assert it is alive; the election timeout is how long a follower waits without hearing one before assuming the leader is dead and starting a new election. Together they form a timeout-based failure detector: heartbeats reset each follower's timer, and an expired timer triggers a leadership change.

• How they interact:

  • The leader sends heartbeats at an interval much shorter than the election timeout.

  • Each received heartbeat resets the follower's election timer; silence lets it expire and triggers a candidacy.

  • Timeouts are randomized per node so followers don't all become candidates at once (avoids split votes).

• If the election timeout is too low:

  • Normal network jitter or a brief GC pause looks like a failure, causing spurious elections and leader churn.

  • Constant elections stall real work and can thrash the system.

• If the election timeout is too high: A genuinely dead leader goes undetected for a long time, hurting availability and increasing latency for writes.

• Rule of thumb: heartbeat interval << election timeout << mean time between failures; set the timeout comfortably above worst-case round-trip time.

These are increasingly strong guarantees about the order in which messages are delivered. FIFO preserves per-sender order, causal additionally preserves cause-and-effect order across senders, and total order forces all receivers to agree on one global sequence.

• FIFO order:

  • Messages from the same sender are delivered in the order that sender sent them; different senders are unconstrained.

  • Matters for per-connection streams: e.g. a single client's sequential updates.

• Causal order:

  • If message A happened-before B (Lamport's relation), everyone delivers A before B; concurrent messages may differ.

  • Matters when replies must not arrive before what they respond to: comment threads, chat, collaborative editing.

• Total order:

  • All nodes deliver all messages in the same order, even causally unrelated ones.

  • Matters for replicated state machines and strongly consistent replication where replicas must not diverge.

• Relationships:

  • Causal order implies FIFO; total order can be combined with causal (causal + total).

  • Cost rises with strength: total order generally requires consensus, FIFO is nearly free per link.

Idempotency means performing an operation multiple times has the same effect as performing it once. It is critical because networks make retries unavoidable: timeouts and lost acknowledgements mean a client often can't tell if a request succeeded, so safe retries require idempotent operations.

• Why distributed systems need it:

  • At-least-once delivery and client retries cause duplicate messages; without idempotency you get double charges, duplicate orders, etc.

  • It turns unreliable networks into safely retryable operations, the basis of resilient clients.

• Designing an idempotent API:

  • Use an idempotency key: client sends a unique key (e.g. UUID) per logical operation.

  • Server stores the key with the result; on a repeat, it returns the stored result instead of re-executing.

  • Make the dedup check and the write atomic (unique constraint or transaction) to avoid races between concurrent retries.

  • Set a retention TTL on keys and define behavior for same-key-different-payload (reject as a conflict).

• Leverage HTTP semantics: GET, PUT, DELETE are naturally idempotent; POST is not, which is why POST especially needs idempotency keys.

Two-Phase Commit (2PC) is an atomic commitment protocol that lets multiple participants agree to either all commit or all abort a distributed transaction, coordinated by a single coordinator. Its biggest weakness is that if the coordinator crashes after participants have voted, they can block indefinitely holding locks.

• Phase 1: prepare (voting):

  • Coordinator asks all participants to prepare; each does the work, writes it durably, and replies yes (prepared) or no.

  • A yes vote is a binding promise: the participant must be able to commit if told to.

• Phase 2: commit/abort:

  • If all voted yes, the coordinator logs commit and tells everyone to commit; any no leads to a global abort.

  • Participants apply the decision and acknowledge.

• The blocking weakness:

  • A participant that voted yes but hasn't heard the decision is stuck: it can't unilaterally commit or abort, so it holds locks until the coordinator recovers.

  • This is why 2PC is called a blocking protocol; it is not partition-tolerant.

• Mitigations:

  • Durable write-ahead logging lets nodes recover state after restart.

  • Three-Phase Commit or consensus-backed coordinators (Paxos/Raft) reduce blocking, at higher cost.

The Saga pattern models a distributed transaction as a sequence of local transactions, each in its own service, where every step has a compensating action that semantically undoes it if a later step fails. Instead of holding a distributed lock across services, it accepts intermediate visibility and restores consistency through compensation.

• Sequence of local transactions: Each service commits its own local ACID transaction and publishes an event/command to trigger the next step.

• Compensating transactions instead of rollback:

  • If step 4 fails, the saga runs compensations for steps 3, 2, 1 in reverse (e.g. refund a charge, release inventory).

  • Compensation is semantic undo, not a true rollback: the original commit already happened.

• No distributed lock needed: Locks are held only briefly within each local transaction, never across the whole workflow, avoiding the availability cost of two-phase locking.

• Trade-off: consistency model: Provides eventual consistency with no isolation, so other reads can see intermediate states; you may need countermeasures (semantic locks, commutative updates).

These describe how many times a message may be delivered when failures and retries occur. At-most-once may lose messages but never duplicates; at-least-once never loses but may duplicate; exactly-once means one effective delivery, achieved only through extra mechanisms.

• At-most-once:

  • Send and don't retry (fire-and-forget): if it's lost, it's gone.

  • Lowest latency/overhead; acceptable for metrics or telemetry where loss is tolerable.

• At-least-once:

  • Retry until acknowledged, so duplicates happen when acks are lost.

  • Most common default; pair with idempotent consumers.

• Exactly-once:

  • Each message takes effect once: no loss, no duplicate effect.

  • Not free on the wire: needs dedup via unique IDs, idempotency keys, or transactional commits.

• Core trade-off: You choose which failure to tolerate: lost messages (at-most-once) or duplicates (at-least-once); exactly-once removes both but adds cost and complexity.

The Transactional Outbox pattern guarantees that a database update and the message announcing it happen atomically, by writing the message into an outbox table within the same local transaction as the business data. A separate process then reads the outbox and publishes to the broker, eliminating the dual-write problem.

• The problem it solves (dual write): Updating the DB and publishing to a broker are two systems with no shared transaction: a crash between them leaves them inconsistent (order saved but event never sent, or vice versa).

• How it works:

  • In one local DB transaction, write the business row AND insert a row into the outbox table; both commit or neither does.

  • A relay (message relay / CDC reading the DB log, e.g. Debezium) polls or tails the outbox and publishes each message to the broker.

  • After successful publish, the row is marked sent or deleted.

• Delivery semantics: Gives at-least-once: the relay may crash after publishing but before marking, causing a resend, so consumers must be idempotent.

• Benefit: Atomicity relies only on the local DB transaction, no distributed transaction or 2PC across DB and broker.

Gossip is a decentralized, probabilistic protocol where each node periodically exchanges state with a few randomly chosen peers, so information spreads through the cluster like an epidemic without any central registry.

• How it spreads:

  • On each round (e.g. every 1s) a node picks a random subset of peers and shares its view of the membership list.

  • Updates propagate exponentially: reach grows roughly as a power of the fanout, so thousands of nodes converge in O(log N) rounds.

• Why no central registry is needed:

  • Every node holds an eventually-consistent copy of the membership list; there is no single source of truth to query or to fail.

  • Membership changes (joins, leaves, failures) are just new state gossiped like anything else.

• Reconciling conflicting state:

  • Each entry carries a version/heartbeat counter or timestamp; nodes keep the higher version, so stale info is overwritten.

  • Failure is inferred: if a node's heartbeat stops advancing for a timeout, peers mark it suspect then dead and gossip that.

• Properties:

  • Scalable and robust: load per node is bounded (fixed fanout) and no single point of failure.

  • Trade-off: only eventual consistency and probabilistic delivery, so views can briefly disagree.

• Used in production by Cassandra, Consul, and DynamoDB-style systems.

Split brain occurs when a network partition (or slow node) causes two subsets of a cluster to each believe they are the sole active leader, so both accept writes and diverge. Quorums prevent two leaders from existing at once, and fencing tokens stop a stale leader's writes from being accepted.

• The problem:

  • A partition splits nodes; each side elects/keeps a leader and processes conflicting writes, corrupting shared state.

  • Common with naive failover: the old leader isn't truly dead, just unreachable.

• Quorums (majority):

  • A leader or write is only valid if it has agreement from a majority (N/2 + 1) of nodes.

  • Only one side of a partition can hold a majority, so at most one leader can commit: the minority side cannot make progress.

• Fencing tokens:

  • Each granted leadership/lock includes a monotonically increasing number (the token).

  • Downstream resources (storage, DB) reject any request carrying a token lower than the highest seen, so a revived stale leader is fenced out even if it still thinks it's in charge.

• Together: Quorum prevents two current leaders; fencing prevents an obsolete leader from causing damage after the fact.

Failover is the automatic promotion of a standby (replica) to take over when the active/primary node fails. The hard part is doing it without inconsistency: you must reliably detect real failure and ensure only one primary is ever writable, or you risk split brain and data divergence.

• Detecting failure:

  • Heartbeats/health checks with timeouts; missing several in a row marks the primary suspect.

  • The core difficulty: you cannot distinguish a crashed node from a slow or partitioned one, so timeouts are always a guess.

• Promoting safely:

  • Require a quorum to agree the primary is gone and to elect the new one, so a minority partition can't promote its own.

  • Pick the most up-to-date replica (furthest along in the replication log) to minimize lost writes.

• Avoiding inconsistency:

  • Fence the old primary: revoke its lease and use fencing tokens so its late writes are rejected if it revives.

  • STONITH ("shoot the other node in the head") physically isolates the old primary in some HA setups.

• Inherent trade-offs:

  • Async replication: fast failover but possible data loss (unacked writes on the dead primary).

  • Timeout tuning: too short causes needless failovers, too long increases downtime.

Service discovery is how a client finds the current network location (host:port) of the instances of a service it needs to call. It's a coordination problem because in dynamic systems instances constantly appear, disappear, and move (autoscaling, deploys, failures), so the mapping of names to addresses is shared, changing state that many parties must agree on and observe.

• Why static config fails: Hardcoded IPs break the moment a container is rescheduled or scaled; ephemeral instances get new addresses constantly.

• The mechanism:

  • A registry holds live instances; services register on startup and are removed on failure or shutdown.

  • Health checks and TTL/ephemeral entries evict dead instances so clients don't route to them.

  • Clients look up (via DNS, a client library, or a sidecar proxy) and often load-balance across returned instances.

• Why it's a coordination problem:

  • The registry itself is shared distributed state, usually backed by a consensus store (etcd, Consul, ZooKeeper) to stay consistent and available.

  • There's a consistency/staleness trade-off: propagating changes fast enough that clients don't hit dead nodes, without overwhelming the system.

• Patterns:

  • Client-side discovery: client queries the registry and chooses an instance.

  • Server-side discovery: a load balancer / proxy does it, hiding the registry from clients.

In the coordinator/worker pattern one node (the coordinator/master) assigns and tracks work while many workers execute tasks and report back. It's simple and centralizes decision-making, but the coordinator becomes a single point of failure and a scalability bottleneck.

• How it works:

  • The coordinator splits a job into tasks, dispatches them to workers, monitors progress, and reassigns failed tasks.

  • Workers are stateless and interchangeable, so you scale throughput by adding more of them.

  • Examples: Hadoop MapReduce (JobTracker), Spark driver, Kubernetes control plane scheduling.

• Why it's useful: Centralized state makes scheduling, deduplication, and global decisions straightforward.

• Risks of a single coordinator:

  • Single point of failure: if it dies, no new work is scheduled and progress halts.

  • Bottleneck: all coordination traffic funnels through it, capping cluster size.

  • Split brain if you naively run two coordinators for HA and both think they're active.

• Mitigations:

  • Run standby coordinators with leader election (via ZooKeeper/etcd) and fencing so exactly one is active.

  • Persist coordinator state so a new leader can recover in-flight work.

  • Make tasks idempotent so reassignment after a crash doesn't corrupt results.

The thundering herd is when many clients (or threads) wake up and act at the same instant, hammering a shared resource in a synchronized wave that overwhelms it. Jitter spreads those attempts out over time, and exponential backoff shrinks the total request rate as failures persist.

• Where it happens:

  • A cache entry expires and every request simultaneously stampedes the database to rebuild it.

  • A service recovers and all clients retry at once, knocking it back down.

• Exponential backoff: Each retry waits longer: roughly base * 2^attempt, so aggregate load drops steeply while the target recovers.

• Jitter:

  • Adds randomness to each delay so clients desynchronize instead of retrying on the same tick.

  • Without jitter, backoff just moves the synchronized spike to a later time.

• Common form: full jitter: Sleep a random value in [0, base * 2^attempt], which AWS found best flattens the load.

• Complementary mitigations: Request coalescing/single-flight so only one caller rebuilds a hot cache key; circuit breakers to stop retrying a dead dependency.

MapReduce processes huge datasets by splitting work into a parallel map phase (transform each input record into key/value pairs) and a reduce phase (aggregate all values sharing a key), with a shuffle in between. Data locality means moving computation to the machine that already holds the data, because shipping code is far cheaper than shipping terabytes over the network.

• Map phase: Input is split into chunks; each mapper emits intermediate (key, value) pairs. Runs fully in parallel across nodes.

• Shuffle and sort: The framework groups intermediate pairs by key and routes each key's values to the same reducer. This is the main network-heavy step.

• Reduce phase: Each reducer receives one key and its list of values, producing the final aggregated output.

• Why data locality is core:

  • Data lives in a distributed file system (like HDFS) already spread across nodes; the scheduler places map tasks on the node holding that block.

  • Network bandwidth is the scarce resource at scale, so a few KB of code travels instead of gigabytes of input.

  • This lets throughput scale near-linearly by adding commodity machines rather than being bottlenecked on a central data store.

• Fault tolerance: Tasks are deterministic and idempotent, so a failed map/reduce task is simply re-run elsewhere without corrupting results.

Backpressure is a feedback mechanism where a downstream consumer signals upstream producers to slow down when it can't keep up. It's critical because without it, an overloaded service silently accumulates work until it exhausts memory or queues and fails catastrophically, often cascading the failure upstream.

• The core problem: If arrival rate exceeds processing rate, unbounded buffering leads to growing latency, memory exhaustion, then a crash.

• What signaling achieves:

  • It converts an implicit, dangerous overload into an explicit, controlled slowdown the producer can react to.

  • Fast failure (a quick 429 or reject) is healthier than slow, silent degradation.

• How it's signaled:

  • Explicit: HTTP 429 Too Many Requests, 503, or a Retry-After header.

  • Implicit: bounded queues that block or reject, TCP flow control, or credit-based schemes (reactive streams request N items).

• Why cascading matters: Without backpressure a slow service holds upstream connections open, exhausting their thread pools too, spreading the outage. Backpressure plus load shedding contains the blast radius.

A distributed file system like GFS or HDFS stores files far larger than any single machine by splitting them into large fixed-size blocks (e.g. 64/128 MB) and spreading those blocks across many commodity servers, with a master tracking metadata. It achieves fault tolerance by replicating each block on multiple nodes so data survives disk and machine failures, which are expected, not exceptional.

• Architecture:

  • A single master/NameNode holds metadata: the namespace and which nodes hold which blocks.

  • Many chunkservers/DataNodes store the actual block data and serve reads/writes directly to clients.

  • Clients ask the master where a block lives, then stream data straight from the DataNode (metadata path is separate from data path).

• Large blocks: Big blocks reduce metadata volume and favor high-throughput sequential scans over low-latency random access.

• Fault tolerance via replication:

  • Each block is replicated (default 3 copies), ideally across racks so a rack failure doesn't lose data.

  • DataNodes send heartbeats; if one dies, the master re-replicates its blocks from surviving copies to restore the replication factor.

  • Checksums detect silent corruption; bad replicas are discarded and re-copied.

• Master resilience: Metadata is logged and checkpointed; modern HDFS runs a standby NameNode to avoid a single point of failure.

• Design assumption: Built for write-once/append-mostly, read-heavy workloads on cheap hardware where failure is routine.

Retries need backoff and jitter because a failing dependency is usually already overloaded, and immediate synchronized retries pile on more load, turning a transient blip into a sustained outage. Backoff progressively reduces retry pressure, and jitter desynchronizes clients so they don't all retry in lockstep.

• How naive retries make it worse:

  • Retry storms: a slow service causes timeouts, every client retries, effective load multiplies (3x with 3 attempts) exactly when the service is weakest.

  • Synchronized waves: fixed-interval retries create repeating spikes that prevent recovery.

  • Metastable failure: the system stays down even after the original trigger passes, sustained purely by retry load.

• What backoff fixes: Exponentially growing waits (base * 2^attempt) shrink aggregate request rate, giving the dependency room to drain its queue.

• What jitter fixes: Randomizing delays spreads retries across time so recovering capacity isn't instantly consumed by a synchronized burst.

• Additional guardrails:

  • Cap max attempts and total time; only retry idempotent, retryable errors.

  • Use circuit breakers and retry budgets (cap retries to a small fraction of traffic) to prevent storms entirely.

Strong consistency makes the system behave as if there is a single, up-to-date copy of the data with a global order over all operations. Causal consistency is weaker: it only guarantees that operations related by cause-and-effect are seen in the same order by everyone, while concurrent (unrelated) operations may be seen in different orders. You prefer causal when you want high availability and low latency but still need interactions to make sense.

• Strong consistency:

  • Every reader agrees on one total order; a read sees the latest write.

  • Requires coordination (consensus/quorums), so it costs latency and cannot stay fully available under partition.

• Causal consistency:

  • Preserves happens-before: if A caused B, no one sees B without A.

  • Concurrent writes can be applied in different orders on different replicas (needs conflict resolution).

  • Achievable with high availability and no cross-node coordination on the read/write path.

• When to prefer causal:

  • Collaborative and social features: a reply must not appear before the comment it answers.

  • Geo-distributed systems where cross-region coordination latency is unacceptable.

  • When you need availability during partitions but plain eventual consistency produces nonsensical orderings.

They're guarantees about different things. Linearizability is about single-object operations and real-time ordering: each operation appears to take effect instantly at some point between its start and end. Serializability is about multi-object transactions: their concurrent execution is equivalent to some serial order. A system can be both, and the combination is called strict serializability.

• Linearizability (consistency model):

  • Concerns single registers/objects, one operation at a time.

  • Respects real-time: if op A finishes before op B starts, A is ordered before B.

• Serializability (isolation level):

  • Concerns transactions grouping many operations across many objects.

  • Only requires equivalence to some serial order, not necessarily the real-time order (a serializable schedule may reorder non-overlapping transactions).

• Can it be both?:

  • Yes: strict serializability = serializable transactions whose serial order also respects real time.

  • Example: Google Spanner targets this using TrueTime.

• One-line contrast: Linearizability = correctness of individual objects in real time; serializability = correctness of transactions as an equivalent serial run.

Both provide a single global order of operations that all processes agree on, but linearizability additionally requires that order to respect real (wall-clock) time, whereas sequential consistency only requires it to respect each process's own program order. Linearizability is strictly stronger and harder to achieve because honoring real-time ordering demands cross-node coordination.

• Sequential consistency:

  • Some single interleaving exists that preserves each process's program order.

  • That interleaving need not match real time: an operation that finished earlier in wall-clock time may be ordered later.

• Linearizability:

  • Adds the real-time constraint: if A completes before B begins, A precedes B in the order.

  • This makes it composable: a system built from linearizable parts is itself linearizable.

• Which is harder and why:

  • Linearizability, because respecting real-time recency means a read must reflect the latest completed write globally, forcing coordination (quorums/consensus) and adding latency.

  • Sequential consistency can be satisfied more cheaply since it only cares about per-process order, not global timing.

PACELC extends CAP by noting that trade-offs exist even when the network is healthy. It reads: if there is a Partition (P), choose between Availability (A) and Consistency (C); Else (E), when running normally, choose between Latency (L) and Consistency (C).

• CAP only describes the partition case: It says nothing about the common case: no partition, system fully connected.

• The "ELC" half is the key insight:

  • Even with no failures, keeping replicas strongly consistent requires coordination (waiting for quorum/acks), which adds latency.

  • So you trade latency against consistency all the time, not just during failures.

• Classification examples:

  • Dynamo/Cassandra are typically PA/EL: favor availability under partition and low latency otherwise.

  • HBase/BigTable-style and many consensus systems are PC/EC: favor consistency in both modes.

  • Some systems are tunable (PA/EC), choosing per-operation.

A Phi-Accrual failure detector outputs a continuous suspicion value (phi) representing how likely a node has failed, instead of a binary alive/dead verdict. It learns the distribution of past heartbeat arrival times and adapts its threshold to real network conditions, unlike a fixed heartbeat timeout.

• Simple heartbeat timeout:

  • Declares a node dead if no beat arrives within a fixed timeout: a single hard threshold.

  • Too short causes false positives on a slow network; too long delays real failure detection. Hard to tune once for all conditions.

• Phi-accrual approach:

  • Records inter-arrival times of heartbeats and models them (often as a normal distribution).

  • Computes phi ~ the log-probability that a beat this late is still going to arrive; phi rises the longer silence lasts.

  • Applications pick a phi threshold that maps to a tolerated false-positive rate, and different components can use different thresholds.

• Why it's better:

  • Adapts automatically to latency jitter and load instead of one static timeout.

  • Gives a tunable confidence knob rather than a brittle yes/no. Used in Cassandra and Akka.

A lease is a time-bounded grant of a right (e.g. to be leader, hold a lock, or own a resource) that automatically expires unless renewed. It improves on plain heartbeats by tying liveness to a clock deadline, so the holder knows when its authority lapses even if it can't reach anyone.

• Heartbeat problem: Missing heartbeats only tells the observer the peer might be dead; it's ambiguous under network delays and gives no bound the holder itself respects.

• A lease adds an expiry:

  • The grantor issues the lease for a duration; the holder must renew before it expires or lose the right.

  • Both sides reason about the same deadline, so authority is self-limiting.

• Safety benefit: If a leader gets partitioned, it stops acting once its lease lapses, preventing two active leaders (split-brain).

• Clock caveat: Leases assume bounded clock drift; use a safety margin so the grantor treats the lease as expired slightly after the holder does.

A cascading failure is when the failure of one component overloads others, which then fail too, spreading until the whole system collapses. It's typically driven by load redistribution and retries: work that would have gone to a dead node piles onto survivors, tipping them over in turn.

• Typical trigger:

  • One node dies, its traffic shifts to peers already near capacity, they slow or crash, and the wave continues.

  • Aggressive client retries amplify load exactly when the system is weakest.

• Load shedding:

  • Deliberately reject a fraction of requests early (return 429/503) so accepted work completes instead of everything timing out.

  • Prioritize by request importance when shedding.

• Other containment techniques:

  • Circuit breakers cut off calls to failing services to stop retry storms.

  • Rate limiting and backpressure bound incoming work to what the system can handle.

  • Exponential backoff with jitter on retries avoids synchronized retry spikes.

  • Autoscaling and headroom give capacity to absorb redistributed load.

Omission and timing failures are intermediate fault models between the benign crash-stop model and the malicious Byzantine model. A process is neither fully dead nor arbitrarily lying: it simply drops messages (omission) or responds too late or too early (timing).

• Crash (fail-stop) is the mildest: A process works correctly until it halts and then does nothing; simplest to reason about.

• Omission failures: A process or channel fails to send or receive some messages (dropped packets, overflowing buffers) but otherwise behaves correctly.

• Timing failures: Correct value but delivered outside its expected time bound (too slow, or a clock drifting); only meaningful in synchronous systems with timing assumptions.

• Byzantine is the most severe: Arbitrary behavior: sending wrong, inconsistent, or deliberately deceptive messages.

• Ordering: Crash ⊂ omission ⊂ timing ⊂ Byzantine in severity: each stronger model subsumes the failures of the weaker one, and tolerating it costs more.

A hotspot (hot key) is a single key or narrow key range that receives a disproportionate share of traffic, overloading one partition even though total cluster load looks fine. You detect it with per-key/per-partition metrics and mitigate by spreading or absorbing the load.

• Why it happens: Skewed access (a celebrity user, a trending item) or sequential keys concentrated on one shard.

• Detection: Track per-partition QPS/latency and top-K key sampling; a single partition saturating while others idle signals a hot key.

• Mitigation:

  • Key salting: append a random/bounded prefix to split one hot key across N sub-keys, then scatter-read to recombine.

  • Caching: put a read cache in front so most reads never reach the shard.

  • Read replicas: fan hot reads across multiple replicas of the partition.

  • Dedicated/split partition: isolate the hot key onto its own shard.

• Tradeoff: salting fixes writes but complicates reads (must query all salts), so apply it selectively to known-hot keys.

Chain replication arranges replicas in an ordered chain: writes enter at the head, propagate node-by-node to the tail, and reads are served by the tail. This gives strong consistency with high read throughput, at the cost of higher write latency and slower failure recovery than leader-based replication.

• How it works:

  • Head accepts writes, forwards down the chain; a write commits only when the tail acknowledges.

  • Tail answers all reads, so any served read reflects a fully-replicated (committed) value: linearizable.

• Advantages:

  • Strong consistency with simple read logic (single point of truth at the tail).

  • Read and write roles are split across distinct nodes, balancing load.

• Tradeoffs vs leader-based:

  • Write latency grows with chain length (sequential hops) vs a leader replicating in parallel.

  • Failure handling is more intricate: head, middle, and tail failures each need distinct reconfiguration, usually via an external coordinator.

  • A single slow node in the chain stalls all writes.

• Variants like CRAQ let intermediate nodes serve reads to boost read throughput while preserving consistency.

Anti-entropy is the background process by which replicas compare and reconcile their data so they eventually converge, complementing the imperfect propagation of individual writes. Read-repair fixes divergence lazily on the read path, while Merkle trees make periodic full comparison cheap.

• Why it's needed: In leaderless systems, writes may miss replicas (down nodes, dropped messages), so replicas drift apart.

• Read-repair:

  • On a quorum read, the coordinator sees stale replicas and writes the freshest value back to them.

  • Cheap and immediate, but only repairs keys that are actually read (cold data stays stale).

• Merkle trees:

  • A hash tree over key ranges; two replicas compare root hashes, then descend only into subtrees that differ.

  • Lets replicas find divergent ranges with minimal data transfer, so background sync scales to large datasets.

• Together they cover both paths: read-repair for hot/read data, Merkle-based anti-entropy for everything else.

A sloppy quorum accepts writes on any W reachable healthy nodes (not necessarily the key's designated home nodes) when some homes are down, keeping the system available. Hinted handoff is the mechanism that stores such a write with a hint and later hands it back to the proper node once it recovers.

• Strict vs sloppy quorum:

  • A strict quorum requires W of the key's exact home replicas; if too many are down, the write fails.

  • A sloppy quorum lets other nodes temporarily accept the write so it still succeeds.

• Hinted handoff: The temporary node tags the data with a hint ("this belongs to node X") and forwards it once X is back.

• Availability benefit: Writes survive node/partition failures instead of being rejected, raising write availability.

• Cost: Weakens the quorum guarantee: a read from home nodes may miss a value still parked on a temporary node, so it's not truly linearizable during failures.