Distributed Systems - Dev Weekends

What are Distributed Systems?

A distributed system is a collection of independent computers that appear to users as a single coherent system. Leslie Lamport put it best: “A distributed system is one in which the failure of a computer you didn’t even know existed can render your own computer unusable.” That quote captures the essential challenge — once your system spans multiple machines, you inherit a whole category of problems (network partitions, clock skew, partial failures) that simply do not exist on a single machine. The analogy that works best: imagine a team of chefs in separate kitchens, communicating only by passing notes through a mail slot. They need to coordinate on what dishes to prepare, in what order, using shared ingredients — but notes can be delayed, lost, or arrive out of order. That is distributed systems in a nutshell. Everything that follows in this chapter is about managing that coordination challenge.

Why Distributed?

Scalability

Handle more load than a single machine

Reliability

Survive individual machine failures

Latency

Serve users from nearby locations

Compliance

Data locality requirements

The Eight Fallacies

These fallacies, originally articulated by Peter Deutsch and James Gosling at Sun Microsystems, are the assumptions that bite engineers hardest when they move from single-machine to distributed systems. In interviews, casually referencing one or two of these when discussing failure modes signals deep understanding.

The network is reliable — Packets get lost. AWS reports measurable packet loss even within a single availability zone. Design for retries and idempotency from day one.
Latency is zero — Cross-region calls take 100-300ms. A design that makes 10 sequential cross-service calls adds a full second of latency. Batch and parallelize.
Bandwidth is infinite — A chatty microservice that transfers 1MB payloads at 10K QPS needs 10 GB/s of bandwidth and costs real money in cloud egress fees.
The network is secure — Always encrypt in transit (mTLS between services). Internal networks are not safe; lateral movement is how most breaches escalate.
Topology doesn’t change — Containers and VMs spin up and down constantly. Hard-coded IP addresses will break. Use service discovery.
There is one administrator — In a cloud/microservices world, dozens of teams own different parts of the infrastructure with different policies.
Transport cost is zero — AWS charges $0.01-0.09/GB for cross-AZ and cross-region data transfer. At scale, this becomes a major line item.
The network is homogeneous — Your services run on different hardware, different OS versions, different network cards with different MTU sizes.

Consistency Models

Strong Consistency

All nodes see the same data at the same time. This is the most intuitive model — it behaves as if there is only one copy of the data. The price you pay is latency (every write must be acknowledged by replicas before returning) and availability (if replicas are unreachable, writes block). Strong consistency is non-negotiable for use cases where “stale reads” cause real harm: bank balances, inventory counts, seat reservations.

Eventual Consistency

All nodes will eventually have the same data.

Consistency Implementation Examples

Python
JavaScript

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, List, Any, Dict
from datetime import datetime
import hashlib
import json
import asyncio

@dataclass
class VectorClock:
    """Track causality in distributed systems"""
    clocks: Dict[str, int] = field(default_factory=dict)
    
    def increment(self, node_id: str) -> None:
        self.clocks[node_id] = self.clocks.get(node_id, 0) + 1
    
    def merge(self, other: 'VectorClock') -> 'VectorClock':
        """Merge two vector clocks (take max of each)"""
        merged = VectorClock()
        all_nodes = set(self.clocks.keys()) | set(other.clocks.keys())
        for node in all_nodes:
            merged.clocks[node] = max(
                self.clocks.get(node, 0),
                other.clocks.get(node, 0)
            )
        return merged
    
    def is_concurrent_with(self, other: 'VectorClock') -> bool:
        """Check if two events are concurrent (neither happened before the other)"""
        self_greater = any(
            self.clocks.get(k, 0) > other.clocks.get(k, 0)
            for k in self.clocks
        )
        other_greater = any(
            other.clocks.get(k, 0) > self.clocks.get(k, 0)
            for k in other.clocks
        )
        return self_greater and other_greater


@dataclass
class ReplicatedValue:
    """Last-Write-Wins Register for eventual consistency"""
    value: Any
    timestamp: float
    node_id: str
    vector_clock: VectorClock
    
    def merge(self, other: 'ReplicatedValue') -> 'ReplicatedValue':
        """Merge with conflict resolution"""
        # Check for concurrent updates
        if self.vector_clock.is_concurrent_with(other.vector_clock):
            # Conflict! Use timestamp + node_id as tiebreaker
            if (self.timestamp, self.node_id) > (other.timestamp, other.node_id):
                winner = self
            else:
                winner = other
        elif self.timestamp > other.timestamp:
            winner = self
        else:
            winner = other
            
        return ReplicatedValue(
            value=winner.value,
            timestamp=winner.timestamp,
            node_id=winner.node_id,
            vector_clock=self.vector_clock.merge(other.vector_clock)
        )


class EventuallyConsistentStore:
    """Distributed key-value store with eventual consistency"""
    
    def __init__(self, node_id: str, replicas: List['EventuallyConsistentStore'] = None):
        self.node_id = node_id
        self.data: Dict[str, ReplicatedValue] = {}
        self.vector_clock = VectorClock()
        self.replicas = replicas or []
        self.pending_sync: List[tuple] = []
    
    def write(self, key: str, value: Any) -> None:
        """Write with vector clock tracking"""
        self.vector_clock.increment(self.node_id)
        
        replicated = ReplicatedValue(
            value=value,
            timestamp=datetime.now().timestamp(),
            node_id=self.node_id,
            vector_clock=VectorClock(clocks=self.vector_clock.clocks.copy())
        )
        
        # Local write
        if key in self.data:
            self.data[key] = self.data[key].merge(replicated)
        else:
            self.data[key] = replicated
        
        # Queue for async replication
        self.pending_sync.append((key, replicated))
    
    def read(self, key: str) -> Optional[Any]:
        """Read (may return stale data)"""
        if key in self.data:
            return self.data[key].value
        return None
    
    async def sync_to_replicas(self) -> None:
        """Background sync to other replicas"""
        while self.pending_sync:
            key, value = self.pending_sync.pop(0)
            for replica in self.replicas:
                try:
                    await replica.receive_sync(key, value)
                except Exception as e:
                    # Re-queue failed sync
                    self.pending_sync.append((key, value))
                    print(f"Sync failed to {replica.node_id}: {e}")
    
    async def receive_sync(self, key: str, incoming: ReplicatedValue) -> None:
        """Receive and merge update from another replica"""
        if key in self.data:
            self.data[key] = self.data[key].merge(incoming)
        else:
            self.data[key] = incoming
        
        # Update our vector clock
        self.vector_clock = self.vector_clock.merge(incoming.vector_clock)


# Usage example
async def demo_eventual_consistency():
    # Three replicas
    replica_a = EventuallyConsistentStore("node-a")
    replica_b = EventuallyConsistentStore("node-b")
    replica_c = EventuallyConsistentStore("node-c")
    
    # Cross-reference replicas
    replica_a.replicas = [replica_b, replica_c]
    replica_b.replicas = [replica_a, replica_c]
    replica_c.replicas = [replica_a, replica_b]
    
    # Concurrent writes (simulating network partition)
    replica_a.write("user:1:name", "Alice")
    replica_b.write("user:1:name", "Alicia")  # Conflict!
    
    # Before sync: different values
    print(f"A sees: {replica_a.read('user:1:name')}")  # Alice
    print(f"B sees: {replica_b.read('user:1:name')}")  # Alicia
    
    # After sync: converges to same value
    await replica_a.sync_to_replicas()
    await replica_b.sync_to_replicas()
    
    print(f"After sync - A: {replica_a.read('user:1:name')}")
    print(f"After sync - B: {replica_b.read('user:1:name')}")
    # Both will show the same value (winner based on timestamp + node_id)

// Vector Clock for causality tracking
class VectorClock {
  constructor(clocks = {}) {
    this.clocks = { ...clocks };
  }

  increment(nodeId) {
    this.clocks[nodeId] = (this.clocks[nodeId] || 0) + 1;
    return this;
  }

  merge(other) {
    const merged = new VectorClock();
    const allNodes = new Set([
      ...Object.keys(this.clocks),
      ...Object.keys(other.clocks)
    ]);
    
    for (const node of allNodes) {
      merged.clocks[node] = Math.max(
        this.clocks[node] || 0,
        other.clocks[node] || 0
      );
    }
    return merged;
  }

  isConcurrentWith(other) {
    const selfGreater = Object.keys(this.clocks).some(
      k => (this.clocks[k] || 0) > (other.clocks[k] || 0)
    );
    const otherGreater = Object.keys(other.clocks).some(
      k => (other.clocks[k] || 0) > (this.clocks[k] || 0)
    );
    return selfGreater && otherGreater;
  }

  clone() {
    return new VectorClock({ ...this.clocks });
  }
}

// CRDT: Last-Write-Wins Register
class LWWRegister {
  constructor(value, timestamp, nodeId, vectorClock) {
    this.value = value;
    this.timestamp = timestamp;
    this.nodeId = nodeId;
    this.vectorClock = vectorClock;
  }

  merge(other) {
    let winner;
    
    if (this.vectorClock.isConcurrentWith(other.vectorClock)) {
      // Conflict resolution: timestamp + nodeId tiebreaker
      const thisPriority = [this.timestamp, this.nodeId];
      const otherPriority = [other.timestamp, other.nodeId];
      
      if (thisPriority > otherPriority) {
        winner = this;
      } else {
        winner = other;
      }
    } else {
      winner = this.timestamp > other.timestamp ? this : other;
    }

    return new LWWRegister(
      winner.value,
      winner.timestamp,
      winner.nodeId,
      this.vectorClock.merge(other.vectorClock)
    );
  }
}

// Eventually Consistent Distributed Store
class EventuallyConsistentStore {
  constructor(nodeId) {
    this.nodeId = nodeId;
    this.data = new Map();
    this.vectorClock = new VectorClock();
    this.replicas = [];
    this.pendingSync = [];
    this.antiEntropyInterval = null;
  }

  write(key, value) {
    this.vectorClock.increment(this.nodeId);
    
    const register = new LWWRegister(
      value,
      Date.now(),
      this.nodeId,
      this.vectorClock.clone()
    );

    // Local write with merge
    if (this.data.has(key)) {
      this.data.set(key, this.data.get(key).merge(register));
    } else {
      this.data.set(key, register);
    }

    // Queue for replication
    this.pendingSync.push({ key, register });
    
    return this.data.get(key).value;
  }

  read(key) {
    const register = this.data.get(key);
    return register ? register.value : null;
  }

  // Read from multiple replicas for higher consistency
  async readWithQuorum(key, quorum = 2) {
    const responses = await Promise.all([
      this.read(key),
      ...this.replicas.slice(0, quorum - 1).map(
        replica => replica.read(key)
      )
    ]);

    // Return most common value (simple quorum)
    const counts = {};
    responses.forEach(val => {
      const str = JSON.stringify(val);
      counts[str] = (counts[str] || 0) + 1;
    });

    let maxCount = 0;
    let result = null;
    for (const [val, count] of Object.entries(counts)) {
      if (count > maxCount) {
        maxCount = count;
        result = JSON.parse(val);
      }
    }
    return result;
  }

  async receiveSync(key, incoming) {
    if (this.data.has(key)) {
      this.data.set(key, this.data.get(key).merge(incoming));
    } else {
      this.data.set(key, incoming);
    }
    this.vectorClock = this.vectorClock.merge(incoming.vectorClock);
  }

  // Anti-entropy: periodic full sync
  startAntiEntropy(intervalMs = 5000) {
    this.antiEntropyInterval = setInterval(async () => {
      for (const replica of this.replicas) {
        try {
          await this.syncWith(replica);
        } catch (err) {
          console.error(`Anti-entropy failed with ${replica.nodeId}:`, err);
        }
      }
    }, intervalMs);
  }

  async syncWith(replica) {
    // Exchange all keys
    for (const [key, register] of this.data.entries()) {
      await replica.receiveSync(key, register);
    }
    for (const [key, register] of replica.data.entries()) {
      await this.receiveSync(key, register);
    }
  }

