Distributed Systems

CAP states that a distributed system can provide at most two of: Consistency (reads return the latest write), Availability (every request gets a non-error response), and Partition tolerance (the system works despite network partitions). What CAP actually means: since network partitions are unavoidable in real systems, the real choice is what to do when a partition occurs: sacrifice consistency (AP — return possibly stale data) or sacrifice availability (CP — return an error until the partition heals). What CAP does not mean: - It's not a permanent choice — a system can be CP during partitions and highly available the rest of the time. Partitions are rare; the PACELC model better captures the latency-vs-consistency trade-off under normal operation. - "Consistency" in CAP means linearizability, not ACID consistency. These are different. - "Available" in CAP means every node responds — not that the system is always reachable by clients. A load balancer routing around a downed node provides availability from the client's perspective even if CAP's strict definition isn't met.

Practical examples: - CP: etcd, ZooKeeper — return an error rather than stale data during partition - AP: Cassandra (default), DynamoDB — continue serving reads/writes during partition, reconcile conflicts afterward - Most SQL databases are effectively CP for single-node deployments

Linearizability: every operation appears to execute atomically at a single point in real time. If a write completes at time T, all reads after T see the new value — globally, on any node. The strongest model; requires coordination. Causal consistency: operations related by causality are seen in causal order. "If you see my comment on a post, you see the post itself." Concurrent (independent) operations can be seen in different order by different nodes. No global coordination needed. Eventual consistency: replicas will converge given no new updates — but any individual read may return stale data. No ordering guarantees for concurrent operations. Bug caused by wrong consistency choice: Consider a social network using eventual consistency for profile reads. User Alice updates her privacy setting to "friends only" for her posts. Immediately after, she posts a private message. Due to replication lag: 1. The privacy setting update has not yet propagated to a read replica. 2. A request to that replica checks Alice's privacy setting — still sees "public." 3. Alice's private post is served to a stranger. The root cause: the privacy check and the post read must be causally ordered — if you read a post, you must see all privacy settings written before it. Causal consistency (or linearizability) would prevent this. Eventual consistency cannot.

Raft nodes are in one of three states: follower, candidate, or leader. Leader election: Followers wait for heartbeats from the leader. If the election timeout passes (randomized 150–300ms) without a heartbeat, the follower becomes a candidate, increments its term, votes for itself, and sends RequestVote RPCs to all other nodes. A candidate wins if it receives votes from a majority ((N/2)+1) of nodes. A node only votes for a candidate whose log is at least as up-to-date as its own (prevents electing a node with stale log). Randomized timeouts prevent split votes — nodes start elections at slightly different times. Log replication: The leader receives client writes, appends them to its log as a new entry, and sends AppendEntries RPCs to all followers. An entry is committed when the leader receives acknowledgment from a majority of nodes. The leader then applies the entry to its state machine and responds to the client. Future AppendEntries (including heartbeats) inform followers of committed entries, which they apply to their state machines. Safety guarantees: - At most one leader per term (election majority) - A committed entry is never lost (only a node with the complete committed log can win an election) - Entries are applied in log order (state machine safety) Raft is not Byzantine-fault-tolerant — it assumes nodes are honest but may crash.

2PC coordinates an atomic transaction across multiple participants (databases, services): Phase 1 — Prepare: the coordinator asks each participant to prepare (lock resources, write to WAL). Each participant responds with "yes" (ready to commit) or "no" (abort). Phase 2 — Commit or Abort: if all participants voted yes, the coordinator sends "commit"; otherwise sends "abort." Participants execute the decision. Failure modes: - Coordinator crashes after prepare, before commit: participants are in the "prepared" state — locks held, transaction neither committed nor aborted. They're blocked until the coordinator recovers. This is the blocking problem — 2PC is a blocking protocol. - Participant crashes after prepare: when it recovers, it reads its WAL and asks the coordinator for the decision. If the coordinator also crashed, nobody knows the decision. - Network partition between coordinator and participant: same blocking scenario. Why microservices avoid 2PC: - It requires participants to hold locks across a network round-trip — high latency window where failures cause indefinite blocking - Tightly couples all participating services — they must all be available for the transaction to complete - Does not compose well with HTTP APIs — 2PC requires a stateful protocol participants must support - Modern distributed systems prefer SAGA (compensating transactions) or outbox pattern for atomicity

