🌐 Distributed Systems — Complete Guide

A distributed system is a collection of independent computers that appear to the user as a single coherent system. They communicate via a network to coordinate their work.

---

1. Core Architecture Overview

---

2. The 8 Fallacies of Distributed Systems

CAUTION

These are common wrong assumptions developers make. Ignoring them causes production failures.

#	Fallacy	Reality
1	The network is reliable	Packets drop, connections time out
2	Latency is zero	Every network hop adds delay
3	Bandwidth is infinite	You can saturate links
4	The network is secure	Assume adversarial conditions
5	Topology doesn't change	IPs change, nodes go down
6	There is one administrator	Many teams, many configs
7	Transport cost is zero	Serialization + bandwidth costs money
8	The network is homogeneous	Mixed OS, languages, protocols

---

3. CAP Theorem

IMPORTANT

In a real distributed system, network partitions will happen. So you must always choose between CP or AP — never CA.

---

4. Key Patterns with Code Examples

4.1 Service Discovery

# service_registry.py — Simple in-memory service registry
import random
from typing import List, Dict
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class ServiceInstance:
    host: str
    port: int
    service_name: str
    last_heartbeat: datetime = field(default_factory=datetime.now)
    healthy: bool = True

class ServiceRegistry:
    """Consul/Eureka-style service registry."""

    def __init__(self, ttl_seconds: int = 30):
        self._registry: Dict[str, List[ServiceInstance]] = {}
        self._ttl = timedelta(seconds=ttl_seconds)

    def register(self, instance: ServiceInstance) -> None:
        """Service registers itself on startup."""
        name = instance.service_name
        if name not in self._registry:
            self._registry[name] = []
        # Avoid duplicate registration
        self._registry[name] = [
            i for i in self._registry[name]
            if not (i.host == instance.host and i.port == instance.port)
        ]
        self._registry[name].append(instance)
        print(f"✅ Registered {name} at {instance.host}:{instance.port}")

    def deregister(self, service_name: str, host: str, port: int) -> None:
        """Service deregisters on graceful shutdown."""
        if service_name in self._registry:
            self._registry[service_name] = [
                i for i in self._registry[service_name]
                if not (i.host == host and i.port == port)
            ]

    def heartbeat(self, service_name: str, host: str, port: int) -> None:
        """Keep-alive ping every N seconds."""
        for instance in self._registry.get(service_name, []):
            if instance.host == host and instance.port == port:
                instance.last_heartbeat = datetime.now()
                instance.healthy = True

    def get_instance(self, service_name: str) -> ServiceInstance | None:
        """Client-side load balancing — Round Robin / Random."""
        self._evict_stale()
        instances = [
            i for i in self._registry.get(service_name, [])
            if i.healthy
        ]
        return random.choice(instances) if instances else None

    def _evict_stale(self) -> None:
        """Remove instances that missed their heartbeat."""
        now = datetime.now()
        for name, instances in self._registry.items():
            for inst in instances:
                if now - inst.last_heartbeat > self._ttl:
                    inst.healthy = False
                    print(f"⚠️  {name} at {inst.host}:{inst.port} marked unhealthy")


# --- Usage ---
registry = ServiceRegistry(ttl_seconds=30)

# Payment service instances register themselves
registry.register(ServiceInstance("10.0.0.1", 8080, "payment-svc"))
registry.register(ServiceInstance("10.0.0.2", 8080, "payment-svc"))
registry.register(ServiceInstance("10.0.0.3", 8080, "payment-svc"))

# Order service discovers payment service
instance = registry.get_instance("payment-svc")
print(f"📡 Connecting to payment-svc at {instance.host}:{instance.port}")

---

4.2 Circuit Breaker

Prevents cascading failures by stopping requests to a failing service.

# circuit_breaker.py
import time
import functools
from enum import Enum
from threading import Lock
from typing import Callable, Any

class CircuitState(Enum):
    CLOSED   = "CLOSED"    # Normal — requests pass through
    OPEN     = "OPEN"      # Tripped — fail fast
    HALF_OPEN = "HALF_OPEN" # Testing recovery