  stopAntiEntropy() {
    if (this.antiEntropyInterval) {
      clearInterval(this.antiEntropyInterval);
    }
  }
}

// Demo
async function demo() {
  const nodeA = new EventuallyConsistentStore('node-a');
  const nodeB = new EventuallyConsistentStore('node-b');
  
  nodeA.replicas = [nodeB];
  nodeB.replicas = [nodeA];

  // Concurrent writes
  nodeA.write('counter', 10);
  nodeB.write('counter', 20); // Conflict!

  console.log('Before sync:');
  console.log('Node A:', nodeA.read('counter')); // 10
  console.log('Node B:', nodeB.read('counter')); // 20

  // Sync resolves conflict
  await nodeA.syncWith(nodeB);

  console.log('After sync:');
  console.log('Node A:', nodeA.read('counter')); // Same value
  console.log('Node B:', nodeB.read('counter')); // Same value
}

Consistency Levels

Level	Description	Trade-off
Strong	All reads see latest write	High latency
Eventual	Reads may be stale	Low latency
Causal	Respects cause-effect	Medium
Read-your-writes	See your own writes	Good UX
Session	Consistency within session	Practical

Consensus Algorithms

How do distributed nodes agree on a value?

Paxos (Simplified)

Raft (Easier to Understand)

Raft Implementation

Python
JavaScript

import asyncio
import random
from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from datetime import datetime

class NodeState(Enum):
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"

@dataclass
class LogEntry:
    term: int
    index: int
    command: str
    data: dict

@dataclass
class RaftNode:
    """Simplified Raft consensus node"""
    node_id: str
    state: NodeState = NodeState.FOLLOWER
    current_term: int = 0
    voted_for: Optional[str] = None
    log: List[LogEntry] = field(default_factory=list)
    commit_index: int = 0
    last_applied: int = 0
    
    # Leader state
    next_index: Dict[str, int] = field(default_factory=dict)
    match_index: Dict[str, int] = field(default_factory=dict)
    
    # Cluster info
    peers: List[str] = field(default_factory=list)
    
    # Timers
    election_timeout: float = 0
    heartbeat_interval: float = 0.15  # 150ms
    
    def __post_init__(self):
        self.reset_election_timeout()
    
    def reset_election_timeout(self):
        """Randomized election timeout (150-300ms typically)"""
        self.election_timeout = random.uniform(0.15, 0.30)
    
    async def start_election(self) -> bool:
        """Candidate starts election"""
        self.state = NodeState.CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id
        votes = 1  # Vote for self
        
        # Request votes from all peers
        vote_requests = []
        for peer_id in self.peers:
            vote_requests.append(
                self.request_vote(peer_id)
            )
        
        results = await asyncio.gather(*vote_requests, return_exceptions=True)
        votes += sum(1 for r in results if r is True)
        
        # Check if won election
        quorum = (len(self.peers) + 1) // 2 + 1
        if votes >= quorum:
            await self.become_leader()
            return True
        
        return False
    
    async def become_leader(self):
        """Transition to leader state"""
        self.state = NodeState.LEADER
        print(f"[{self.node_id}] Became leader for term {self.current_term}")
        
        # Initialize leader state
        for peer_id in self.peers:
            self.next_index[peer_id] = len(self.log) + 1
            self.match_index[peer_id] = 0
        
        # Start heartbeat
        asyncio.create_task(self.send_heartbeats())
    
    async def send_heartbeats(self):
        """Leader sends periodic heartbeats"""
        while self.state == NodeState.LEADER:
            for peer_id in self.peers:
                asyncio.create_task(self.append_entries(peer_id, []))
            await asyncio.sleep(self.heartbeat_interval)
    
    async def request_vote(self, peer_id: str) -> bool:
        """Request vote from a peer"""
        # In real implementation, this would be RPC
        last_log_index = len(self.log)
        last_log_term = self.log[-1].term if self.log else 0
        
        # Simulate RPC response
        # Peer grants vote if:
        # 1. Candidate's term >= peer's term
        # 2. Peer hasn't voted for someone else
        # 3. Candidate's log is at least as up-to-date
        return True  # Simplified
    
    async def append_entries(
        self, 
        peer_id: str, 
        entries: List[LogEntry]
    ) -> bool:
        """AppendEntries RPC (heartbeat + log replication)"""
        prev_log_index = self.next_index.get(peer_id, 1) - 1
        prev_log_term = self.log[prev_log_index - 1].term if prev_log_index > 0 else 0
        
        # Simulate RPC
        success = True  # In real implementation, check response
        
        if success and entries:
            self.match_index[peer_id] = prev_log_index + len(entries)
            self.next_index[peer_id] = self.match_index[peer_id] + 1
            
            # Update commit index
            self.update_commit_index()
        
        return success
    
    def update_commit_index(self):
        """Leader updates commit index based on replication"""
        for n in range(self.commit_index + 1, len(self.log) + 1):
            replicated_count = 1  # Count self
            for peer_id in self.peers:
                if self.match_index.get(peer_id, 0) >= n:
                    replicated_count += 1
            
            quorum = (len(self.peers) + 1) // 2 + 1
            if replicated_count >= quorum:
                if self.log[n - 1].term == self.current_term:
                    self.commit_index = n
    
    def propose(self, command: str, data: dict) -> Optional[LogEntry]:
        """Client proposes a command (only leader can accept)"""
        if self.state != NodeState.LEADER:
            return None  # Redirect to leader
        
        entry = LogEntry(
            term=self.current_term,
            index=len(self.log) + 1,
            command=command,
            data=data
        )
        self.log.append(entry)
        
        # Replicate to followers (async)
        asyncio.create_task(self.replicate_log())
        
        return entry
    
    async def replicate_log(self):
        """Replicate latest log entries to all followers"""
        for peer_id in self.peers:
            next_idx = self.next_index.get(peer_id, 1)
            entries_to_send = self.log[next_idx - 1:]
            
            if entries_to_send:
                await self.append_entries(peer_id, entries_to_send)


# Leader election simulation
async def simulate_raft():
    nodes = [
        RaftNode(node_id="node-1", peers=["node-2", "node-3"]),
        RaftNode(node_id="node-2", peers=["node-1", "node-3"]),
        RaftNode(node_id="node-3", peers=["node-1", "node-2"]),
    ]
    
    # Simulate node-1 winning election
    leader = nodes[0]
    await leader.start_election()
    
    # Propose a command
    if leader.state == NodeState.LEADER:
        entry = leader.propose("SET", {"key": "name", "value": "Alice"})
        print(f"Proposed: {entry}")

// Raft Node States
const NodeState = {
  FOLLOWER: 'follower',
  CANDIDATE: 'candidate',
  LEADER: 'leader'
};

class LogEntry {
  constructor(term, index, command, data) {
    this.term = term;
    this.index = index;
    this.command = command;
    this.data = data;
    this.timestamp = Date.now();
  }
}

class RaftNode {
  constructor(nodeId, peers = []) {
    this.nodeId = nodeId;
    this.state = NodeState.FOLLOWER;
    this.currentTerm = 0;
    this.votedFor = null;
    this.log = [];
    
    // Volatile state
    this.commitIndex = 0;
    this.lastApplied = 0;
    
    // Leader state
    this.nextIndex = new Map();
    this.matchIndex = new Map();
    
    // Cluster
    this.peers = peers;
    this.leaderId = null;
    
    // Timers
    this.electionTimer = null;
    this.heartbeatTimer = null;
    this.heartbeatInterval = 150; // ms
    
    this.resetElectionTimeout();
  }

  resetElectionTimeout() {
    if (this.electionTimer) {
      clearTimeout(this.electionTimer);
    }
    
    // Randomized timeout (150-300ms)
    const timeout = 150 + Math.random() * 150;
    
    this.electionTimer = setTimeout(() => {
      if (this.state !== NodeState.LEADER) {
        this.startElection();
      }
    }, timeout);
  }

  async startElection() {
    console.log(`[${this.nodeId}] Starting election for term ${this.currentTerm + 1}`);
    
    this.state = NodeState.CANDIDATE;
    this.currentTerm++;
    this.votedFor = this.nodeId;
    let votes = 1; // Vote for self

    // Request votes in parallel
    const votePromises = this.peers.map(peerId => 
      this.requestVote(peerId)
    );

    const results = await Promise.allSettled(votePromises);
    
    for (const result of results) {
      if (result.status === 'fulfilled' && result.value.granted) {
        votes++;
        
        // Update term if needed
        if (result.value.term > this.currentTerm) {
          this.stepDown(result.value.term);
          return;
        }
      }
    }

    // Check quorum
    const quorum = Math.floor((this.peers.length + 1) / 2) + 1;
    
    if (votes >= quorum && this.state === NodeState.CANDIDATE) {
      this.becomeLeader();
    } else {
      this.state = NodeState.FOLLOWER;
      this.resetElectionTimeout();
    }
  }

  async requestVote(peerId) {
    const lastLogIndex = this.log.length;
    const lastLogTerm = lastLogIndex > 0 ? this.log[lastLogIndex - 1].term : 0;

    // Simulate RPC call
    const request = {
      term: this.currentTerm,
      candidateId: this.nodeId,
      lastLogIndex,
      lastLogTerm
    };

    // In real implementation, this would be network call
    return { granted: true, term: this.currentTerm };
  }

  becomeLeader() {
    console.log(`[${this.nodeId}] Became LEADER for term ${this.currentTerm}`);
    
    this.state = NodeState.LEADER;
    this.leaderId = this.nodeId;

    // Initialize leader state
    for (const peerId of this.peers) {
      this.nextIndex.set(peerId, this.log.length + 1);
      this.matchIndex.set(peerId, 0);
    }

    // Clear election timer
    if (this.electionTimer) {
      clearTimeout(this.electionTimer);
    }

    // Start heartbeats
    this.sendHeartbeats();
  }

  sendHeartbeats() {
    if (this.state !== NodeState.LEADER) return;

    // Send AppendEntries to all peers
    for (const peerId of this.peers) {
      this.appendEntries(peerId, []);
    }

    this.heartbeatTimer = setTimeout(
      () => this.sendHeartbeats(),
      this.heartbeatInterval
    );
  }

  async appendEntries(peerId, entries) {
    const nextIdx = this.nextIndex.get(peerId) || 1;
    const prevLogIndex = nextIdx - 1;
    const prevLogTerm = prevLogIndex > 0 ? this.log[prevLogIndex - 1].term : 0;

    const request = {
      term: this.currentTerm,
      leaderId: this.nodeId,
      prevLogIndex,
      prevLogTerm,
      entries,
      leaderCommit: this.commitIndex
    };

    // Simulate RPC - in real implementation, handle response
    const success = true;

    if (success && entries.length > 0) {
      this.matchIndex.set(peerId, prevLogIndex + entries.length);
      this.nextIndex.set(peerId, this.matchIndex.get(peerId) + 1);
      this.updateCommitIndex();
    }

    return success;
  }

  updateCommitIndex() {
    // Find highest N where majority has replicated entry N
    for (let n = this.log.length; n > this.commitIndex; n--) {
      let replicationCount = 1; // Count self

      for (const peerId of this.peers) {
        if ((this.matchIndex.get(peerId) || 0) >= n) {
          replicationCount++;
        }
      }

      const quorum = Math.floor((this.peers.length + 1) / 2) + 1;
      
      if (replicationCount >= quorum && this.log[n - 1].term === this.currentTerm) {
        this.commitIndex = n;
        this.applyCommittedEntries();
        break;
      }
    }
  }

  applyCommittedEntries() {
    while (this.lastApplied < this.commitIndex) {
      this.lastApplied++;
      const entry = this.log[this.lastApplied - 1];
      console.log(`[${this.nodeId}] Applied: ${entry.command} ${JSON.stringify(entry.data)}`);
      // Apply to state machine here
    }
  }