A SAGA decomposes a long-lived transaction into a sequence of local transactions, each atomic within one service. If a step fails, previously completed steps are compensated (undone) with dedicated compensating transactions — not rolled back (no distributed lock). Choreography: each service publishes events and reacts to events from other services. No central coordinator. Loosely coupled — services don't know about each other's internals. Hard to understand the overall flow; debugging requires tracing events across services. Good for simple, linear flows with few participants. Orchestration: a central saga orchestrator (a dedicated service or workflow engine like Temporal, AWS Step Functions) drives the sequence by calling services in order and deciding what to do on failure. Easier to understand flow; the orchestrator is a potential bottleneck and single point of failure (mitigated by making it stateful and replicated). Good for complex flows with many branches and error paths. What makes compensations tricky: - Compensations must be idempotent — they may be called multiple times if the saga retries. - Compensations cannot always fully undo — a sent email, a charged card, or a shipped package cannot be "rolled back." Compensations must handle semantic undos (refund, cancel notification). - Pivot transaction: the step after which it's too late to cancel (e.g., payment captured). Design the saga so the pivot is as late as possible. - Concurrent modifications: if another saga or user modifies the data between the original step and its compensation, the compensation may conflict. Use versioning or optimistic locking.

The dual-write problem: a service needs to write to its database AND publish an event to a message broker atomically. These are two separate systems with no distributed transaction support. Three failure scenarios: 1. DB write succeeds, broker publish fails → state change not propagated → downstream services miss the event. 2. Broker publish succeeds, DB write fails → downstream processes an event for a change that didn't happen. 3. Process crashes between DB write and broker publish → same as scenario 1. Outbox pattern solves this by making both writes local: 1. Within a single local database transaction, write the business record AND an outbox table row (the event payload). 2. A separate relay process (or CDC via Debezium) reads unprocessed outbox rows and publishes them to the broker. 3. Mark rows processed (or delete them) after successful publish. The relay publishes at-least-once — retries on failure mean the consumer must be idempotent. The database transaction is the unit of atomicity — either both the business record and the outbox row are written, or neither is. No distributed transaction needed.

The problem: wall clocks on different machines can't be trusted for event ordering. Even with NTP, clocks differ by milliseconds, and NTP corrections can jump clocks backward. Lamport timestamps: each node maintains a counter. On any event, increment the counter. On sending a message, include the counter. On receiving a message, set counter = max(local, received) + 1. Lamport timestamps guarantee: if event A happened before event B (causally), then timestamp(A) < timestamp(B). But the converse is not true — timestamp(A) < timestamp(B) does not mean A happened before B. They establish a partial order. Vector clocks: each node maintains a vector of counters, one per node ([N1: 3, N2: 1, N3: 2]). On event: increment own entry. On send: include vector. On receive: take element-wise max, then increment own entry. Two events are concurrent if neither vector dominates the other. This detects causality precisely — you can tell if two events are causally related or truly concurrent (and thus may conflict). Used by Dynamo, Riak for conflict detection. When you need them: - Detecting write conflicts in leaderless replication (were these two writes causally related or concurrent? If concurrent, they conflict and need resolution) - Event ordering in distributed logs where wall-clock ordering is unreliable - Debugging distributed systems (causal trace of which event caused which)