class CircuitBreakerOpenError(Exception):
    """Raised when the circuit is open and requests are blocked."""
    pass

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        expected_exception: type = Exception,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception

        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._last_failure_time: float = 0
        self._lock = Lock()

    @property
    def state(self) -> CircuitState:
        with self._lock:
            if self._state == CircuitState.OPEN:
                # Check if recovery timeout has elapsed
                if time.monotonic() - self._last_failure_time >= self.recovery_timeout:
                    self._state = CircuitState.HALF_OPEN
                    print("🟡 Circuit HALF-OPEN — sending probe request")
            return self._state

    def call(self, func: Callable, *args, **kwargs) -> Any:
        state = self.state

        if state == CircuitState.OPEN:
            raise CircuitBreakerOpenError(
                f"⛔ Circuit OPEN — {func.__name__} blocked. "
                f"Retry in {self.recovery_timeout}s"
            )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise e

    def _on_success(self):
        with self._lock:
            self._failure_count = 0
            if self._state == CircuitState.HALF_OPEN:
                self._state = CircuitState.CLOSED
                print("🟢 Circuit CLOSED — service recovered!")

    def _on_failure(self):
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = time.monotonic()
            print(f"❌ Failure {self._failure_count}/{self.failure_threshold}")
            if self._failure_count >= self.failure_threshold:
                self._state = CircuitState.OPEN
                print("🔴 Circuit OPEN — blocking all requests!")


# --- Decorator usage ---
payment_circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=10.0)

def charge_customer(user_id: str, amount: float) -> dict:
    """Simulated payment API call."""
    # Simulate a failing downstream service
    raise ConnectionError("Payment gateway timeout")

def safe_charge(user_id: str, amount: float):
    try:
        return payment_circuit.call(charge_customer, user_id, amount)
    except CircuitBreakerOpenError as e:
        # Return cached response or graceful fallback
        print(f"Fallback: {e}")
        return {"status": "pending", "message": "Payment queued for retry"}
    except ConnectionError:
        return {"status": "error", "message": "Payment failed"}

# Test it
for i in range(6):
    result = safe_charge("user_123", 99.99)
    print(f"Attempt {i+1}: {result}\n")

---

4.3 Event-Driven Architecture with Kafka

# event_bus.py — Kafka-like pub/sub event bus
import json
import time
import threading
from collections import defaultdict, deque
from dataclasses import dataclass, asdict
from typing import Callable, Dict, List, Any
from uuid import uuid4

@dataclass
class Event:
    topic: str
    payload: dict
    event_id: str = ""
    timestamp: float = 0.0

    def __post_init__(self):
        if not self.event_id:
            self.event_id = str(uuid4())
        if not self.timestamp:
            self.timestamp = time.time()

ConsumerHandler = Callable[[Event], None]

class EventBus:
    """
    Simplified in-process event bus.
    In production: replace with confluent-kafka or aiokafka.
    """

    def __init__(self):
        # topic -> list of (group_id, handler)
        self._subscribers: Dict[str, List[tuple]] = defaultdict(list)
        # group_id -> offset per topic (for exactly-once delivery simulation)
        self._offsets: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
        self._topic_queues: Dict[str, deque] = defaultdict(deque)
        self._lock = threading.RLock()

    def publish(self, event: Event) -> None:
        """Publish an event to a topic."""
        with self._lock:
            self._topic_queues[event.topic].append(event)
            print(f"📤 Published [{event.topic}] event_id={event.event_id[:8]}")

        # Notify all consumer groups
        self._dispatch(event.topic)

    def subscribe(self, topic: str, group_id: str, handler: ConsumerHandler) -> None:
        """Subscribe to a topic as part of a consumer group."""
        with self._lock:
            self._subscribers[topic].append((group_id, handler))
            print(f"📥 {group_id} subscribed to [{topic}]")

    def _dispatch(self, topic: str) -> None:
        """Dispatch events to all consumer groups in parallel."""
        seen_groups = set()
        for group_id, handler in self._subscribers.get(topic, []):
            if group_id not in seen_groups:
                seen_groups.add(group_id)
                # Each consumer group gets its own thread (simulates parallel processing)
                t = threading.Thread(
                    target=self._consume,
                    args=(topic, group_id, handler),
                    daemon=True
                )
                t.start()

    def _consume(self, topic: str, group_id: str, handler: ConsumerHandler) -> None:
        """Consume events from a topic, tracking offset per consumer group."""
        with self._lock:
            queue = self._topic_queues[topic]
            offset = self._offsets[group_id][topic]
            events_to_process = list(queue)[offset:]
            self._offsets[group_id][topic] = len(queue)

        for event in events_to_process:
            try:
                handler(event)
            except Exception as e:
                print(f"💥 [{group_id}] failed to process {event.event_id[:8]}: {e}")
                # In real Kafka: send to Dead Letter Queue