  // Client interface
  propose(command, data) {
    if (this.state !== NodeState.LEADER) {
      return { 
        success: false, 
        error: 'Not leader',
        leaderId: this.leaderId 
      };
    }

    const entry = new LogEntry(
      this.currentTerm,
      this.log.length + 1,
      command,
      data
    );

    this.log.push(entry);
    
    // Replicate to followers
    this.replicateLog();

    return { 
      success: true, 
      entry,
      message: 'Proposed - will be committed after replication' 
    };
  }

  async replicateLog() {
    if (this.state !== NodeState.LEADER) return;

    const promises = this.peers.map(peerId => {
      const nextIdx = this.nextIndex.get(peerId) || 1;
      const entries = this.log.slice(nextIdx - 1);
      
      if (entries.length > 0) {
        return this.appendEntries(peerId, entries);
      }
      return Promise.resolve(true);
    });

    await Promise.allSettled(promises);
    this.updateCommitIndex();
  }

  stepDown(newTerm) {
    console.log(`[${this.nodeId}] Stepping down, new term: ${newTerm}`);
    this.currentTerm = newTerm;
    this.state = NodeState.FOLLOWER;
    this.votedFor = null;
    
    if (this.heartbeatTimer) {
      clearTimeout(this.heartbeatTimer);
    }
    
    this.resetElectionTimeout();
  }
}

// Usage
const node1 = new RaftNode('node-1', ['node-2', 'node-3']);
const node2 = new RaftNode('node-2', ['node-1', 'node-3']);
const node3 = new RaftNode('node-3', ['node-1', 'node-2']);

// Simulate leader election
node1.becomeLeader();

// Propose a command
const result = node1.propose('SET', { key: 'name', value: 'Alice' });
console.log('Proposal result:', result);

Distributed Transactions

Two-Phase Commit (2PC)

Two-Phase Commit Implementation

Python
JavaScript

import asyncio
from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
import uuid

class TxState(Enum):
    PENDING = "pending"
    PREPARED = "prepared"
    COMMITTED = "committed"
    ABORTED = "aborted"

@dataclass
class Transaction:
    tx_id: str
    state: TxState = TxState.PENDING
    participants: List[str] = field(default_factory=list)
    votes: Dict[str, bool] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    timeout: timedelta = timedelta(seconds=30)

class TwoPhaseCommitCoordinator:
    """Coordinates distributed transactions using 2PC"""
    
    def __init__(self, coordinator_id: str):
        self.coordinator_id = coordinator_id
        self.transactions: Dict[str, Transaction] = {}
        self.participants: Dict[str, 'TransactionParticipant'] = {}
    
    def register_participant(self, participant: 'TransactionParticipant'):
        self.participants[participant.participant_id] = participant
    
    async def begin_transaction(
        self, 
        participant_ids: List[str],
        operations: Dict[str, Any]
    ) -> str:
        """Start a new distributed transaction"""
        tx_id = f"tx-{uuid.uuid4().hex[:8]}"
        tx = Transaction(tx_id=tx_id, participants=participant_ids)
        self.transactions[tx_id] = tx
        
        print(f"[Coordinator] Starting transaction {tx_id}")
        
        try:
            # Phase 1: Prepare
            prepared = await self.prepare_phase(tx_id, operations)
            
            if prepared:
                # Phase 2: Commit
                await self.commit_phase(tx_id)
                return tx_id
            else:
                # Phase 2: Abort
                await self.abort_phase(tx_id)
                raise Exception("Transaction aborted - not all participants prepared")
                
        except asyncio.TimeoutError:
            await self.abort_phase(tx_id)
            raise Exception("Transaction timed out")
    
    async def prepare_phase(
        self, 
        tx_id: str, 
        operations: Dict[str, Any]
    ) -> bool:
        """Phase 1: Ask all participants to prepare"""
        tx = self.transactions[tx_id]
        print(f"[Coordinator] Phase 1: PREPARE for {tx_id}")
        
        prepare_tasks = []
        for participant_id in tx.participants:
            participant = self.participants.get(participant_id)
            if participant:
                op = operations.get(participant_id, {})
                prepare_tasks.append(
                    participant.prepare(tx_id, op)
                )
        
        # Wait for all votes with timeout
        try:
            results = await asyncio.wait_for(
                asyncio.gather(*prepare_tasks, return_exceptions=True),
                timeout=tx.timeout.total_seconds()
            )
            
            # Check all votes
            for participant_id, result in zip(tx.participants, results):
                if isinstance(result, Exception):
                    tx.votes[participant_id] = False
                else:
                    tx.votes[participant_id] = result
            
            all_prepared = all(tx.votes.values())
            tx.state = TxState.PREPARED if all_prepared else TxState.PENDING
            
            return all_prepared
            
        except asyncio.TimeoutError:
            return False
    
    async def commit_phase(self, tx_id: str):
        """Phase 2: Tell all participants to commit"""
        tx = self.transactions[tx_id]
        print(f"[Coordinator] Phase 2: COMMIT for {tx_id}")
        
        commit_tasks = []
        for participant_id in tx.participants:
            participant = self.participants.get(participant_id)
            if participant:
                commit_tasks.append(participant.commit(tx_id))
        
        await asyncio.gather(*commit_tasks, return_exceptions=True)
        tx.state = TxState.COMMITTED
        print(f"[Coordinator] Transaction {tx_id} COMMITTED")
    
    async def abort_phase(self, tx_id: str):
        """Phase 2: Tell all participants to abort"""
        tx = self.transactions[tx_id]
        print(f"[Coordinator] Phase 2: ABORT for {tx_id}")
        
        abort_tasks = []
        for participant_id in tx.participants:
            participant = self.participants.get(participant_id)
            if participant:
                abort_tasks.append(participant.abort(tx_id))
        
        await asyncio.gather(*abort_tasks, return_exceptions=True)
        tx.state = TxState.ABORTED
        print(f"[Coordinator] Transaction {tx_id} ABORTED")


class TransactionParticipant:
    """Participant in a distributed transaction"""
    
    def __init__(self, participant_id: str):
        self.participant_id = participant_id
        self.prepared_data: Dict[str, Any] = {}
        self.committed_data: Dict[str, Any] = {}
        self.undo_log: Dict[str, Any] = {}
    
    async def prepare(self, tx_id: str, operation: Dict) -> bool:
        """Prepare to commit - acquire locks, validate, write to redo log"""
        print(f"[{self.participant_id}] Preparing {tx_id}: {operation}")
        
        try:
            # Simulate validation and lock acquisition
            await asyncio.sleep(0.1)  # Simulate I/O
            
            # Write to redo log (would be durable in real implementation)
            self.prepared_data[tx_id] = operation
            
            # Save undo information
            key = operation.get("key")
            if key:
                self.undo_log[tx_id] = {
                    "key": key,
                    "old_value": self.committed_data.get(key)
                }
            
            print(f"[{self.participant_id}] Voted YES for {tx_id}")
            return True
            
        except Exception as e:
            print(f"[{self.participant_id}] Voted NO for {tx_id}: {e}")
            return False
    
    async def commit(self, tx_id: str):
        """Commit the prepared transaction"""
        print(f"[{self.participant_id}] Committing {tx_id}")
        
        operation = self.prepared_data.pop(tx_id, {})
        
        # Apply the operation
        key = operation.get("key")
        value = operation.get("value")
        if key:
            self.committed_data[key] = value
        
        # Clear undo log
        self.undo_log.pop(tx_id, None)
    
    async def abort(self, tx_id: str):
        """Abort - rollback using undo log"""
        print(f"[{self.participant_id}] Aborting {tx_id}")
        
        # Rollback using undo log
        undo = self.undo_log.pop(tx_id, None)
        if undo:
            key = undo["key"]
            old_value = undo["old_value"]
            if old_value is not None:
                self.committed_data[key] = old_value
            elif key in self.committed_data:
                del self.committed_data[key]
        
        self.prepared_data.pop(tx_id, None)


# Usage example
async def demo_2pc():
    coordinator = TwoPhaseCommitCoordinator("coordinator-1")
    
    # Register participants (e.g., database shards)
    orders_db = TransactionParticipant("orders-db")
    inventory_db = TransactionParticipant("inventory-db")
    
    coordinator.register_participant(orders_db)
    coordinator.register_participant(inventory_db)
    
    # Execute distributed transaction
    try:
        tx_id = await coordinator.begin_transaction(
            participant_ids=["orders-db", "inventory-db"],
            operations={
                "orders-db": {"key": "order:123", "value": {"product": "laptop", "qty": 1}},
                "inventory-db": {"key": "product:laptop:stock", "value": 99}
            }
        )
        print(f"Transaction {tx_id} completed successfully!")
        
    except Exception as e:
        print(f"Transaction failed: {e}")

asyncio.run(demo_2pc())

const { EventEmitter } = require('events');
const { v4: uuidv4 } = require('uuid');

// Transaction states
const TxState = {
  PENDING: 'pending',
  PREPARED: 'prepared',
  COMMITTED: 'committed',
  ABORTED: 'aborted'
};

class Transaction {
  constructor(txId, participants) {
    this.txId = txId;
    this.state = TxState.PENDING;
    this.participants = participants;
    this.votes = new Map();
    this.createdAt = Date.now();
    this.timeout = 30000; // 30 seconds
  }
}

class TwoPhaseCommitCoordinator extends EventEmitter {
  constructor(coordinatorId) {
    super();
    this.coordinatorId = coordinatorId;
    this.transactions = new Map();
    this.participants = new Map();
  }

  registerParticipant(participant) {
    this.participants.set(participant.participantId, participant);
  }

  async beginTransaction(participantIds, operations) {
    const txId = `tx-${uuidv4().slice(0, 8)}`;
    const tx = new Transaction(txId, participantIds);
    this.transactions.set(txId, tx);

    console.log(`[Coordinator] Starting transaction ${txId}`);

    try {
      // Phase 1: Prepare
      const prepared = await this.preparePhase(txId, operations);

      if (prepared) {
        // Phase 2: Commit
        await this.commitPhase(txId);
        return { success: true, txId };
      } else {
        // Phase 2: Abort
        await this.abortPhase(txId);
        throw new Error('Transaction aborted - not all participants prepared');
      }
    } catch (error) {
      if (error.message !== 'Transaction aborted - not all participants prepared') {
        await this.abortPhase(txId);
      }
      throw error;
    }
  }

  async preparePhase(txId, operations) {
    const tx = this.transactions.get(txId);
    console.log(`[Coordinator] Phase 1: PREPARE for ${txId}`);

    const preparePromises = tx.participants.map(async (participantId) => {
      const participant = this.participants.get(participantId);
      if (!participant) return false;

      const operation = operations[participantId] || {};
      
      try {
        const vote = await this.withTimeout(
          participant.prepare(txId, operation),
          tx.timeout
        );
        return { participantId, vote };
      } catch (error) {
        console.error(`[Coordinator] ${participantId} failed to prepare:`, error.message);
        return { participantId, vote: false };
      }
    });

    const results = await Promise.all(preparePromises);

    // Collect votes
    let allPrepared = true;
    for (const { participantId, vote } of results) {
      tx.votes.set(participantId, vote);
      if (!vote) allPrepared = false;
    }

    tx.state = allPrepared ? TxState.PREPARED : TxState.PENDING;
    return allPrepared;
  }

  async commitPhase(txId) {
    const tx = this.transactions.get(txId);
    console.log(`[Coordinator] Phase 2: COMMIT for ${txId}`);

    const commitPromises = tx.participants.map(async (participantId) => {
      const participant = this.participants.get(participantId);
      if (participant) {
        await participant.commit(txId);
      }
    });

    await Promise.allSettled(commitPromises);
    tx.state = TxState.COMMITTED;
    console.log(`[Coordinator] Transaction ${txId} COMMITTED`);
    
    this.emit('committed', txId);
  }

  async abortPhase(txId) {
    const tx = this.transactions.get(txId);
    console.log(`[Coordinator] Phase 2: ABORT for ${txId}`);

    const abortPromises = tx.participants.map(async (participantId) => {
      const participant = this.participants.get(participantId);
      if (participant) {
        await participant.abort(txId);
      }
    });

    await Promise.allSettled(abortPromises);
    tx.state = TxState.ABORTED;
    console.log(`[Coordinator] Transaction ${txId} ABORTED`);
    
    this.emit('aborted', txId);
  }

