Distributed Systems

← Back to System Design 101 | ← Previous: Microservices

Introduction

Distributed systems are inherently complex. Networks are unreliable, clocks are not synchronized, and failures are the norm. Understanding these challenges is critical for building reliable distributed applications.

Consistency Models

Strong Consistency

All nodes see the same data at the same time.

# Example: Using distributed locks for strong consistency
import redis
from redis.lock import Lock

class DistributedCounter:
    """Strongly consistent distributed counter using Redis."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.key = "counter"
    
    def increment(self) -> int:
        """Increment counter with strong consistency."""
        lock_key = f"{self.key}:lock"
        lock = self.redis.lock(lock_key, timeout=10)
        
        with lock:
            current = int(self.redis.get(self.key) or 0)
            new_value = current + 1
            self.redis.set(self.key, new_value)
            return new_value

Eventual Consistency

Nodes may temporarily disagree, but eventually converge.

# Example: Eventual consistency with CRDTs (Conflict-Free Replicated Data Types)
from collections import defaultdict

class GCounter:
    """
    Grow-only counter CRDT.
    Different replicas can increment independently.
    """
    
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counts = defaultdict(int)  # {node_id: count}
    
    def increment(self):
        """Increment counter on this node."""
        self.counts[self.node_id] += 1
    
    def value(self) -> int:
        """Get total value across all nodes."""
        return sum(self.counts.values())
    
    def merge(self, other: 'GCounter'):
        """Merge with another replica."""
        for node_id, count in other.counts.items():
            self.counts[node_id] = max(self.counts[node_id], count)

# Usage across distributed nodes
node1 = GCounter("node-1")
node2 = GCounter("node-2")

node1.increment()
node1.increment()
node2.increment()

# Merge replicas
node1.merge(node2)
print(node1.value())  # 3

Distributed Consensus

Raft Consensus Algorithm

# Simplified Raft leader election concept
from enum import Enum
from random import uniform
import asyncio

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

class RaftNode:
    """
    Simplified Raft node for educational purposes.
    Production systems use etcd, Consul, or similar.
    """
    
    def __init__(self, node_id: str, peers: list):
        self.node_id = node_id
        self.peers = peers
        self.state = NodeState.FOLLOWER
        self.current_term = 0
        self.voted_for = None
        self.votes_received = 0
    
    async def start_election(self):
        """Transition to candidate and request votes."""
        self.state = NodeState.CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id
        self.votes_received = 1  # Vote for self
        
        print(f"{self.node_id}: Starting election for term {self.current_term}")
        
        # Request votes from peers
        for peer in self.peers:
            if await self.request_vote(peer):
                self.votes_received += 1
        
        # Check if won election
        if self.votes_received > len(self.peers) // 2:
            self.become_leader()
    
    async def request_vote(self, peer) -> bool:
        """Request vote from peer."""
        # Simplified - in real Raft, includes term and log info
        return True  # Peer grants vote
    
    def become_leader(self):
        """Transition to leader state."""
        self.state = NodeState.LEADER
        print(f"{self.node_id}: Became leader for term {self.current_term}")
        # Start sending heartbeats

Distributed Locking

import time
from contextlib import contextmanager

class RedisDistributedLock:
    """
    Distributed lock using Redis.
    Based on Redlock algorithm.
    """
    
    def __init__(self, redis_client, lock_name: str):
        self.redis = redis_client
        self.lock_name = f"lock:{lock_name}"
        self.lock_timeout = 10  # seconds
    
    def acquire(self, timeout: int = 10) -> bool:
        """Acquire distributed lock."""
        identifier = str(time.time())
        end = time.time() + timeout
        
        while time.time() < end:
            # Try to acquire lock
            if self.redis.set(
                self.lock_name,
                identifier,
                ex=self.lock_timeout,
                nx=True  # Only set if not exists
            ):
                return identifier
            
            time.sleep(0.001)  # Brief pause before retry
        
        return None
    
    def release(self, identifier: str):
        """Release distributed lock."""
        # Use Lua script for atomic check-and-delete
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(script, 1, self.lock_name, identifier)
    
    @contextmanager
    def lock(self):
        """Context manager for distributed lock."""
        identifier = self.acquire()
        if not identifier:
            raise Exception("Failed to acquire lock")
        
        try:
            yield
        finally:
            self.release(identifier)

# Usage
lock = RedisDistributedLock(redis_client, "order:123")
with lock.lock():
    # Critical section - only one process can execute this
    process_order("123")

Distributed Caching

import hashlib

class ConsistentHash:
    """
    Consistent hashing for distributed cache.
    Minimizes rebalancing when nodes are added/removed.
    """
    
    def __init__(self, nodes: list, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring = {}
        self.sorted_keys = []
        
        for node in nodes:
            self.add_node(node)
    
    def _hash(self, key: str) -> int:
        """Hash function."""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def add_node(self, node: str):
        """Add node to the hash ring."""
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}:{i}"
            hash_key = self._hash(virtual_key)
            self.ring[hash_key] = node
        
        self.sorted_keys = sorted(self.ring.keys())
    
    def remove_node(self, node: str):
        """Remove node from the hash ring."""
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}:{i}"
            hash_key = self._hash(virtual_key)
            del self.ring[hash_key]
        
        self.sorted_keys = sorted(self.ring.keys())
    
    def get_node(self, key: str) -> str:
        """Get node responsible for key."""
        if not self.ring:
            return None
        
        hash_key = self._hash(key)
        
        # Find first node clockwise from hash
        for ring_key in self.sorted_keys:
            if ring_key >= hash_key:
                return self.ring[ring_key]
        
        # Wrap around to first node
        return self.ring[self.sorted_keys[0]]

# Usage
cache_nodes = ["cache-1", "cache-2", "cache-3"]
hash_ring = ConsistentHash(cache_nodes)

# Distribute keys across nodes
key = "user:12345"
node = hash_ring.get_node(key)
print(f"Key {key} maps to {node}")

Quorum Reads/Writes

class QuorumStore:
    """
    Distributed storage with quorum reads and writes.
    Ensures consistency across replicas.
    """
    
    def __init__(self, replicas: list, quorum_size: int = 2):
        self.replicas = replicas  # List of storage nodes
        self.quorum_size = quorum_size
    
    async def write(self, key: str, value: str, version: int):
        """Write to quorum of replicas."""
        successful_writes = 0
        
        for replica in self.replicas:
            try:
                await replica.write(key, value, version)
                successful_writes += 1
                
                if successful_writes >= self.quorum_size:
                    return True
            except Exception as e:
                print(f"Write to {replica} failed: {e}")
        
        return False
    
    async def read(self, key: str):
        """Read from quorum of replicas."""
        responses = []
        
        for replica in self.replicas:
            try:
                value, version = await replica.read(key)
                responses.append((value, version))
                
                if len(responses) >= self.quorum_size:
                    break
            except Exception as e:
                print(f"Read from {replica} failed: {e}")
        
        if len(responses) < self.quorum_size:
            raise Exception("Failed to achieve read quorum")
        
        # Return value with highest version
        return max(responses, key=lambda x: x[1])

Fault Tolerance

Retry with Exponential Backoff

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type(Exception)
)
async def call_external_service(data):
    """Call external service with automatic retries."""
    response = await httpx.post("https://api.example.com/data", json=data)
    response.raise_for_status()
    return response.json()

Bulkhead Pattern

from asyncio import Semaphore

class BulkheadExecutor:
    """
    Bulkhead pattern to isolate failures.
    Limits concurrent requests to prevent resource exhaustion.
    """
    
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = Semaphore(max_concurrent)
    
    async def execute(self, func, *args, **kwargs):
        """Execute with concurrency limit."""
        async with self.semaphore:
            return await func(*args, **kwargs)

# Usage
bulkhead = BulkheadExecutor(max_concurrent=5)

async def protected_call():
    return await bulkhead.execute(external_api_call)

Lessons Learned

What worked:

1. Accept eventual consistency where possible

2. Use proven consensus algorithms (etcd, Consul)

3. Implement proper retry logic with backoff

4. Monitor and alert on distributed system metrics

5. Design for partial failures

What didn't work:

1. Assuming networks are reliable

2. Distributed transactions everywhere

3. Not handling split-brain scenarios

4. Ignoring clock skew issues

5. No timeout on distributed calls

What's Next

Understanding distributed systems, let's explore observability and monitoring:

• Observability & Monitoring →: Metrics, logging, and tracing

---

Navigation:

• ← Previous: Microservices

• Next: Observability →

• Back to Overview