# =======================
# Define our microservices
# =======================

bus = EventBus()

def inventory_handler(event: Event):
    """Inventory service: reserve stock when order is created."""
    order = event.payload
    print(f"📦 [Inventory] Reserving {order['quantity']}x {order['product_id']}")

def payment_handler(event: Event):
    """Payment service: charge customer when order is created."""
    order = event.payload
    print(f"💳 [Payment] Charging ${order['amount']} to user {order['user_id']}")

def notification_handler(event: Event):
    """Notification service: send confirmation email."""
    order = event.payload
    print(f"📧 [Notification] Sending confirmation to {order.get('email', 'user@example.com')}")

def analytics_handler(event: Event):
    """Analytics service: track order metrics."""
    print(f"📊 [Analytics] Recording order event for dashboard")

# Wire up subscriptions
bus.subscribe("order.created", "inventory-group",    inventory_handler)
bus.subscribe("order.created", "payment-group",      payment_handler)
bus.subscribe("order.created", "notification-group", notification_handler)
bus.subscribe("order.created", "analytics-group",    analytics_handler)

# Order service publishes an event
order_event = Event(
    topic="order.created",
    payload={
        "order_id": "ord_9921",
        "user_id": "user_123",
        "product_id": "SKU-LAPTOP-PRO",
        "quantity": 1,
        "amount": 1299.99,
        "email": "alice@example.com",
    }
)

bus.publish(order_event)
time.sleep(0.1)  # Allow async handlers to complete

Output:

📤 Published [order.created] event_id=9b4f12a3
📦 [Inventory] Reserving 1x SKU-LAPTOP-PRO
💳 [Payment] Charging $1299.99 to user user_123
📧 [Notification] Sending confirmation to alice@example.com
📊 [Analytics] Recording order event for dashboard

---

4.4 Distributed Consensus — Raft Algorithm

---

4.5 Consistent Hashing (for Distributed Cache / Sharding)

# consistent_hash.py
import hashlib
from bisect import bisect_right, insort
from typing import Optional

class ConsistentHashRing:
    """
    A consistent hash ring for distributing keys across nodes.
    Virtual nodes reduce key imbalance when nodes are added/removed.
    """

    def __init__(self, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self._ring: list[int] = []               # sorted hash positions
        self._hash_to_node: dict[int, str] = {}  # hash → node name

    def _hash(self, key: str) -> int:
        """Deterministic 32-bit hash of a string."""
        return int(hashlib.md5(key.encode()).hexdigest(), 16) % (2**32)

    def add_node(self, node: str) -> None:
        """Add a node (with virtual replicas) to the ring."""
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}#vnode{i}"
            h = self._hash(virtual_key)
            insort(self._ring, h)
            self._hash_to_node[h] = node
        print(f"✅ Added node '{node}' with {self.virtual_nodes} virtual nodes")

    def remove_node(self, node: str) -> None:
        """Remove a node and its virtual replicas."""
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}#vnode{i}"
            h = self._hash(virtual_key)
            self._ring.remove(h)
            del self._hash_to_node[h]
        print(f"🗑️  Removed node '{node}'")

    def get_node(self, key: str) -> Optional[str]:
        """Route a key to its responsible node."""
        if not self._ring:
            return None
        h = self._hash(key)
        # Find the first node position >= h (clockwise on the ring)
        idx = bisect_right(self._ring, h) % len(self._ring)
        return self._hash_to_node[self._ring[idx]]

    def get_distribution(self, keys: list[str]) -> dict[str, int]:
        """Show how keys are distributed across nodes."""
        distribution: dict[str, int] = {}
        for key in keys:
            node = self.get_node(key)
            distribution[node] = distribution.get(node, 0) + 1
        return distribution