  withTimeout(promise, ms) {
    return Promise.race([
      promise,
      new Promise((_, reject) =>
        setTimeout(() => reject(new Error('Timeout')), ms)
      )
    ]);
  }
}

class TransactionParticipant extends EventEmitter {
  constructor(participantId) {
    super();
    this.participantId = participantId;
    this.preparedData = new Map();
    this.committedData = new Map();
    this.undoLog = new Map();
  }

  async prepare(txId, operation) {
    console.log(`[${this.participantId}] Preparing ${txId}:`, operation);

    try {
      // Simulate validation, lock acquisition, redo log
      await this.simulateIO(100);

      // Write to redo log
      this.preparedData.set(txId, operation);

      // Save undo information
      const { key } = operation;
      if (key) {
        this.undoLog.set(txId, {
          key,
          oldValue: this.committedData.get(key)
        });
      }

      console.log(`[${this.participantId}] Voted YES for ${txId}`);
      return true;
    } catch (error) {
      console.log(`[${this.participantId}] Voted NO for ${txId}:`, error.message);
      return false;
    }
  }

  async commit(txId) {
    console.log(`[${this.participantId}] Committing ${txId}`);

    const operation = this.preparedData.get(txId);
    this.preparedData.delete(txId);

    if (operation && operation.key) {
      this.committedData.set(operation.key, operation.value);
    }

    // Clear undo log
    this.undoLog.delete(txId);
    
    this.emit('committed', txId);
  }

  async abort(txId) {
    console.log(`[${this.participantId}] Aborting ${txId}`);

    // Rollback using undo log
    const undo = this.undoLog.get(txId);
    if (undo) {
      if (undo.oldValue !== undefined) {
        this.committedData.set(undo.key, undo.oldValue);
      } else {
        this.committedData.delete(undo.key);
      }
    }

    this.preparedData.delete(txId);
    this.undoLog.delete(txId);
    
    this.emit('aborted', txId);
  }

  simulateIO(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  // Utility to get committed data
  get(key) {
    return this.committedData.get(key);
  }
}

// Usage
async function demo2PC() {
  const coordinator = new TwoPhaseCommitCoordinator('coordinator-1');

  const ordersDb = new TransactionParticipant('orders-db');
  const inventoryDb = new TransactionParticipant('inventory-db');

  coordinator.registerParticipant(ordersDb);
  coordinator.registerParticipant(inventoryDb);

  try {
    const result = await coordinator.beginTransaction(
      ['orders-db', 'inventory-db'],
      {
        'orders-db': { key: 'order:123', value: { product: 'laptop', qty: 1 } },
        'inventory-db': { key: 'product:laptop:stock', value: 99 }
      }
    );
    console.log('Transaction completed:', result);
    console.log('Orders DB:', ordersDb.get('order:123'));
    console.log('Inventory DB:', inventoryDb.get('product:laptop:stock'));
  } catch (error) {
    console.error('Transaction failed:', error.message);
  }
}

demo2PC();

Saga Pattern

Choreography vs Orchestration

Choreography (Event-driven)

Order ─► OrderCreated event
              │
    ┌─────────┼─────────┐
    ▼         ▼         ▼
Inventory  Payment  Shipping
    │         │         │
    └─────────┴─────────┘
          Events

• No central controller
• Services react to events
• More decoupled


Orchestration (Command-driven)

    ┌───────────────────┐
    │   Saga Manager    │
    └─────────┬─────────┘
              │
    ┌─────────┼─────────┐
    ▼         ▼         ▼
Inventory  Payment  Shipping
    
• Central controller
• Explicit flow control
• Easier to understand

Handling Failures

Circuit Breaker Pattern

Python
JavaScript

import time
import asyncio
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, TypeVar, Generic, Optional, Any
from functools import wraps
from datetime import datetime, timedelta
import logging

T = TypeVar('T')
logger = logging.getLogger(__name__)

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if service recovered

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    success_threshold: int = 2  # Successes needed to close from half-open
    timeout: timedelta = timedelta(seconds=60)
    half_open_max_calls: int = 3
    excluded_exceptions: tuple = ()  # Exceptions that don't count as failures

class CircuitBreakerOpenException(Exception):
    """Raised when circuit breaker is open"""
    def __init__(self, circuit_name: str, retry_after: float):
        self.circuit_name = circuit_name
        self.retry_after = retry_after
        super().__init__(f"Circuit {circuit_name} is OPEN. Retry after {retry_after:.1f}s")

@dataclass
class CircuitBreaker:
    """Production-ready circuit breaker with metrics"""
    name: str
    config: CircuitBreakerConfig = field(default_factory=CircuitBreakerConfig)
    
    state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    failure_count: int = field(default=0, init=False)
    success_count: int = field(default=0, init=False)
    last_failure_time: Optional[datetime] = field(default=None, init=False)
    half_open_calls: int = field(default=0, init=False)
    
    # Metrics
    total_calls: int = field(default=0, init=False)
    total_failures: int = field(default=0, init=False)
    total_rejections: int = field(default=0, init=False)

    def __call__(self, func: Callable) -> Callable:
        """Use as decorator"""
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            return await self.call_async(lambda: func(*args, **kwargs))
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            return self.call_sync(lambda: func(*args, **kwargs))
        
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper

    def call_sync(self, func: Callable[[], T]) -> T:
        """Execute function with circuit breaker (sync)"""
        self._check_state()
        self.total_calls += 1
        
        try:
            result = func()
            self._on_success()
            return result
        except self.config.excluded_exceptions:
            raise  # Don't count as failure
        except Exception as e:
            self._on_failure()
            raise

    async def call_async(self, func: Callable) -> Any:
        """Execute function with circuit breaker (async)"""
        self._check_state()
        self.total_calls += 1
        
        try:
            result = await func() if asyncio.iscoroutinefunction(func) else func()
            self._on_success()
            return result
        except self.config.excluded_exceptions:
            raise
        except Exception as e:
            self._on_failure()
            raise

    def _check_state(self):
        """Check if request should be allowed"""
        if self.state == CircuitState.CLOSED:
            return
        
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self._transition_to_half_open()
            else:
                self.total_rejections += 1
                retry_after = (
                    self.last_failure_time + self.config.timeout - datetime.now()
                ).total_seconds()
                raise CircuitBreakerOpenException(self.name, max(0, retry_after))
        
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.config.half_open_max_calls:
                self.total_rejections += 1
                raise CircuitBreakerOpenException(self.name, 1.0)
            self.half_open_calls += 1

    def _should_attempt_reset(self) -> bool:
        """Check if enough time has passed to try again"""
        if self.last_failure_time is None:
            return True
        return datetime.now() - self.last_failure_time >= self.config.timeout

    def _transition_to_half_open(self):
        """Move to half-open state"""
        logger.info(f"Circuit {self.name}: OPEN -> HALF_OPEN")
        self.state = CircuitState.HALF_OPEN
        self.half_open_calls = 0
        self.success_count = 0

    def _on_success(self):
        """Handle successful call"""
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self._close()
        else:
            self.failure_count = 0  # Reset on success

    def _on_failure(self):
        """Handle failed call"""
        self.total_failures += 1
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.state == CircuitState.HALF_OPEN:
            self._open()
        elif self.failure_count >= self.config.failure_threshold:
            self._open()

    def _open(self):
        """Open the circuit"""
        logger.warning(f"Circuit {self.name}: -> OPEN (failures: {self.failure_count})")
        self.state = CircuitState.OPEN

    def _close(self):
        """Close the circuit"""
        logger.info(f"Circuit {self.name}: -> CLOSED")
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.half_open_calls = 0

    @property
    def metrics(self) -> dict:
        """Get circuit breaker metrics"""
        return {
            "name": self.name,
            "state": self.state.value,
            "total_calls": self.total_calls,
            "total_failures": self.total_failures,
            "total_rejections": self.total_rejections,
            "failure_rate": self.total_failures / max(1, self.total_calls),
            "current_failure_count": self.failure_count
        }


# Usage as decorator
payment_circuit = CircuitBreaker(
    name="payment-service",
    config=CircuitBreakerConfig(
        failure_threshold=3,
        success_threshold=2,
        timeout=timedelta(seconds=30)
    )
)

@payment_circuit
async def call_payment_service(amount: float):
    # Simulate external API call
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://api.payment.com/charge",
            json={"amount": amount}
        ) as response:
            if response.status != 200:
                raise Exception(f"Payment failed: {response.status}")
            return await response.json()


# Usage with context manager
class CircuitBreakerContext:
    def __init__(self, circuit: CircuitBreaker):
        self.circuit = circuit
    
    def __enter__(self):
        self.circuit._check_state()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.circuit._on_success()
        elif exc_type not in self.circuit.config.excluded_exceptions:
            self.circuit._on_failure()
        return False  # Don't suppress exceptions


# Example with fallback
async def get_user_with_fallback(user_id: str):
    try:
        return await call_user_service(user_id)
    except CircuitBreakerOpenException:
        # Fallback: return cached data
        return get_cached_user(user_id)

const EventEmitter = require('events');

// Circuit States
const CircuitState = {
  CLOSED: 'closed',
  OPEN: 'open',
  HALF_OPEN: 'half_open'
};

class CircuitBreakerOpenError extends Error {
  constructor(circuitName, retryAfter) {
    super(`Circuit ${circuitName} is OPEN. Retry after ${retryAfter.toFixed(1)}s`);
    this.name = 'CircuitBreakerOpenError';
    this.circuitName = circuitName;
    this.retryAfter = retryAfter;
  }
}

class CircuitBreaker extends EventEmitter {
  constructor(name, options = {}) {
    super();
    this.name = name;
    this.config = {
      failureThreshold: options.failureThreshold || 5,
      successThreshold: options.successThreshold || 2,
      timeout: options.timeout || 60000, // 60 seconds
      halfOpenMaxCalls: options.halfOpenMaxCalls || 3,
      excludedErrors: options.excludedErrors || []
    };

    this.state = CircuitState.CLOSED;
    this.failureCount = 0;
    this.successCount = 0;
    this.lastFailureTime = null;
    this.halfOpenCalls = 0;

    // Metrics
    this.metrics = {
      totalCalls: 0,
      totalFailures: 0,
      totalRejections: 0,
      totalSuccesses: 0
    };
  }

  async execute(fn) {
    this.checkState();
    this.metrics.totalCalls++;

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      // Check if error should be excluded
      const isExcluded = this.config.excludedErrors.some(
        ErrorClass => error instanceof ErrorClass
      );

      if (!isExcluded) {
        this.onFailure();
      }
      throw error;
    }
  }

  // Wrap a function with circuit breaker
  wrap(fn) {
    return async (...args) => {
      return this.execute(() => fn(...args));
    };
  }

  checkState() {
    if (this.state === CircuitState.CLOSED) {
      return;
    }

    if (this.state === CircuitState.OPEN) {
      if (this.shouldAttemptReset()) {
        this.transitionToHalfOpen();
      } else {
        this.metrics.totalRejections++;
        const retryAfter = (
          this.lastFailureTime + this.config.timeout - Date.now()
        ) / 1000;
        throw new CircuitBreakerOpenError(this.name, Math.max(0, retryAfter));
      }
    }

    if (this.state === CircuitState.HALF_OPEN) {
      if (this.halfOpenCalls >= this.config.halfOpenMaxCalls) {
        this.metrics.totalRejections++;
        throw new CircuitBreakerOpenError(this.name, 1.0);
      }
      this.halfOpenCalls++;
    }
  }

  shouldAttemptReset() {
    if (!this.lastFailureTime) return true;
    return Date.now() - this.lastFailureTime >= this.config.timeout;
  }

  transitionToHalfOpen() {
    console.log(`Circuit ${this.name}: OPEN -> HALF_OPEN`);
    this.state = CircuitState.HALF_OPEN;
    this.halfOpenCalls = 0;
    this.successCount = 0;
    this.emit('half-open');
  }

  onSuccess() {
    this.metrics.totalSuccesses++;

    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= this.config.successThreshold) {
        this.close();
      }
    } else {
      this.failureCount = 0;
    }
  }

  onFailure() {
    this.metrics.totalFailures++;
    this.failureCount++;
    this.lastFailureTime = Date.now();

    if (this.state === CircuitState.HALF_OPEN) {
      this.open();
    } else if (this.failureCount >= this.config.failureThreshold) {
      this.open();
    }
  }

  open() {
    console.log(`Circuit ${this.name}: -> OPEN (failures: ${this.failureCount})`);
    this.state = CircuitState.OPEN;
    this.emit('open');
  }

  close() {
    console.log(`Circuit ${this.name}: -> CLOSED`);
    this.state = CircuitState.CLOSED;
    this.failureCount = 0;
    this.successCount = 0;
    this.halfOpenCalls = 0;
    this.emit('close');
  }

  getMetrics() {
    return {
      name: this.name,
      state: this.state,
      ...this.metrics,
      failureRate: this.metrics.totalFailures / Math.max(1, this.metrics.totalCalls),
      currentFailureCount: this.failureCount
    };
  }

  // Manual control
  forceOpen() {
    this.open();
  }

  forceClose() {
    this.close();
  }
}