An idempotent API returns the same result for the same logical request, regardless of how many times it's called. For payments, the client generates a unique idempotency key per payment attempt and includes it in every retry. Server-side implementation: sql CREATE TABLE idempotency_keys ( key VARCHAR PRIMARY KEY, -- unique constraint is the atomic check response JSONB NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW() ); On receipt of a request with key K: 1. INSERT INTO idempotency_keys (key, response) VALUES (K, NULL) ON CONFLICT DO NOTHING returns whether this is a new or duplicate request. 2. If new: process the payment, then update response with the result. 3. If duplicate: return the stored response immediately. What makes idempotency hard: - Race condition: two identical requests arrive simultaneously. Both check "is this key new?" before either inserts. Both see "new" and both process the payment. The unique constraint prevents this — only one INSERT succeeds; the other gets a conflict and waits. - Partial failure: request is processed, payment charged, but the response is lost before being stored. On retry, the key insert succeeds (new key), payment is charged again. Fix: write the key row before charging (response=NULL), then update after. If the process crashes mid-charge, the key row exists with response=NULL — a sentinel that indicates an in-progress or failed payment requiring manual review. - Idempotency key scope: a key reused across different operations (different amounts, different merchants) must be rejected. Validate that the key's associated request matches the current request parameters.

A single-instance rate limiter (token bucket or fixed window in memory) doesn't work when the service scales horizontally — each instance has its own counter, so the effective rate limit is limit × instance_count. Redis-based token bucket: Use a Redis key per (user_id, window) with atomic Lua script to check and decrement the counter in a single round-trip. Redis's single-threaded command execution makes the check-and-decrement atomic without a distributed lock. lua local current = redis.call('GET', KEYS[1]) if current and tonumber(current) >= tonumber(ARGV[1]) then return 0 -- rate limit exceeded end redis.call('INCR', KEYS[1]) redis.call('EXPIRE', KEYS[1], ARGV[2]) return 1 -- allowed Sliding window log: store timestamps of each request in a sorted set. Count entries in the window [now - window, now]. More accurate than fixed windows but higher memory usage. Trade-offs: - Redis becomes a synchronous dependency on every request — it must be highly available and low-latency (<1ms). Use Redis Cluster for HA. - Network round-trip to Redis adds latency. For latency-critical paths, use a local approximate counter (token bucket with periodic Redis sync) — accepts slight over-limit for reduced latency. - Redis failure: fail-open (allow all requests) or fail-closed (reject all). Fail-open is usually correct — a rate limiter outage shouldn't take down the service.

Sliding window with approximate local counting: each instance tracks a local counter and periodically reports to Redis. Redis aggregates. Local counter is the rate limit minus the reported counts from other instances. Allows requests without a Redis round-trip most of the time; syncs periodically. Trades exactness for performance.

Split-brain occurs when a network partition separates the cluster into groups, each of which elects its own leader. Both leaders accept writes — the cluster has two masters with diverging state. Reconciliation is expensive or impossible for some data types. Prevention — quorum-based election: a leader is only valid if it has a majority of votes. In a 3-node cluster partitioned 2+1, only the majority side (2) can elect a leader. The minority side (1) cannot reach quorum and stops accepting writes. This is the CP trade-off — the minority side becomes unavailable. Fencing tokens: each leadership term has a monotonically increasing token. Every write to the storage layer includes the current token. The storage layer rejects writes with an outdated token. If a partitioned old leader recovers and tries to write with token 5 while the new leader has token 6, the writes are rejected. The old leader discovers it's no longer the leader via rejection. Lease-based leadership: the leader holds a time-limited lease. It cannot accept writes after the lease expires. Even if the leader can't communicate with the cluster (partition), it stops writing when the lease expires, giving the cluster time to elect a new leader whose lease starts after the old one expired. etcd's leader leases work this way. Recovery from split-brain (if it occurs despite prevention): compare the two diverged logs. The Raft algorithm's log comparison rejects candidates without the complete committed log — in practice, only the majority side has committed entries, and the minority side's uncommitted writes are discarded on rejoin.