# --- Demo ---
ring = ConsistentHashRing(virtual_nodes=150)
ring.add_node("node-A")
ring.add_node("node-B")
ring.add_node("node-C")
ring.add_node("node-D")

# Route some keys
test_keys = [f"user_{i}" for i in range(1000)]
print("\n📊 Key distribution (1000 keys across 4 nodes):")
dist = ring.get_distribution(test_keys)
for node, count in sorted(dist.items()):
    bar = "█" * (count // 10)
    print(f"  {node}: {count:4d} keys  {bar}")

# Add a new node — minimal key redistribution!
print("\n➕ Adding node-E...")
ring.add_node("node-E")
new_dist = ring.get_distribution(test_keys)
print("\n📊 New distribution after adding node-E:")
for node, count in sorted(new_dist.items()):
    bar = "█" * (count // 10)
    print(f"  {node}: {count:4d} keys  {bar}")

---

4.6 Saga Pattern (Distributed Transactions)

NOTE

In microservices, you can't use database transactions across services. The Saga Pattern breaks a transaction into local steps with compensating rollbacks.

# saga.py — Choreography-based Saga with compensating transactions
from dataclasses import dataclass, field
from typing import Callable, List, Tuple
from enum import Enum

class SagaStatus(Enum):
    PENDING    = "PENDING"
    SUCCESS    = "SUCCESS"
    ROLLING_BACK = "ROLLING_BACK"
    FAILED     = "FAILED"

@dataclass
class SagaStep:
    name: str
    action: Callable        # Forward action
    compensate: Callable    # Rollback action

@dataclass
class SagaResult:
    success: bool
    completed_steps: List[str]
    error: str = ""
    status: SagaStatus = SagaStatus.PENDING

class SagaOrchestrator:
    """
    Executes a sequence of steps and automatically
    rolls back on failure using compensating transactions.
    """

    def __init__(self, name: str, steps: List[SagaStep]):
        self.name = name
        self.steps = steps

    def execute(self, context: dict) -> SagaResult:
        completed: List[SagaStep] = []

        print(f"\n🚀 Starting Saga: {self.name}")
        print("─" * 50)

        for step in self.steps:
            try:
                print(f"  ▶️  [{step.name}] Executing...")
                step.action(context)
                completed.append(step)
                print(f"  ✅ [{step.name}] Success")
            except Exception as e:
                print(f"  ❌ [{step.name}] Failed: {e}")
                print(f"\n⏪ Rolling back {len(completed)} completed step(s)...")

                # Compensate in reverse order
                for done_step in reversed(completed):
                    try:
                        print(f"  ↩️  [{done_step.name}] Compensating...")
                        done_step.compensate(context)
                        print(f"  ✅ [{done_step.name}] Compensated")
                    except Exception as comp_err:
                        print(f"  💥 [{done_step.name}] Compensation FAILED: {comp_err}")
                        # Log to outbox for manual recovery

                return SagaResult(
                    success=False,
                    completed_steps=[s.name for s in completed],
                    error=str(e),
                    status=SagaStatus.FAILED
                )

        return SagaResult(
            success=True,
            completed_steps=[s.name for s in self.steps],
            status=SagaStatus.SUCCESS
        )


# --- Define steps for "Create Order" saga ---
def reserve_payment(ctx):
    ctx["payment_id"] = "pay_abc123"
    print(f"    💳 Reserved ${ctx['amount']} → payment_id={ctx['payment_id']}")

def refund_payment(ctx):
    print(f"    💳 Refunded payment_id={ctx.get('payment_id', 'N/A')}")

def reserve_inventory(ctx):
    ctx["reservation_id"] = "res_xyz789"
    print(f"    📦 Reserved {ctx['quantity']}x {ctx['product_id']}")

def release_inventory(ctx):
    print(f"    📦 Released reservation_id={ctx.get('reservation_id', 'N/A')}")

def create_shipment(ctx):
    # Simulate failure
    raise RuntimeError("No carrier available for region US-WEST-3")

def cancel_shipment(ctx):
    print(f"    🚚 Cancelled shipment (if any)")


order_saga = SagaOrchestrator(
    name="CreateOrder",
    steps=[
        SagaStep("ReservePayment",   reserve_payment,   refund_payment),
        SagaStep("ReserveInventory", reserve_inventory, release_inventory),
        SagaStep("CreateShipment",   create_shipment,   cancel_shipment),
    ]
)

result = order_saga.execute({
    "order_id": "ord_001",
    "user_id": "user_123",
    "product_id": "SKU-LAPTOP",
    "quantity": 1,
    "amount": 1299.99,
})

print(f"\n📋 Saga Result: {result.status.value}")

---

5. Observability: The Three Pillars

# observability.py — Structured logging with trace IDs
import json
import time
import uuid
import contextlib
from contextvars import ContextVar

# Thread-local trace context
_trace_id: ContextVar[str] = ContextVar("trace_id", default="")
_span_id: ContextVar[str] = ContextVar("span_id", default="")

def new_trace_id() -> str:
    return uuid.uuid4().hex[:16]

def new_span_id() -> str:
    return uuid.uuid4().hex[:8]

class StructuredLogger:
    def __init__(self, service_name: str):
        self.service = service_name

    def _log(self, level: str, message: str, **kwargs):
        entry = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "level": level,
            "service": self.service,
            "trace_id": _trace_id.get(),
            "span_id": _span_id.get(),
            "message": message,
            **kwargs
        }
        print(json.dumps(entry))

    def info(self, msg, **kw):  self._log("INFO", msg, **kw)
    def warn(self, msg, **kw):  self._log("WARN", msg, **kw)
    def error(self, msg, **kw): self._log("ERROR", msg, **kw)

@contextlib.contextmanager
def traced_operation(operation_name: str, logger: StructuredLogger):
    """Context manager that creates a new span for an operation."""
    trace_id = _trace_id.get() or new_trace_id()
    span_id = new_span_id()

    _trace_id.set(trace_id)
    _span_id.set(span_id)

    start = time.monotonic()
    logger.info(f"Started {operation_name}", operation=operation_name)

    try:
        yield
        duration_ms = (time.monotonic() - start) * 1000
        logger.info(f"Completed {operation_name}",
                   operation=operation_name,
                   duration_ms=round(duration_ms, 2),
                   status="success")
    except Exception as e:
        duration_ms = (time.monotonic() - start) * 1000
        logger.error(f"Failed {operation_name}",
                    operation=operation_name,
                    duration_ms=round(duration_ms, 2),
                    status="error",
                    error=str(e))
        raise

# Usage
log = StructuredLogger("order-service")

_trace_id.set(new_trace_id())
with traced_operation("create_order", log):
    log.info("Validating order payload", user_id="user_123", amount=99.99)
    with traced_operation("charge_payment", log):
        log.info("Calling payment gateway", gateway="stripe")
    with traced_operation("reserve_stock", log):
        log.info("Checking inventory", product_id="SKU-001", quantity=2)

---

6. Common Failure Scenarios & Solutions

Scenario	Problem	Solution Pattern
Service crashes	Requests lost	Health checks + Auto-restart (K8s)
Slow dependency	Thread pool exhausted	Timeouts + Circuit Breaker
Message flood	Consumer overwhelmed	Backpressure + Rate limiting
Split brain	Two leaders elected	Leader election (Raft/Paxos)
Stale cache	Serving old data	Cache invalidation + TTL
Data skew	One shard overloaded	Consistent hashing + Virtual nodes
Lost messages	Broker restart	Write-ahead log + Replication
Duplicate events	At-least-once delivery	Idempotency keys

---

7. Technology Decision Matrix

---

8. Summary: The Distributed Systems Cheat Sheet

TIP

Design Principles to live by:

• Design for failure — assume every service, network, and disk can fail
• Be idempotent — operations should be safe to retry
• Prefer async — decouple services with message queues
• Embrace eventual consistency — don't fight the CAP theorem
• Observe everything — logs + metrics + traces are non-negotiable
• Bound your latency — always set timeouts, never wait forever
• Make rollback easy — use sagas, feature flags, blue-green deployments

---

Generated with ❤️ — Distributed Systems Complete Guide