// Usage
const paymentCircuit = new CircuitBreaker('payment-service', {
  failureThreshold: 3,
  successThreshold: 2,
  timeout: 30000
});

// Event listeners
paymentCircuit.on('open', () => {
  console.log('🔴 Payment circuit opened - switching to fallback');
  // Alert, switch to backup, etc.
});

paymentCircuit.on('close', () => {
  console.log('🟢 Payment circuit closed - back to normal');
});

// Wrap existing function
async function chargePayment(amount) {
  const response = await fetch('https://api.payment.com/charge', {
    method: 'POST',
    body: JSON.stringify({ amount })
  });
  
  if (!response.ok) {
    throw new Error(`Payment failed: ${response.status}`);
  }
  
  return response.json();
}

const safeChargePayment = paymentCircuit.wrap(chargePayment);

// With fallback
async function chargeWithFallback(amount) {
  try {
    return await safeChargePayment(amount);
  } catch (error) {
    if (error instanceof CircuitBreakerOpenError) {
      // Use fallback payment processor
      return await fallbackPaymentProcessor(amount);
    }
    throw error;
  }
}

// Express middleware example
function circuitBreakerMiddleware(circuit) {
  return (req, res, next) => {
    req.circuit = circuit;
    
    // Override res.json to track success/failure
    const originalJson = res.json.bind(res);
    res.json = function(data) {
      if (res.statusCode >= 500) {
        circuit.onFailure();
      } else {
        circuit.onSuccess();
      }
      return originalJson(data);
    };
    
    next();
  };
}

Retry with Exponential Backoff

Python
JavaScript

import asyncio
import random
import time
from functools import wraps
from typing import Callable, TypeVar, Type, Tuple, Optional
from dataclasses import dataclass
import logging

T = TypeVar('T')
logger = logging.getLogger(__name__)

@dataclass
class RetryConfig:
    max_retries: int = 5
    base_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0
    jitter: bool = True
    retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,)
    on_retry: Optional[Callable[[Exception, int], None]] = None

class RetryExhausted(Exception):
    """All retries failed"""
    def __init__(self, last_exception: Exception, attempts: int):
        self.last_exception = last_exception
        self.attempts = attempts
        super().__init__(f"All {attempts} retry attempts failed: {last_exception}")

def retry_with_backoff(config: Optional[RetryConfig] = None):
    """Decorator for retry with exponential backoff"""
    if config is None:
        config = RetryConfig()
    
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def async_wrapper(*args, **kwargs) -> T:
            last_exception = None
            
            for attempt in range(config.max_retries):
                try:
                    return await func(*args, **kwargs)
                    
                except config.retryable_exceptions as e:
                    last_exception = e
                    
                    if attempt == config.max_retries - 1:
                        break
                    
                    # Calculate delay with exponential backoff
                    delay = min(
                        config.base_delay * (config.exponential_base ** attempt),
                        config.max_delay
                    )
                    
                    # Add jitter to prevent thundering herd
                    if config.jitter:
                        delay = delay * (0.5 + random.random())
                    
                    logger.warning(
                        f"Attempt {attempt + 1}/{config.max_retries} failed: {e}. "
                        f"Retrying in {delay:.2f}s"
                    )
                    
                    if config.on_retry:
                        config.on_retry(e, attempt + 1)
                    
                    await asyncio.sleep(delay)
            
            raise RetryExhausted(last_exception, config.max_retries)
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs) -> T:
            last_exception = None
            
            for attempt in range(config.max_retries):
                try:
                    return func(*args, **kwargs)
                    
                except config.retryable_exceptions as e:
                    last_exception = e
                    
                    if attempt == config.max_retries - 1:
                        break
                    
                    delay = min(
                        config.base_delay * (config.exponential_base ** attempt),
                        config.max_delay
                    )
                    
                    if config.jitter:
                        delay = delay * (0.5 + random.random())
                    
                    logger.warning(
                        f"Attempt {attempt + 1}/{config.max_retries} failed: {e}. "
                        f"Retrying in {delay:.2f}s"
                    )
                    
                    if config.on_retry:
                        config.on_retry(e, attempt + 1)
                    
                    time.sleep(delay)
            
            raise RetryExhausted(last_exception, config.max_retries)
        
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    
    return decorator


