Guarantees:Fault tolerance via replication + checksums. High throughput for sequential reads. GFS evolved into Colossus. HDFS is foundation of Hadoop ecosystem.
Guarantee:No SPOF — at least 2 of everything. Netflix Chaos Monkey randomly kills production instances.
More Patterns:Leader Election — Raft: candidates request votes, majority wins, leader sends heartbeats. If heartbeat missed → new election. Used by etcd, CockroachDB, Consul. Chaos Engineering — intentionally inject failures in production to find weaknesses. Chaos Monkey (Netflix), Litmus (K8s), Gremlin. Test: kill pods, add latency, partition network, fill disk. Distributed Tracing Internals — inject trace_id + span_id into every request header. Each service creates a child span. Collect via OpenTelemetry SDK → Collector → Jaeger/Datadog. Enables: latency breakdown per service, dependency mapping.
Data Redundancy & Recovery
Survive hardware failures, data corruption, and disasters without losing data
Strategy
How
RPO / RTO
Use Case
Replication
Sync/async copies across nodes
RPO: 0 (sync) or seconds (async). RTO: seconds (auto-failover)
HA databases (Postgres replicas, Cassandra ring)
Erasure Coding
Split data into fragments + parity. Reconstruct from subset.
RPO: 0. RTO: minutes (rebuild)
Object storage (S3 uses Reed-Solomon), HDFS
Snapshots
Point-in-time copy of entire dataset
RPO: hours (last snapshot). RTO: minutes–hours
DB backups (RDS snapshots), VM snapshots
WAL / Binlog
Continuous log of all changes. Replay to recover.
RPO: ~0 (continuous). RTO: minutes (replay)
Postgres PITR, MySQL binlog replication
Geo-Redundancy
Replicate across regions (100+ km apart)
RPO: seconds (async). RTO: minutes (DNS failover)
Disaster recovery — survive entire region failure
Key Terms:RPO (Recovery Point Objective) — max data loss tolerable (how far back?). RTO (Recovery Time Objective) — max downtime tolerable (how fast to recover?). 3-2-1 Rule — 3 copies, 2 different media, 1 offsite. Hot standby = instant failover. Warm = minutes. Cold = hours (cheapest).
Leader Election for Auto-Recovery
When the leader dies, the cluster must automatically elect a new one — no human intervention
Algorithm
How It Works
Used By
Raft
Nodes start as followers. On leader timeout → become candidate → request votes from majority → win = new leader → send heartbeats. Understandable by design.
etcd, CockroachDB, Consul, TiKV
ZAB
ZooKeeper Atomic Broadcast. Leader proposes, followers ACK. On failure → new leader elected from most up-to-date follower.
ZooKeeper, Kafka (pre-KRaft)
KRaft
Kafka's built-in Raft (replaces ZooKeeper). Controller quorum elects active controller. Simpler ops.
Kafka 3.3+
Bully
Highest-ID node wins. Simple but not partition-tolerant. Rarely used in production.
Legacy systems
Auto-Recovery Flow: Leader dies → followers detect missing heartbeat (timeout: 150-300ms) → election starts → new leader elected (majority quorum) → new leader takes over writes → clients redirected → total downtime: 1-5 seconds. No human needed. Split-brain prevention: quorum ensures only one leader (majority of N/2+1 must agree).
Real-world:etcd (K8s control plane) — Raft, 5-node cluster, leader election in ~1s. Kafka KRaft — controller quorum, no ZooKeeper dependency. Redis Sentinel — monitors master, promotes replica on failure (not consensus-based, simpler). Postgres — Patroni + etcd for automated failover.
Strong leader required; all writes go through leader
No inherent leader (Multi-Paxos adds one)
Log
Contiguous log, no gaps allowed
Gaps possible, out-of-order commits
Membership Change
Joint consensus (built-in)
Not specified in original paper
Implementation
~2000 lines (etcd/raft)
Notoriously difficult to implement correctly
Liveness
Guaranteed with leader (leader liveness assumed)
Can livelock with competing proposers
Performance
1 RTT for committed writes (leader → followers)
2 RTTs (Prepare + Accept) unless Multi-Paxos
Used By
etcd, CockroachDB, Consul, TiKV, RethinkDB
Chubby, Spanner, Megastore
Consensus Protocols
Agreement among distributed nodes — the hardest problem in distributed systems. FLP impossibility: no deterministic consensus in async system with even 1 crash.
▸ Classic Paxos — 4 Phases (5-Node Cluster)
Multi-Paxos Optimization: Classic Paxos requires 2 RTTs per decision (Prepare + Accept). Multi-Paxos elects a stable leader who skips Phase 1 for subsequent proposals — reducing to 1 RTT (just Accept). Leader holds a "lease" on proposal numbers. If leader fails, new leader runs full Paxos once, then resumes fast path. Used by: Google Spanner, Chubby.
Raft Log Replication Detail (AppendEntries RPC): Leader maintains nextIndex[] and matchIndex[] per follower. On client write: (1) Leader appends to local log. (2) Sends AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries[], leaderCommit) to all followers. (3) Follower checks prevLogIndex/prevLogTerm match — if not, rejects (leader decrements nextIndex and retries). (4) On majority ACK → leader advances commitIndex. (5) Next heartbeat carries updated leaderCommit → followers apply to state machine.
▸ Consensus Protocol Comparison
Protocol
Leader Required?
RTTs (steady state)
Log Ordering
Membership Change
Used By
Raft
Yes (strong leader)
1 RTT
Strict sequential, no gaps
Joint consensus (built-in)
etcd, CockroachDB, Consul
Multi-Paxos
Yes (elected proposer)
1 RTT (after leader elected)
Gaps possible, out-of-order
Separate protocol needed
Spanner, Chubby, Megastore
ZAB
Yes (primary)
1 RTT
FIFO + causal ordering
Reconfiguration protocol
ZooKeeper
Viewstamped Replication
Yes (primary)
1 RTT
Sequential within view
View change protocol
Academic, influenced Raft
EPaxos
No (leaderless)
1 RTT (fast path, no conflicts)
Dependency graph
Complex
Research, some Google systems
Key Insight: All practical consensus protocols converge on a stable leader pattern for performance. The difference is in how they handle leader failure, log gaps, and membership changes. Raft wins on understandability; Multi-Paxos wins on flexibility; ZAB wins on ordering guarantees.
Clocks & Time in Distributed Systems
No global clock exists — ordering events across machines is fundamentally hard. Three approaches: physical, logical, hybrid.
▸ Physical vs Logical vs Hybrid Logical Clocks
▸ Lamport Timestamps — Example
Node A: event(1) → send(2) ─────────────────→ event(5)
Node B: event(1) → event(2) → recv(3) → event(4) → send(5)
Node C: event(1) ──────────────────────→ recv(6) → event(7)
Rules: On local event: C = C + 1
On send: attach timestamp C
On receive: C = max(local_C, msg_C) + 1
Limitation: L(a) < L(b) does NOT imply a happened-before b
(could be concurrent events with coincidental ordering)
▸ Vector Clocks — Detecting Causality
Node A: [1,0,0] → [2,0,0] → send([2,0,0]) ──────→ [4,0,0]
Node B: [0,1,0] → recv → [2,2,0] → [2,3,0] → send([2,3,0])
Node C: [0,0,1] ────────────────────→ recv → [2,3,2] → [2,3,3]
Causality detection:
[2,0,0] → [2,2,0] (A caused B's event: all A's slots ≤ B's)
[2,3,0] vs [4,0,0] CONCURRENT (neither dominates the other)
Conflict resolution on concurrent writes:
• Application-level merge (CRDTs)
• Last-Writer-Wins (LWW) — use physical timestamp as tiebreaker
• Return both versions to client (Riak "siblings")
NTP Limitations: Network Time Protocol syncs clocks to ~1-10ms on LAN, ~100ms on WAN. Problems: (1) Clock skew — different rates of drift between machines. (2) Clock jumps — NTP can step clock backward. (3) Monotonic vs wall-clock — use monotonic for measuring durations, wall-clock only for human display. (4) Leap seconds — can cause 1s jump. Solution: leap smear (Google spreads over 24h).
Practical Guidance: Use Lamport timestamps when you only need total ordering (Kafka offsets). Use Vector clocks when you need to detect conflicts (DynamoDB, Riak). Use HLC when you need causality + wall-clock proximity (CockroachDB MVCC). Use TrueTime if you have GPS/atomic clocks (Spanner).
Replication Strategies
Copying data across nodes for fault tolerance and read scalability. The fundamental tradeoff: consistency vs availability vs latency.
▸ Replication Topologies Compared
▸ Synchronous vs Asynchronous Replication
Aspect
Synchronous
Asynchronous
Semi-Synchronous
Write Latency
High (wait for all/majority replicas)
Low (return after local write)
Medium (wait for 1 replica)
Durability
Strong (data on multiple nodes)
Weak (data loss on leader crash)
Good (1 replica guaranteed)
Availability
Lower (blocked if replica down)
Higher (leader independent)
Balanced
Consistency
Strong (all replicas up-to-date)
Eventual (replication lag)
Bounded staleness
Used By
Raft commit, Spanner
Postgres streaming, MySQL
MySQL semi-sync, Kafka (acks=1)
▸ Conflict Resolution Strategies
Strategy
How
Pros
Cons
Used By
Last-Writer-Wins (LWW)
Highest timestamp wins, discard others
Simple, deterministic
Data loss — concurrent writes silently dropped
Cassandra, DynamoDB
Vector Clocks + Siblings
Detect conflicts, return all versions to client
No data loss
Client must resolve; complex
Riak, DynamoDB (optional)
CRDTs
Data structures that auto-merge (counters, sets, registers)
Automatic, no coordination
Limited data types; space overhead
Redis CRDT, Riak, Figma
Operational Transform
Transform concurrent ops to preserve intent
Works for text editing
Complex; centralized server often needed
Google Docs
Application-level
Custom merge logic (e.g., union of shopping carts)
Domain-specific correctness
Developer burden
Amazon shopping cart
▸ When to Use Which Replication Strategy
Scenario
Recommended
Why
Single-region, strong consistency needed
Single-leader, sync
Simplest model, no conflicts, Raft/Paxos for HA
Multi-region, low-latency writes
Multi-leader
Write to local DC, async replicate; accept conflict resolution cost
High availability, tunable consistency
Leaderless (quorum)
No single point of failure; tune W/R/N per use case
Read-heavy, eventual consistency OK
Single-leader, async replicas
Scale reads with followers; accept replication lag
Collaborative editing
CRDTs or OT
Automatic conflict-free merging for concurrent edits
Replication Lag: In async replication, followers can be seconds behind leader. This causes: read-after-write inconsistency (user writes, reads from stale follower), monotonic read violations (time goes backward), causality violations (see reply before original message). Solutions: read-your-writes guarantee (route to leader after write), causal consistency (track dependencies).
Partitioning & Sharding
Split data across nodes to scale beyond a single machine. Key challenge: even distribution + efficient queries.
▸ Hash Partitioning vs Range Partitioning
▸ Consistent Hashing with Virtual Nodes
▸ Rebalancing Strategies
Strategy
How
Pros
Cons
Used By
Fixed Partitions
Create many more partitions than nodes (e.g., 1000 partitions, 10 nodes). Assign partitions to nodes.
Simple rebalancing (move whole partitions)
Must choose partition count upfront; too few = large partitions, too many = overhead
Riak, Elasticsearch, Couchbase
Dynamic Splitting
Start with 1 partition per node. Split when too large, merge when too small.
Adapts to data size; no upfront decision
Complex; split/merge coordination
HBase, RethinkDB, MongoDB
Proportional to Nodes
Fixed number of partitions per node. New node steals random partitions from existing nodes.
Balanced as cluster grows
Random splits can create uneven sizes
Cassandra (vnodes)
▸ Cross-Partition Queries & Scatter-Gather
Scatter-Gather Pattern: Query that spans multiple partitions: (1) Scatter — coordinator sends query to all relevant partitions in parallel. (2) Gather — coordinator collects results, merges, sorts, returns to client. Problem: tail latency — response time = slowest partition. Mitigation: hedged requests (send to multiple replicas, use first response).
Secondary Indexes on Partitioned Data: Two approaches: (1) Local index (document-partitioned) — each partition maintains its own index. Writes are fast (single partition), reads require scatter-gather. (2) Global index (term-partitioned) — index itself is partitioned by term. Reads hit single partition, writes must update multiple index partitions (async, eventually consistent). Used by: DynamoDB (local), Elasticsearch (global).
Detecting node failures in an asynchronous network is impossible to do perfectly (FLP). All detectors trade off detection speed vs false positive rate.
▸ Heartbeat-Based Failure Detection Timeline
▸ Phi Accrual Failure Detector
Problem with fixed timeouts: Network latency varies. A fixed timeout (e.g., 5s) either: (1) too short → false positives (healthy node marked dead during GC pause), or (2) too long → slow detection (actual failures take too long to detect).
Phi (φ) Accrual Detector: Instead of binary alive/dead, outputs a suspicion level φ (continuous value). Higher φ = more likely dead. Application chooses threshold (e.g., φ > 8 = dead). How it works: (1) Track inter-arrival times of heartbeats. (2) Compute mean and variance of arrival distribution. (3) φ = -log₁₀(P(next heartbeat hasn't arrived yet)). (4) φ grows exponentially with time since last heartbeat. Adaptive: automatically adjusts to network conditions — works on both LAN (low latency) and WAN (high latency) without tuning.
// Phi Accrual — Conceptual
φ(t_now) = -log₁₀(1 - F(t_now - t_last))
where F = CDF of normal distribution fitted to heartbeat intervals
Example: mean_interval = 1000ms, std_dev = 200ms
t_now - t_last = 1000ms → φ ≈ 0.3 (probably alive)
t_now - t_last = 2000ms → φ ≈ 3.0 (suspicious)
t_now - t_last = 5000ms → φ ≈ 12.0 (almost certainly dead)
Threshold: φ > 8 → mark as dead (Cassandra default)
φ > 12 → very conservative (fewer false positives)
▸ SWIM Protocol — Gossip-Based Failure Detection
SWIM (Scalable Weakly-consistent Infection-style Membership): Combines failure detection with membership dissemination via gossip. Steps: (1) Node A picks random node B, sends ping. (2) If B responds → alive. (3) If B doesn't respond within timeout → A picks K random nodes, asks them to ping-req(B) (indirect probe). (4) If any indirect probe succeeds → B is alive (was just network issue between A and B). (5) If all fail → B marked as suspected. (6) After suspicion timeout → B marked dead. (7) Membership changes piggybacked on ping/ack messages (gossip dissemination).
SWIM Properties:O(1) per-node load (each node probes 1 random peer per interval). O(log N) detection time (gossip convergence). False positive rate tunable via K (indirect probes) and suspicion timeout. Used by: HashiCorp Memberlist (Consul, Nomad, Serf), Uber Ringpop.
▸ Detection Time vs False Positive Tradeoff
Detector
Detection Time
False Positive Rate
Scalability
Adaptivity
Used By
Fixed Timeout
timeout value (e.g., 5s)
High if timeout too short
O(N) messages from monitor
None — manual tuning
Simple systems, Redis Sentinel
Heartbeat + Lease
lease_duration (e.g., 10s)
Low (conservative timeout)
O(N) per monitor
None
ZooKeeper sessions, Chubby
Phi Accrual
Adaptive (based on history)
Tunable via φ threshold
O(N) per monitor
Self-adjusting
Cassandra, Akka Cluster
SWIM (Gossip)
O(log N) protocol periods
Low (indirect probes reduce)
O(1) per node
Moderate
Consul, Serf, Memberlist
SWIM + Suspicion
O(log N) + suspicion timeout
Very low
O(1) per node
Good
Consul (Lifeguard)
Design Guidance: For small clusters (3-7 nodes): simple heartbeat with Raft-style election timeout (150-300ms) works well. For medium clusters (10-100 nodes): Phi Accrual gives best accuracy. For large clusters (100+ nodes): SWIM/gossip is essential — O(1) per-node overhead vs O(N) for centralized heartbeat. Always add suspicion period before declaring dead — reduces false positives from transient network issues or GC pauses.