The problem with naive sharding: if you assign keys to nodes by hash(key) % N, adding or removing a node changes N, which remaps almost every key — requires massive data redistribution. Consistent hashing: arrange a "ring" of hash values from 0 to 2³²-1. Place each node at one or more points on the ring (by hashing the node's ID). To find which node owns a key, hash the key and walk the ring clockwise to the first node. Adding a node places it on the ring — only the keys between it and its predecessor move to the new node (~1/N of total keys). Removing a node moves only that node's keys to its successor. Virtual nodes (vnodes): instead of placing each physical node once on the ring, place it at V points (vnodes, typically 128–256 per node). This: - Evenly distributes load even with heterogeneous or few nodes - Makes rebalancing more granular — a new node picks up vnodes from many existing nodes, distributing the transfer load - Allows different nodes to hold different numbers of vnodes (weighted by capacity) Replication with consistent hashing: a key is assigned to the N clockwise successor nodes. For Cassandra with replication_factor = 3, each key lives on 3 consecutive nodes on the ring. If one node fails, reads and writes fall to the remaining 2 (or 1, depending on quorum settings).

This is a multi-constraint problem: low read latency (local reads), strong consistency (no stale financial data), and high availability (no single region failure takes you down). These constraints are partially in tension — strong consistency requires cross-region coordination which adds latency. Tiered consistency architecture: Tier 1 — Strongly consistent (financial data: balances, transactions): - Use a globally distributed SQL database (Google Spanner or CockroachDB) with external consistency via TrueTime / HLC (Hybrid Logical Clocks). - Writes go to a global leader (or are multi-leader with serializable isolation). Latency is bounded by the speed of light + algorithm overhead (~30–100ms cross-region). - Accept that writes are slower; financial writes are lower frequency than reads. - For reads: use stale reads (bounded staleness) for user-facing balance displays — returning data up to 10 seconds stale is acceptable for a balance page. Use linearizable reads only for the atomic check-then-debit operations.

Tier 2 — Causally consistent (user profile, preferences, non-financial state): - Replicate to all regions with causal ordering. Users always see their own writes. - Read from local region; writes propagate asynchronously. Conflicts resolved by last-writer-wins or CRDTs (counters, sets).

Tier 3 — Eventual consistency (feeds, analytics, leaderboards): - Replicate asynchronously. Seconds of staleness acceptable. Read from local region. Multi-region HA: - Active-active for AP tiers; writes go to the nearest region. - Active-active for the CP tier is possible with multi-region Paxos groups (Spanner's model), with the latency cost paid on each write. - Circuit breakers between regions: if region A can't reach region B's consensus group, fail writes rather than silently accept stale reads that violate consistency guarantees.

Chaos engineering is the practice of deliberately injecting failures to discover system weaknesses before they manifest as incidents. Structured approach: 1. Define the steady state: what does "healthy" look like? Success rate, p99 latency, error rate, key business metric (orders per minute). Chaos experiments verify that the steady state holds under failure, not that services stay running. 2. Start with known failure modes: before random chaos, exhaustively test expected failures: single-node crashes, single-AZ failure, network latency injection, dependency timeouts, disk full. These should be in your runbooks — verify the runbooks are correct. 3. Hypothesis-driven experiments: "If Kafka broker 2 goes down, consumer lag should not exceed 30 seconds and no messages should be lost." Run the experiment, measure the outcome, compare against the hypothesis. A passing experiment builds confidence; a failing one reveals a weakness. 4. Increase scope gradually: start in dev/staging, then production during low-traffic periods, then during peak traffic. GameDay events involve the full team practicing response. Commonly missed failure modes: - Slow dependencies (not down, just slow): a 2-second response from a downstream is worse than a crash — threads accumulate waiting, cascading to the caller. Test with latency injection, not just kill injection. - Partial availability: one instance of a downstream is healthy; others are not. Load balancers may route some requests to the bad instance — circuit breakers must act per-instance, not per-service. - Clock skew: inject clock drift and verify that Lamport/vector clock logic and lease-based systems still function correctly. - Resource exhaustion: file descriptors, thread pool, connection pool — not just CPU and memory. - Data corruption at the boundary: malformed messages in Kafka, corrupted DB rows. Does your consumer handle them? Or does one bad message halt all processing? - Thundering herd after recovery: a dependency comes back online — do all instances simultaneously flood it? Test restart behavior, not just failure behavior.