# Usage examples
@retry_with_backoff(RetryConfig(
    max_retries=3,
    base_delay=1.0,
    retryable_exceptions=(ConnectionError, TimeoutError)
))
async def fetch_user_data(user_id: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as resp:
            if resp.status == 503:
                raise ConnectionError("Service unavailable")
            return await resp.json()


# Retry with callback for metrics
def on_retry_callback(exception: Exception, attempt: int):
    metrics.increment("api.retries", tags={"attempt": attempt})

@retry_with_backoff(RetryConfig(
    max_retries=5,
    on_retry=on_retry_callback
))
def call_external_api():
    response = requests.get("https://api.example.com/data")
    response.raise_for_status()
    return response.json()


# Combine with circuit breaker
@payment_circuit
@retry_with_backoff(RetryConfig(max_retries=2))
async def charge_payment_resilient(amount: float):
    """Retries within circuit breaker protection"""
    return await call_payment_service(amount)

class RetryConfig {
  constructor(options = {}) {
    this.maxRetries = options.maxRetries || 5;
    this.baseDelay = options.baseDelay || 1000; // ms
    this.maxDelay = options.maxDelay || 60000;
    this.exponentialBase = options.exponentialBase || 2;
    this.jitter = options.jitter !== false;
    this.retryableErrors = options.retryableErrors || [Error];
    this.onRetry = options.onRetry || null;
    this.shouldRetry = options.shouldRetry || null; // Custom retry predicate
  }
}

class RetryExhaustedError extends Error {
  constructor(lastError, attempts) {
    super(`All ${attempts} retry attempts failed: ${lastError.message}`);
    this.name = 'RetryExhaustedError';
    this.lastError = lastError;
    this.attempts = attempts;
  }
}

async function retryWithBackoff(fn, config = new RetryConfig()) {
  let lastError;

  for (let attempt = 0; attempt < config.maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      lastError = error;

      // Check if we should retry this error
      const isRetryable = config.shouldRetry 
        ? config.shouldRetry(error)
        : config.retryableErrors.some(ErrorClass => error instanceof ErrorClass);

      if (!isRetryable || attempt === config.maxRetries - 1) {
        break;
      }

      // Calculate delay with exponential backoff
      let delay = Math.min(
        config.baseDelay * Math.pow(config.exponentialBase, attempt),
        config.maxDelay
      );

      // Add jitter
      if (config.jitter) {
        delay = delay * (0.5 + Math.random());
      }

      console.warn(
        `Attempt ${attempt + 1}/${config.maxRetries} failed: ${error.message}. ` +
        `Retrying in ${(delay / 1000).toFixed(2)}s`
      );

      if (config.onRetry) {
        await config.onRetry(error, attempt + 1);
      }

      await sleep(delay);
    }
  }

  throw new RetryExhaustedError(lastError, config.maxRetries);
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

// Decorator-style wrapper
function withRetry(config = new RetryConfig()) {
  return function(fn) {
    return async function(...args) {
      return retryWithBackoff(() => fn.apply(this, args), config);
    };
  };
}

// Usage examples
const fetchUserData = withRetry(new RetryConfig({
  maxRetries: 3,
  baseDelay: 1000,
  shouldRetry: (error) => {
    // Retry on network errors or 5xx
    return error.code === 'ECONNREFUSED' ||
           error.code === 'ETIMEDOUT' ||
           (error.status && error.status >= 500);
  }
}))(async function(userId) {
  const response = await fetch(`https://api.example.com/users/${userId}`);
  if (!response.ok) {
    const error = new Error(`HTTP ${response.status}`);
    error.status = response.status;
    throw error;
  }
  return response.json();
});

// With metrics callback
async function callApiWithMetrics() {
  return retryWithBackoff(
    () => fetch('https://api.example.com/data'),
    new RetryConfig({
      maxRetries: 5,
      onRetry: async (error, attempt) => {
        // Send metrics
        metrics.increment('api.retries', { attempt });
        
        // Maybe alert if approaching max retries
        if (attempt >= 4) {
          await alertTeam(`API retry attempt ${attempt}`);
        }
      }
    })
  );
}

// Combine with circuit breaker
const paymentService = paymentCircuit.wrap(
  withRetry(new RetryConfig({ maxRetries: 2 }))(
    async function chargePayment(amount) {
      const response = await fetch('https://api.payment.com/charge', {
        method: 'POST',
        body: JSON.stringify({ amount })
      });
      if (!response.ok) throw new Error('Payment failed');
      return response.json();
    }
  )
);

// Express error handling middleware with retry
function retryMiddleware(config) {
  return (req, res, next) => {
    // Attach retry helper to request
    req.retryable = async (fn) => {
      return retryWithBackoff(fn, config);
    };
    next();
  };
}

// Usage in route
app.get('/users/:id', retryMiddleware(new RetryConfig()), async (req, res) => {
  try {
    const user = await req.retryable(() => fetchUserFromDb(req.params.id));
    res.json(user);
  } catch (error) {
    if (error instanceof RetryExhaustedError) {
      res.status(503).json({ error: 'Service temporarily unavailable' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
});

Bulkhead Pattern

Isolate failures to prevent cascade.

Python
JavaScript

import asyncio
from dataclasses import dataclass, field
from typing import Dict, Optional, Callable, TypeVar, Any
from functools import wraps
from contextlib import asynccontextmanager
import time
from enum import Enum

T = TypeVar('T')

class BulkheadRejectedException(Exception):
    """Raised when bulkhead rejects request"""
    def __init__(self, bulkhead_name: str, reason: str):
        self.bulkhead_name = bulkhead_name
        self.reason = reason
        super().__init__(f"Bulkhead {bulkhead_name} rejected: {reason}")

@dataclass
class BulkheadConfig:
    max_concurrent: int = 10
    max_queue: int = 100
    queue_timeout: float = 5.0  # seconds

@dataclass
class BulkheadMetrics:
    active_count: int = 0
    queue_size: int = 0
    total_accepted: int = 0
    total_rejected: int = 0
    total_timeout: int = 0

class Bulkhead:
    """
    Bulkhead pattern - isolate components to prevent cascade failures.
    Limits concurrent execution and queues excess requests.
    """
    
    def __init__(self, name: str, config: BulkheadConfig = None):
        self.name = name
        self.config = config or BulkheadConfig()
        self.semaphore = asyncio.Semaphore(self.config.max_concurrent)
        self.metrics = BulkheadMetrics()
        self._queue = asyncio.Queue(maxsize=self.config.max_queue)
    
    def __call__(self, func: Callable) -> Callable:
        """Use as decorator"""
        @wraps(func)
        async def wrapper(*args, **kwargs):
            async with self.acquire():
                return await func(*args, **kwargs)
        return wrapper
    
    @asynccontextmanager
    async def acquire(self):
        """Acquire a slot in the bulkhead"""
        # Check if we can get a semaphore slot
        if self.semaphore.locked() and self._queue.full():
            self.metrics.total_rejected += 1
            raise BulkheadRejectedException(
                self.name, 
                f"Max concurrent ({self.config.max_concurrent}) and queue ({self.config.max_queue}) reached"
            )
        
        # Try to acquire with timeout
        try:
            await asyncio.wait_for(
                self.semaphore.acquire(),
                timeout=self.config.queue_timeout
            )
        except asyncio.TimeoutError:
            self.metrics.total_timeout += 1
            raise BulkheadRejectedException(
                self.name,
                f"Queue timeout after {self.config.queue_timeout}s"
            )
        
        self.metrics.active_count += 1
        self.metrics.total_accepted += 1
        
        try:
            yield
        finally:
            self.metrics.active_count -= 1
            self.semaphore.release()
    
    def get_metrics(self) -> dict:
        return {
            "name": self.name,
            "active": self.metrics.active_count,
            "max_concurrent": self.config.max_concurrent,
            "accepted": self.metrics.total_accepted,
            "rejected": self.metrics.total_rejected,
            "timeouts": self.metrics.total_timeout,
            "utilization": self.metrics.active_count / self.config.max_concurrent
        }


class ThreadPoolBulkhead:
    """
    Thread pool bulkhead for sync operations.
    Each pool isolates a group of operations.
    """
    
    def __init__(self, name: str, max_workers: int = 10):
        from concurrent.futures import ThreadPoolExecutor
        self.name = name
        self.executor = ThreadPoolExecutor(
            max_workers=max_workers,
            thread_name_prefix=f"bulkhead-{name}"
        )
        self.max_workers = max_workers
    
    def submit(self, func: Callable, *args, **kwargs):
        """Submit work to the bulkhead thread pool"""
        return self.executor.submit(func, *args, **kwargs)
    
    async def run(self, func: Callable, *args, **kwargs) -> Any:
        """Run sync function in bulkhead pool"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.executor,
            lambda: func(*args, **kwargs)
        )


# Bulkhead manager for multiple isolated pools
class BulkheadManager:
    """Manage multiple bulkheads for different services"""
    
    def __init__(self):
        self.bulkheads: Dict[str, Bulkhead] = {}
    
    def create(self, name: str, config: BulkheadConfig = None) -> Bulkhead:
        bulkhead = Bulkhead(name, config)
        self.bulkheads[name] = bulkhead
        return bulkhead
    
    def get(self, name: str) -> Optional[Bulkhead]:
        return self.bulkheads.get(name)
    
    def get_all_metrics(self) -> Dict[str, dict]:
        return {
            name: bh.get_metrics() 
            for name, bh in self.bulkheads.items()
        }


# Usage
bulkhead_manager = BulkheadManager()

# Separate pools for different services
payment_bulkhead = bulkhead_manager.create(
    "payment",
    BulkheadConfig(max_concurrent=5, max_queue=20)
)

inventory_bulkhead = bulkhead_manager.create(
    "inventory", 
    BulkheadConfig(max_concurrent=10, max_queue=50)
)

notification_bulkhead = bulkhead_manager.create(
    "notification",
    BulkheadConfig(max_concurrent=20, max_queue=100)
)

@payment_bulkhead
async def process_payment(order_id: str, amount: float):
    """Isolated payment processing"""
    await asyncio.sleep(0.5)  # Simulate API call
    return {"status": "success", "order_id": order_id}

@inventory_bulkhead  
async def check_inventory(product_id: str):
    """Isolated inventory check"""
    await asyncio.sleep(0.1)
    return {"available": True, "quantity": 100}

# Even if payment service is slow/failing,
# inventory and notifications continue working
async def handle_order(order):
    try:
        # These run in isolated pools
        inventory = await check_inventory(order["product_id"])
        payment = await process_payment(order["id"], order["amount"])
        await send_notification(order["user_id"])
        return {"status": "completed"}
    except BulkheadRejectedException as e:
        return {"status": "retry_later", "reason": str(e)}

class BulkheadRejectedException extends Error {
  constructor(bulkheadName, reason) {
    super(`Bulkhead ${bulkheadName} rejected: ${reason}`);
    this.name = 'BulkheadRejectedException';
    this.bulkheadName = bulkheadName;
    this.reason = reason;
  }
}

class Bulkhead {
  constructor(name, options = {}) {
    this.name = name;
    this.maxConcurrent = options.maxConcurrent || 10;
    this.maxQueue = options.maxQueue || 100;
    this.queueTimeout = options.queueTimeout || 5000;

    this.activeCount = 0;
    this.queue = [];
    
    // Metrics
    this.metrics = {
      totalAccepted: 0,
      totalRejected: 0,
      totalTimeout: 0
    };
  }

  async execute(fn) {
    // Check if we can accept the request
    if (this.activeCount >= this.maxConcurrent) {
      if (this.queue.length >= this.maxQueue) {
        this.metrics.totalRejected++;
        throw new BulkheadRejectedException(
          this.name,
          `Max concurrent (${this.maxConcurrent}) and queue (${this.maxQueue}) reached`
        );
      }

      // Queue the request
      try {
        await this.waitForSlot();
      } catch (error) {
        this.metrics.totalTimeout++;
        throw new BulkheadRejectedException(
          this.name,
          `Queue timeout after ${this.queueTimeout}ms`
        );
      }
    }

    this.activeCount++;
    this.metrics.totalAccepted++;

    try {
      return await fn();
    } finally {
      this.activeCount--;
      this.releaseNext();
    }
  }

  waitForSlot() {
    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        // Remove from queue
        const index = this.queue.findIndex(item => item.resolve === resolve);
        if (index !== -1) {
          this.queue.splice(index, 1);
        }
        reject(new Error('Queue timeout'));
      }, this.queueTimeout);

      this.queue.push({
        resolve: () => {
          clearTimeout(timeout);
          resolve();
        },
        reject,
        enqueuedAt: Date.now()
      });
    });
  }

  releaseNext() {
    if (this.queue.length > 0 && this.activeCount < this.maxConcurrent) {
      const next = this.queue.shift();
      next.resolve();
    }
  }

  // Wrap a function with bulkhead
  wrap(fn) {
    return async (...args) => {
      return this.execute(() => fn(...args));
    };
  }

  getMetrics() {
    return {
      name: this.name,
      active: this.activeCount,
      queueSize: this.queue.length,
      maxConcurrent: this.maxConcurrent,
      maxQueue: this.maxQueue,
      ...this.metrics,
      utilization: this.activeCount / this.maxConcurrent
    };
  }
}

// Bulkhead manager
class BulkheadManager {
  constructor() {
    this.bulkheads = new Map();
  }

  create(name, options) {
    const bulkhead = new Bulkhead(name, options);
    this.bulkheads.set(name, bulkhead);
    return bulkhead;
  }

  get(name) {
    return this.bulkheads.get(name);
  }

  getAllMetrics() {
    const metrics = {};
    for (const [name, bulkhead] of this.bulkheads) {
      metrics[name] = bulkhead.getMetrics();
    }
    return metrics;
  }
}

// Usage
const manager = new BulkheadManager();

// Create isolated pools for different services
const paymentBulkhead = manager.create('payment', {
  maxConcurrent: 5,
  maxQueue: 20,
  queueTimeout: 3000
});

const inventoryBulkhead = manager.create('inventory', {
  maxConcurrent: 10,
  maxQueue: 50
});

const notificationBulkhead = manager.create('notification', {
  maxConcurrent: 20,
  maxQueue: 100
});

// Wrap service functions
const processPayment = paymentBulkhead.wrap(async (orderId, amount) => {
  const response = await fetch('https://payment.api/charge', {
    method: 'POST',
    body: JSON.stringify({ orderId, amount })
  });
  return response.json();
});

const checkInventory = inventoryBulkhead.wrap(async (productId) => {
  const response = await fetch(`https://inventory.api/check/${productId}`);
  return response.json();
});

const sendNotification = notificationBulkhead.wrap(async (userId, message) => {
  await fetch('https://notification.api/send', {
    method: 'POST',
    body: JSON.stringify({ userId, message })
  });
});

// Handle order with isolated failures
async function handleOrder(order) {
  const results = {
    inventory: null,
    payment: null,
    notification: null
  };

  try {
    results.inventory = await checkInventory(order.productId);
  } catch (error) {
    if (error instanceof BulkheadRejectedException) {
      return { status: 'retry_later', service: 'inventory' };
    }
    throw error;
  }

  try {
    results.payment = await processPayment(order.id, order.amount);
  } catch (error) {
    if (error instanceof BulkheadRejectedException) {
      // Payment bulkhead full, but inventory still works
      return { status: 'retry_later', service: 'payment' };
    }
    throw error;
  }

  // Fire and forget notification
  sendNotification(order.userId, 'Order confirmed').catch(console.error);

  return { status: 'completed', ...results };
}

// Express middleware for bulkhead metrics
function bulkheadMetricsEndpoint(manager) {
  return (req, res) => {
    res.json(manager.getAllMetrics());
  };
}

// app.get('/metrics/bulkheads', bulkheadMetricsEndpoint(manager));

Key Takeaways

Concept	Remember
CAP Theorem	Pick 2 of 3: Consistency, Availability, Partition Tolerance
Consensus	Use Raft for leader election, state machine replication
Transactions	2PC for strong consistency, Sagas for microservices
Failures	Design for failure: retries, circuit breakers, bulkheads

Distributed systems are hard. Every network call can fail, be slow, or return stale data. Design for failure from day one.

Interview Questions

Q1: Explain the CAP theorem. Then tell me why most real-world system design decisions aren’t actually “pick two out of three.”

Strong Answer

The CAP theorem states that a distributed system can provide at most two of three guarantees simultaneously: Consistency (every read sees the most recent write), Availability (every request gets a non-error response), and Partition tolerance (the system continues to operate despite network partitions between nodes).

Why “pick two” is misleading: Network partitions aren’t optional — they will happen in any distributed system. So the real choice is between CP (consistency + partition tolerance) and AP (availability + partition tolerance). You’re never truly choosing to give up partition tolerance, because the network will partition whether you designed for it or not.
In practice, it’s a spectrum, not a binary: Systems like DynamoDB let you tune consistency per-request. A shopping cart read can be eventually consistent (AP), while a checkout that decrements inventory uses strongly consistent reads (CP). You’re making the CAP trade-off at the operation level, not the system level.
The PACELC extension is more useful: It says “if there’s a Partition, choose A or C; Else, choose Latency or Consistency.” This captures what engineers actually deal with day-to-day — even when the network is fine, strong consistency adds latency (because you’re waiting for replica acknowledgment). DynamoDB is PA/EL (available during partitions, low latency otherwise). Spanner is PC/EC (consistent always, pays the latency cost via TrueTime).
Real-world example: At a company running a global e-commerce platform, product catalog reads are AP (showing a slightly stale price for 2 seconds is fine), but the payment ledger is CP (an incorrect balance means real money loss). Same infrastructure, different CAP choices per use case.

Red flag answer: “You pick two out of three: consistency, availability, or partition tolerance” and stops there. This is the textbook recitation that every candidate gives. It shows memorization, not understanding. Even worse: “I’d pick CA” — this means the candidate doesn’t understand that partitions aren’t a choice. Follow-ups:

You’re designing a global inventory system for a flash sale with 100K concurrent users. Walk me through specifically where you’d choose CP vs AP within that single system.
Your team just adopted CockroachDB because “it’s CP and still highly available.” How is that possible, and what are you actually giving up compared to a system like Cassandra?

Q2: What is a vector clock, and why can’t you just use wall-clock timestamps to order events in a distributed system?

Strong Answer

A vector clock is a data structure that tracks causal ordering of events across multiple nodes. Each node maintains a vector of counters — one per node in the system — and increments its own counter on every local event. When messages are exchanged, nodes merge vectors by taking the element-wise maximum.

Why wall clocks fail: Physical clocks on different machines are never perfectly synchronized. Even with NTP, you get clock skew of 1-10ms between machines in the same datacenter, and much worse across regions. If Node A writes at T=100 and Node B writes at T=101, you can’t be sure A actually happened first — Node B’s clock might just be 2ms ahead. Google’s Spanner solves this with TrueTime (atomic clocks + GPS), but that’s a $100M+ infrastructure investment.
What vector clocks actually give you: They establish a partial order. If event A’s vector clock is strictly less than or equal to B’s on all components, A happened before B. If neither dominates, the events are concurrent — meaning there’s a genuine conflict that needs resolution. This is fundamentally different from timestamps, which impose a total order that might be wrong.
The practical trade-off: Vector clocks grow linearly with the number of nodes. In a system with 1,000 nodes, every piece of data carries a 1,000-element vector — that’s real overhead on every message and storage operation. This is why systems like Dynamo and Riak used them but with pruning strategies, and why many systems (Cassandra, for example) opted for Last-Write-Wins with timestamps instead, accepting the rare data loss for simplicity.
Real-world example: Amazon’s original Dynamo paper used vector clocks for their shopping cart. When concurrent updates happened (say, two users on the same account adding items simultaneously), the system could detect the conflict and merge the carts instead of silently dropping one write.

Red flag answer: “Just use timestamps” or “NTP keeps clocks in sync so this isn’t a problem.” This reveals the candidate hasn’t worked with distributed systems at scale, where clock skew causes subtle, hard-to-reproduce bugs — the kind that show up only under load or during a network hiccup. Follow-ups:

If vector clocks grow too large for a 500-node cluster, what alternatives exist? How do systems like Cassandra handle this differently?
Walk me through a concrete scenario where using Last-Write-Wins timestamps would silently lose data but vector clocks would catch the conflict.

Q3: Your system uses Raft for consensus across 5 nodes. The leader crashes. Walk me through exactly what happens next, step by step.

Strong Answer

When the Raft leader crashes, the cluster goes through a leader election process. Here’s the step-by-step sequence:

Step 1 — Detection via heartbeat timeout: Each follower has a randomized election timeout (typically 150-300ms). The leader sends heartbeats at a shorter interval (e.g., every 50-100ms). When a follower doesn’t receive a heartbeat within its timeout, it suspects the leader is dead.
Step 2 — A follower becomes a candidate: The first follower whose timeout expires increments its currentTerm, transitions to candidate state, votes for itself, and sends RequestVote RPCs to all other nodes. The randomized timeout is critical — it makes split votes unlikely because one node almost always times out first.
Step 3 — Voting: Each node votes for at most one candidate per term. A node grants its vote only if the candidate’s log is at least as up-to-date as the voter’s log (this is the “election safety” property). This guarantees the new leader has all committed entries.
Step 4 — Winning the election: The candidate needs a majority (3 out of 5 nodes). Once it gets 3 votes, it becomes the new leader and immediately sends heartbeats to assert authority and prevent other elections.
Step 5 — Uncommitted entries: Any log entries the old leader had replicated to fewer than 3 nodes are not committed and may be overwritten. Entries replicated to 3+ nodes are committed and will survive. This is why clients should only consider a write successful after the leader confirms it’s committed (replicated to a majority).
During the election, the system is unavailable for writes — typically 150-500ms of downtime. Reads can be served from followers if your system allows stale reads; otherwise reads also block.

Red flag answer: “A new leader is automatically selected” without being able to describe the mechanism. Also a red flag: not mentioning the safety property that prevents a node with a stale log from becoming leader — this is the entire point of Raft’s election restriction. Follow-ups:

What happens if the network partitions into a group of 2 and a group of 3, and the old leader is in the group of 2? Does the old leader know it’s no longer the leader?
You’re running Raft in production and noticing frequent leader elections even though no nodes are actually crashing. What’s causing this and how do you fix it?

Q4: Explain Two-Phase Commit. What is the fundamental flaw, and what alternatives exist for distributed transactions in a microservices architecture?

Strong Answer

Two-Phase Commit (2PC) is a protocol that ensures atomicity across multiple participants in a distributed transaction. Phase 1 (Prepare): the coordinator asks each participant “can you commit?” and each participant acquires locks, writes to a redo log, and votes yes or no. Phase 2 (Commit/Abort): if all voted yes, the coordinator tells everyone to commit; if any voted no, everyone aborts.

The fundamental flaw is the blocking problem: If the coordinator crashes after sending “prepare” but before sending the commit/abort decision, all participants are stuck holding locks indefinitely. They’ve voted yes, so they can’t unilaterally abort (another participant might have already committed). They can’t commit either (the coordinator might have decided to abort). This makes 2PC a blocking protocol — a single coordinator failure can freeze the entire system.
In a microservices world, 2PC is usually the wrong choice because: (1) it requires all services to be available simultaneously, defeating the independence that microservices are supposed to provide; (2) holding distributed locks across services creates latency and contention; (3) the coordinator is a single point of failure unless you add complexity like 3PC or Paxos-based commit.
The Saga pattern is the practical alternative: Instead of one big transaction, you decompose it into a sequence of local transactions. Each service does its work and publishes an event. If a step fails, you execute compensating transactions to undo previous steps. For example: Order Service creates order -> Payment Service charges card -> Inventory Service reserves stock. If inventory fails, you refund the payment and cancel the order.
Saga trade-offs: You lose isolation (intermediate states are visible) and compensating transactions can be complex (how do you “un-send” an email?). But you gain availability, loose coupling, and independent scalability — which matters more in microservices. Uber processes millions of trips daily using sagas, not 2PC.

Red flag answer: “Two-Phase Commit guarantees consistency across services, so use it for everything distributed.” This shows the candidate hasn’t felt the operational pain of distributed locking in production. Also a red flag: describing 2PC without mentioning the coordinator failure problem — that’s the one thing every interviewer expects you to know. Follow-ups:

You’re implementing a saga for an e-commerce checkout. The payment succeeds but the inventory reservation fails. Your compensating transaction to refund the payment also fails (the payment provider is down). What now?
How would you ensure exactly-once semantics in a saga when messages can be delivered more than once?

Q5: You’re running a distributed key-value store with 3 replicas. A client writes a value and immediately reads it back but gets stale data. Explain why this happens and how you’d fix it.

Strong Answer

This is the classic read-after-write consistency problem in eventually consistent systems. Here’s exactly what’s happening:

The write goes to one replica (or the leader) and returns success before all replicas are updated. If the subsequent read hits a different replica that hasn’t received the update yet, it returns stale data. With 3 replicas using async replication, the write might succeed on 1 node in 5ms while the other 2 nodes receive the update 50-200ms later.
Fix 1 — Quorum reads and writes (W + R > N): With N=3 replicas, set W=2 (write must succeed on 2 nodes) and R=2 (read must come from 2 nodes). Since W+R=4 > N=3, at least one node in the read set must have the latest write. The client picks the value with the highest version/timestamp. This is how DynamoDB’s strongly consistent reads work, and Cassandra’s QUORUM consistency level.
Fix 2 — Read-your-writes consistency (session-based): Route all of a user’s reads to the same replica that accepted their writes, typically using session affinity (sticky sessions by user ID). This is simpler than quorums and good enough for many use cases — the user sees their own writes even if other users see stale data briefly.
Fix 3 — Write to leader, read from leader: In a leader-follower setup, route both reads and writes to the leader for operations that need consistency. The downside: the leader becomes a bottleneck. This is why most systems only do this for specific critical paths, not all reads.
What I’d actually recommend depends on the use case: For a user profile update, read-your-writes (Fix 2) is sufficient and cheap. For an inventory decrement during checkout, quorum (Fix 1) is necessary. For a social media feed, you might just accept eventual consistency because the cost of stale reads is low.

Red flag answer: “Set everything to strong consistency” without discussing the latency and availability trade-offs. Also a red flag: not knowing the quorum formula W + R > N — this is bread-and-butter distributed systems knowledge. Follow-ups:

You’re using quorum reads/writes (W=2, R=2, N=3) and one replica goes down permanently. What happens to your read and write availability?
A customer reports that even with quorum reads, they’re occasionally seeing stale data. What could cause this, and how would you debug it?

Q6: What is the difference between the Saga pattern’s choreography and orchestration approaches? When would you choose one over the other?

Strong Answer

Both choreography and orchestration are ways to coordinate the steps of a saga (a sequence of local transactions with compensating actions). The difference is in who controls the flow.

Choreography (event-driven): Each service publishes domain events after completing its local transaction, and other services subscribe to those events. There’s no central controller — the flow emerges from event subscriptions. Order Service publishes OrderCreated, Payment Service hears it and charges the card, publishes PaymentCompleted, Inventory Service hears that and reserves stock. For compensation, services publish failure events: PaymentFailed triggers OrderCancelled.
Orchestration (command-driven): A central Saga Orchestrator sends explicit commands to each service in sequence. The orchestrator knows the full workflow and handles branching/compensation logic. “Step 1: tell Payment to charge. Step 2: if success, tell Inventory to reserve. If Payment failed, tell Order to cancel.”
Choose choreography when: You have 3-4 simple services, the flow is mostly linear, teams are autonomous and don’t want a central coordinator, and you already have an event bus (Kafka, SNS/SQS). It scales well organizationally because no single team owns the flow.
Choose orchestration when: The workflow has more than 5 steps, complex branching logic (if payment is partial, do X; if international, add customs step), you need clear visibility into saga state (which step are we on? where did it fail?), or you need to add/reorder steps without touching every service. Netflix uses orchestrators for their complex content workflows.
The hidden cost of choreography: At scale, the implicit flow becomes very hard to reason about. When something goes wrong, you’re tracing events across 6 services’ logs to reconstruct what happened. With an orchestrator, you look at one place. I’ve seen teams start with choreography for its elegance and migrate to orchestration after the third production incident where no one could figure out the sequence of events.

Red flag answer: “Always use choreography because it’s more decoupled” or “always use orchestration because it’s easier.” Both absolutes ignore the context. Also a red flag: not mentioning observability as a factor — in production, the ability to debug a failed saga is often the deciding factor, not theoretical coupling. Follow-ups:

You chose choreography and now have 8 services in the saga. A step in the middle fails and the compensating events create a cascade that takes 45 seconds to fully resolve. How do you improve this?
How do you handle the case where the orchestrator itself crashes mid-saga? How does the system recover?

Q7: What is the Circuit Breaker pattern, and how does it differ from simple retry logic? Describe a production scenario where retries would make the problem worse but a circuit breaker would help.

Strong Answer

A circuit breaker monitors calls to a downstream service and “opens” (stops sending traffic) when failures exceed a threshold. It has three states: Closed (normal — requests flow through), Open (tripped — requests fail immediately without calling the downstream), and Half-Open (testing — a limited number of requests are let through to check if the service has recovered).

How it differs from retries: Retries address transient failures — a single request that might succeed on the second attempt. A circuit breaker addresses sustained failures — when the downstream service is down or degraded and retrying will only make things worse. They’re complementary: retries inside a circuit breaker means “try a few times, but if the service is consistently failing, stop trying and fail fast.”
The production scenario where retries are destructive: Imagine your payment service is overloaded and responding with 503s at 2-second latency. You have 10 upstream servers each sending 100 requests/second, all with retry logic (3 retries). Without a circuit breaker, each failed request generates 3 more attempts. Your 1,000 req/s becomes 3,000 req/s to an already overloaded service. This is the retry storm (or “thundering herd”) — you’re pouring gasoline on the fire. A circuit breaker detects the failure rate, opens the circuit, and returns errors immediately to callers. The payment service gets breathing room to recover.
Key configuration decisions: The failure threshold (too low and you trip on transient errors; too high and you’re slow to react), the timeout duration (how long to stay open before trying again), and what counts as a failure (5xx responses yes, 4xx responses no — a 400 Bad Request means the client sent garbage, not that the service is broken).
What most people miss — the fallback strategy: The circuit breaker is only half the pattern. The other half is what you do when it’s open. For a payment service: queue the charge for later processing. For a recommendation engine: show popular items instead. For a search service: return cached results. The fallback is where the real engineering judgment lives.

Red flag answer: “Just retry with exponential backoff and it’ll be fine.” This works for transient errors but is exactly wrong for sustained outages — you need to stop sending traffic, not send it more slowly. Also a red flag: describing the pattern without mentioning the half-open state — that’s the recovery mechanism that makes the pattern work. Follow-ups:

You have circuit breakers on 5 downstream services. All 5 trip simultaneously at 3 AM. What’s your incident response playbook, and what does this correlated failure tell you about the root cause?
Your circuit breaker is flapping — rapidly alternating between open and closed. What’s causing this and how do you tune it?

Q8: Explain the Bulkhead pattern. How does it complement the Circuit Breaker pattern, and when would you use both together?

Strong Answer

The Bulkhead pattern isolates different parts of your system into separate resource pools so that a failure in one doesn’t cascade to others. The name comes from ship design — bulkheads divide a ship’s hull into watertight compartments so a breach in one section doesn’t sink the entire vessel.

Concrete implementation: You create separate thread pools (or connection pools, or semaphore pools) for each downstream dependency. Your payment service gets 5 concurrent slots, inventory gets 10, notifications get 20. If the payment service hangs, only those 5 threads are blocked — the other 30 threads continue serving inventory and notification requests normally. Without a bulkhead, all 35 threads could pile up waiting for the slow payment service, and your entire application becomes unresponsive.
How it complements Circuit Breaker: A circuit breaker stops calling a failing service. A bulkhead limits the damage while the failure is being detected. They operate on different timescales: the bulkhead provides immediate isolation (thread pool exhaustion is prevented from second one), while the circuit breaker kicks in after enough failures accumulate (maybe 10-30 seconds). Together, the bulkhead contains the blast radius while the circuit breaker stops the bleeding.
When to use both together: Almost always, in production systems with multiple downstream dependencies. Example: your API gateway talks to 4 microservices. Each gets its own bulkhead (connection pool with a max size and a queue timeout). Each also gets its own circuit breaker. If the recommendation service goes down, its bulkhead prevents the 50 pending requests from consuming all your gateway threads, and its circuit breaker prevents new requests from even trying after the failure threshold is hit. Meanwhile, search, auth, and cart services continue serving traffic without any impact.
The configuration challenge: Setting bulkhead sizes requires understanding your traffic patterns. Too small and you reject legitimate requests during peak load. Too large and the isolation is ineffective. In practice, you start with per-service p99 latency times expected concurrency, add 20% headroom, and tune based on production metrics. Netflix’s Hystrix library popularized this pattern, and Resilience4j carries it forward in modern Java.

Red flag answer: “Just use a big thread pool for everything” — this is exactly the anti-pattern that bulkheads solve. Also a red flag: only describing the pattern in theory without mentioning pool sizing, queue overflow behavior, or how it interacts with circuit breakers in a real deployment. Follow-ups:

You’ve set up bulkheads per-service, but a bug in your API gateway layer means all services share the same outbound HTTP connection pool, bypassing the bulkheads. How would you detect this in production?
Your payment bulkhead is set to 5 concurrent slots. During a Black Friday traffic spike, legitimate payment requests are being rejected because the pool is full. How do you handle this without removing the isolation benefit?

Q9: A distributed system you operate is experiencing a “split brain” scenario. What does this mean, what damage can it cause, and how do you resolve it?

Strong Answer

Split brain occurs when a network partition causes a distributed system to divide into two or more groups, each believing it’s the “real” cluster. Both sides continue operating independently — accepting writes, serving reads, potentially electing their own leaders — without knowing the other side is doing the same thing.

The damage is real and specific: In a database cluster, split brain means two primary nodes both accepting writes. Client A writes balance = 100 to Primary-Left while Client B writes balance = 200 to Primary-Right. When the partition heals, you have conflicting data with no way to automatically determine which is correct. In the worst case, both sides committed financial transactions with different balances — reconciliation is a manual nightmare. This happened to GitHub in 2018 when a brief network partition caused their MySQL primary to split, leading to data inconsistency that took hours to resolve.
Prevention mechanisms: (1) Quorum-based systems (Raft, ZooKeeper) prevent split brain by design — a leader needs a majority to commit, so only one side of a partition can have a quorum. With 5 nodes split 2-3, only the group of 3 can elect a leader. (2) Fencing tokens (also called “epoch numbers”) — when a new leader is elected, it gets a monotonically increasing token. Storage systems reject writes from tokens lower than the latest they’ve seen, so a stale leader’s writes are rejected. (3) STONITH (Shoot The Other Node In The Head) — in traditional HA systems, when you suspect split brain, you forcefully power off the other node via out-of-band management (IPMI/iLO) before promoting yourself. Brutal, but effective.
Resolution after the fact: If split brain already occurred, you need to: identify the divergence point (last common write), choose a winning side (usually the one with more confirmed transactions), replay the losing side’s writes as conflicts for manual resolution, and alert the operations team. This is why many systems prefer to become unavailable (CP) rather than risk split brain (AP with no conflict resolution).

Red flag answer: “Just run more replicas” — this doesn’t solve split brain; it can actually make it worse if you don’t have proper quorum rules. Also a red flag: not being able to name a concrete prevention mechanism (fencing tokens, quorum, STONITH) — it suggests the candidate has only read about split brain but never operated a system where it could happen. Follow-ups:

You’re running a 2-node database cluster for cost reasons (no quorum possible). How do you prevent split brain without adding a third node?
Your ZooKeeper ensemble of 5 nodes is split 2-2-1 across three network segments. What happens? Can any side make progress?

Q10: You’re designing a system that needs to process each message exactly once, but your message broker provides at-least-once delivery. How do you achieve exactly-once semantics?

Strong Answer

True exactly-once delivery is impossible in a distributed system (proven by the Two Generals Problem). What we actually implement is “effectively exactly-once” processing by combining at-least-once delivery with idempotent processing on the consumer side.

The core pattern — idempotency keys: Every message gets a unique identifier (UUID, or a natural key like order_id + action). The consumer maintains an idempotency store (a table or cache of processed message IDs). Before processing, check: “Have I seen this ID before?” If yes, skip or return the cached result. If no, process and record the ID atomically.
The atomic part is critical: The message processing and the idempotency record must be in the same transaction. If you process the message, then crash before recording the ID, the message will be redelivered and reprocessed. The pattern is: BEGIN TRANSACTION; INSERT INTO idempotency_log (message_id); do_the_work(); COMMIT;. If the work involves a different datastore than the idempotency log, you need the Outbox Pattern — write the idempotency record and a “pending work” record to the same DB in one transaction, then a separate worker picks up and forwards the pending work.
Idempotency store considerations: Use a fast lookup store (Redis with TTL, or a database table with an index on message_id). Set a TTL on entries — you don’t need to remember every message forever, just long enough to cover the broker’s redelivery window (typically 7 days for SQS, configurable for Kafka). At high throughput (100K+ messages/sec), the idempotency lookup itself becomes a bottleneck — use bloom filters as a first pass to avoid hitting the database for messages you’ve definitely never seen.
Kafka’s approach: Kafka 0.11+ added idempotent producers (deduplication using sequence numbers per producer-partition pair) and transactional producers (atomic writes across multiple partitions). This gets you exactly-once within Kafka, but the moment data leaves Kafka and hits your database, you’re back to needing application-level idempotency.

Red flag answer: “Our message broker supports exactly-once delivery” — no broker truly does end-to-end; they provide it within their system boundary, but you still need idempotency at the application layer. Also a red flag: “Just use a database unique constraint on the message ID” without discussing the transactional atomicity requirement — a unique constraint prevents duplicate inserts but doesn’t prevent duplicate side effects. Follow-ups:

Your idempotency store is in Redis and the Redis node crashes, losing all keys. Messages are redelivered and reprocessed. How do you make the system resilient to this?
You’re processing 200K messages/second. Your idempotency table has 500 million rows and lookups are getting slow. How do you optimize this without losing the exactly-once guarantee?

​What are Distributed Systems?

​Why Distributed?

Scalability

Reliability

Latency

Compliance

​The Eight Fallacies

​Consistency Models

​Strong Consistency

​Eventual Consistency

​Consistency Implementation Examples

​Consistency Levels

​Consensus Algorithms

​Paxos (Simplified)

​Raft (Easier to Understand)

​Raft Implementation

​Distributed Transactions

​Two-Phase Commit (2PC)

​Two-Phase Commit Implementation

​Saga Pattern

​Choreography vs Orchestration

​Handling Failures

​Circuit Breaker Pattern

​Retry with Exponential Backoff

​Bulkhead Pattern

​Key Takeaways

​Interview Questions

​Q1: Explain the CAP theorem. Then tell me why most real-world system design decisions aren’t actually “pick two out of three.”

​Q2: What is a vector clock, and why can’t you just use wall-clock timestamps to order events in a distributed system?

​Q3: Your system uses Raft for consensus across 5 nodes. The leader crashes. Walk me through exactly what happens next, step by step.

​Q4: Explain Two-Phase Commit. What is the fundamental flaw, and what alternatives exist for distributed transactions in a microservices architecture?

​Q5: You’re running a distributed key-value store with 3 replicas. A client writes a value and immediately reads it back but gets stale data. Explain why this happens and how you’d fix it.

​Q6: What is the difference between the Saga pattern’s choreography and orchestration approaches? When would you choose one over the other?

​Q7: What is the Circuit Breaker pattern, and how does it differ from simple retry logic? Describe a production scenario where retries would make the problem worse but a circuit breaker would help.

​Q8: Explain the Bulkhead pattern. How does it complement the Circuit Breaker pattern, and when would you use both together?

​Q9: A distributed system you operate is experiencing a “split brain” scenario. What does this mean, what damage can it cause, and how do you resolve it?

​Q10: You’re designing a system that needs to process each message exactly once, but your message broker provides at-least-once delivery. How do you achieve exactly-once semantics?

What are Distributed Systems?

Why Distributed?

The Eight Fallacies

Consistency Models

Strong Consistency

Eventual Consistency

Consistency Implementation Examples

Consistency Levels

Consensus Algorithms

Paxos (Simplified)

Raft (Easier to Understand)

Raft Implementation

Distributed Transactions

Two-Phase Commit (2PC)

Two-Phase Commit Implementation

Saga Pattern

Choreography vs Orchestration

Handling Failures

Circuit Breaker Pattern

Retry with Exponential Backoff

Bulkhead Pattern

Key Takeaways

Interview Questions

Q1: Explain the CAP theorem. Then tell me why most real-world system design decisions aren’t actually “pick two out of three.”

Q2: What is a vector clock, and why can’t you just use wall-clock timestamps to order events in a distributed system?

Q3: Your system uses Raft for consensus across 5 nodes. The leader crashes. Walk me through exactly what happens next, step by step.

Q4: Explain Two-Phase Commit. What is the fundamental flaw, and what alternatives exist for distributed transactions in a microservices architecture?

Q5: You’re running a distributed key-value store with 3 replicas. A client writes a value and immediately reads it back but gets stale data. Explain why this happens and how you’d fix it.

Q6: What is the difference between the Saga pattern’s choreography and orchestration approaches? When would you choose one over the other?

Q7: What is the Circuit Breaker pattern, and how does it differ from simple retry logic? Describe a production scenario where retries would make the problem worse but a circuit breaker would help.

Q8: Explain the Bulkhead pattern. How does it complement the Circuit Breaker pattern, and when would you use both together?

Q9: A distributed system you operate is experiencing a “split brain” scenario. What does this mean, what damage can it cause, and how do you resolve it?

Q10: You’re designing a system that needs to process each message exactly once, but your message broker provides at-least-once delivery. How do you achieve exactly-once semantics?