Distributed Systems Guide

Distributed Systems Guide

A skill for researching and designing distributed systems, covering consensus algorithms, replication strategies, consistency models, fault tolerance, and performance analysis. Provides theoretical foundations and practical implementations relevant to systems research.

Consistency Models

Consistency Hierarchy

Strongest
  |  Linearizability (atomic, real-time ordering)
  |  Sequential consistency (program order respected)
  |  Causal consistency (causally related ops ordered)
  |  PRAM / FIFO consistency (per-process order)
  |  Eventual consistency (converges if updates stop)
Weakest

CAP Theorem and PACELC

The CAP theorem states that during a network partition, a distributed system must choose between consistency and availability:

Consensus Algorithms

Raft Implementation Sketch

from enum import Enum
from dataclasses import dataclass, field
import random

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

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

@dataclass
class RaftNode:
    """
    Simplified Raft consensus node for educational purposes.
    Implements leader election and log replication state machine.
    """
    node_id: str
    state: NodeState = NodeState.FOLLOWER
    current_term: int = 0
    voted_for: str = None
    log: list = field(default_factory=list)
    commit_index: int = 0
    last_applied: int = 0

    # Leader state
    next_index: dict = field(default_factory=dict)
    match_index: dict = field(default_factory=dict)

    def start_election(self, peers: list[str]) -> dict:
        """Transition to candidate and request votes."""
        self.state = NodeState.CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id

        last_log_index = len(self.log) - 1 if self.log else -1
        last_log_term = self.log[-1].term if self.log else 0

        return {
            "type": "RequestVote",
            "term": self.current_term,
            "candidate_id": self.node_id,
            "last_log_index": last_log_index,
            "last_log_term": last_log_term,
        }

    def handle_vote_request(self, term: int, candidate_id: str,
                              last_log_index: int,
                              last_log_term: int) -> dict:
        """Process a RequestVote RPC."""
        if term < self.current_term:
            return {"term": self.current_term, "vote_granted": False}

        if term > self.current_term:
            self.current_term = term
            self.state = NodeState.FOLLOWER
            self.voted_for = None

        # Check if candidate's log is at least as up-to-date
        my_last_term = self.log[-1].term if self.log else 0
        my_last_index = len(self.log) - 1 if self.log else -1

        log_ok = (last_log_term > my_last_term or
                  (last_log_term == my_last_term and
                   last_log_index >= my_last_index))

        vote_granted = (
            (self.voted_for is None or self.voted_for == candidate_id)
            and log_ok
        )

        if vote_granted:
            self.voted_for = candidate_id

        return {"term": self.current_term, "vote_granted": vote_granted}

    def append_entry(self, command: str) -> LogEntry:
        """Leader appends a new entry to its log."""
        entry = LogEntry(
            term=self.current_term,
            index=len(self.log),
            command=command,
        )
        self.log.append(entry)
        return entry

Paxos vs Raft vs PBFT Comparison

Replication Strategies

State Machine Replication

class ReplicatedStateMachine:
    """
    State machine replication with configurable consistency.
    Demonstrates read/write quorum intersection for correctness.
    """

    def __init__(self, n_replicas: int, read_quorum: int = None,
                 write_quorum: int = None):
        self.n = n_replicas
        self.R = read_quorum or (n_replicas // 2 + 1)
        self.W = write_quorum or (n_replicas // 2 + 1)

        # Quorum intersection guarantees: R + W > N
        assert self.R + self.W > self.n, (
            f"Quorum intersection violated: R({self.R}) + W({self.W}) "
            f"must be > N({self.n})"
        )

        self.replicas = [{} for _ in range(n_replicas)]
        self.version_clock = 0

    def write(self, key: str, value: str) -> dict:
        """Write to W replicas."""
        self.version_clock += 1
        # Select W replicas (in practice, based on availability)
        targets = random.sample(range(self.n), self.W)
        for i in targets:
            self.replicas[i][key] = (value, self.version_clock)

        return {
            "key": key,
            "version": self.version_clock,
            "acked_by": len(targets),
            "quorum_met": True,
        }

    def read(self, key: str) -> dict:
        """Read from R replicas, return latest version."""
        targets = random.sample(range(self.n), self.R)
        responses = []
        for i in targets:
            if key in self.replicas[i]:
                responses.append(self.replicas[i][key])

        if not responses:
            return {"key": key, "value": None, "found": False}

        # Return the value with the highest version
        latest = max(responses, key=lambda x: x[1])
        return {
            "key": key,
            "value": latest[0],
            "version": latest[1],
            "found": True,
        }

Clock Synchronization and Ordering

Vector Clocks

class VectorClock:
    """Vector clock for tracking causality in distributed systems."""

    def __init__(self, process_id: str, processes: list[str]):
        self.pid = process_id
        self.clock = {p: 0 for p in processes}

    def increment(self):
        """Local event: increment own counter."""
        self.clock[self.pid] += 1

    def send(self) -> dict:
        """Prepare clock for sending with a message."""
        self.increment()
        return dict(self.clock)

    def receive(self, other_clock: dict):
        """Merge received clock: element-wise max, then increment."""
        for p in self.clock:
            self.clock[p] = max(self.clock[p], other_clock.get(p, 0))
        self.increment()

    def happened_before(self, other: dict) -> bool:
        """Check if this clock happened-before other (causal ordering)."""
        return (all(self.clock[p] <= other.get(p, 0) for p in self.clock) and
                any(self.clock[p] < other.get(p, 0) for p in self.clock))

Performance Analysis

Latency and Throughput Modeling

Key metrics for evaluating distributed systems:

• Tail latency (p99, p999): Critical for real-world SLAs; often dominated by slow replicas
• Throughput under contention: How performance degrades with conflict rate
• Scalability: Linear vs sub-linear throughput increase with added nodes
• Recovery time: Time to restore consistency after node failure

Key Research Papers

• Lamport, L. (1998). The Part-Time Parliament (Paxos). ACM TOCS.
• Ongaro, D. and Ousterhout, J. (2014). In Search of an Understandable Consensus Algorithm (Raft). USENIX ATC.
• Corbett, J. et al. (2013). Spanner: Google's Globally-Distributed Database. ACM TOCS.
• DeCandia, G. et al. (2007). Dynamo: Amazon's Highly Available Key-value Store. SOSP.

Tools and Frameworks

• etcd / ZooKeeper: Production consensus stores for coordination
• Jepsen: Distributed systems correctness testing framework
• TLA+ / PlusCal: Formal specification and model checking
• ns-3 / OMNeT++: Network simulation for distributed protocols
• gRPC / Cap'n Proto: High-performance RPC frameworks
• FoundationDB: Multi-model distributed database with strong consistency