Distributed Systems

Message queues, distributed caching, service mesh, and resilience patterns

Welcome to Distributed Systems

A distributed system is one where components run on different machines and coordinate through network communication. These systems power everything from Netflix streaming to Amazon shopping to Google search. While they enable massive scale and resilience, they introduce fundamental challenges: networks are unreliable, machines fail, clocks drift, and operations can't be instantaneous. This lesson covers the essential tools and patterns for building reliable distributed systems: message queues for async communication, distributed caching for performance, service meshes for service-to-service networking, and resilience patterns to handle inevitable failures.

Quick Navigation

The CAP Theorem: Pick Two

The CAP theorem states that in a distributed system, you can only guarantee two of three properties: Consistency (all nodes see the same data), Availability (every request gets a response), and Partition tolerance (system works despite network failures). Since network partitions are inevitable, you're really choosing between CP (consistent but unavailable during partitions) and AP (available but potentially inconsistent).

CP Systems

Choose: Consistency + Partition Tolerance

Sacrifice availability during network issues. Better to be unavailable than serve wrong data.

Examples: Banks, inventory systems, distributed databases (MongoDB, HBase)

AP Systems

Choose: Availability + Partition Tolerance

Always respond, even if data might be stale. Better to serve something than nothing.

Examples: Social media, DNS, distributed caches (Cassandra, DynamoDB)

CA Systems

Choose: Consistency + Availability

Only possible without network partitions, essentially single-node systems.

Examples: Traditional RDBMS on single server (PostgreSQL, MySQL)

Reality: CAP is oversimplified. Modern systems use nuanced approaches like eventual consistency, tunable consistency levels, and read/write quorums to balance these trade-offs dynamically.

Message Queues (Asynchronous Communication)

Message queues enable asynchronous, decoupled communication between services. Producers send messages to a queue, consumers process them independently. This enables scalability (add more consumers), reliability (messages persist), and resilience (consumers can fail and retry).

Core Concepts

Producer: Sends messages to queue

The Producer focuses on high availability, returning a response to the user as quickly as possible by offloading heavy tasks.

class OrderProducer:
    def __init__(self, queue_client):
        self.queue = queue_client

    def create_order(self, user_id, items):
        # Create order in database
        order = self.queue.create_order(user_id, items)

        # Send message to queue for async processing
        message = {
            "event": "order.created",
            "order_id": order.id,
            "user_id": user_id,
            "items": items,
            "total": order.total
        }

        self.queue.send("orders", message)

        # Return immediately - don't wait for processing
        return order

Consumer: Processes messages from queue

The Consumer runs independently, pulling messages and performing the heavy lifting like sending emails or updating inventory.

class OrderConsumer:
    def __init__(self, queue_client):
        self.queue = queue_client

    def start(self):
        """Start consuming messages"""
        while True:
            # Pull message from queue
            message = self.queue.receive("orders", timeout=30)

            if message:
                try:
                    self.process_order(message)
                    # Acknowledge successful processing
                    self.queue.ack(message)
                except Exception as e:
                    logging.error(f"Failed to process: {e}")
                    # Reject - message goes back to queue or dead letter queue
                    self.queue.nack(message)

    def process_order(self, message):
        """Process order asynchronously"""
        order_id = message["order_id"]

        # Send confirmation email
        self.send_email(message["user_id"], order_id)

        # Update inventory
        self.update_inventory(message["items"])

        # Trigger shipping
        self.create_shipment(order_id)

Benefits of Queue-Based Architecture

Decoupling

Producer doesn't know about consumers; services evolve independently.

Scalability

Add more consumers to process spikes in traffic faster.

Reliability

Messages are persisted; if a consumer fails, the message is retried.

Load Smoothing

The queue acts as a buffer, preventing traffic bursts from overwhelming the DB.

Message Queue Patterns

Choosing the right pattern depends on whether you need a simple task list or a complex event-driven notification system.

1. Work Queues (Task Queues)

Distributes time-consuming tasks among multiple workers. Each message is processed by exactly one consumer.

# 1. WORK QUEUE: Single consumer processes each message
class WorkQueue:
    """One message processed by one consumer"""
    def send_job(self, job_data):
        # Add to queue
        queue.send("jobs", {
            "job_id": uuid.uuid4(),
            "data": job_data,
            "retry_count": 0
        })

    def process_job(self):
        message = queue.receive("jobs")
        # Only one worker processes this message
        do_work(message["data"])
        queue.ack(message)

2. Pub/Sub (Publish/Subscribe)

Broadcasts a message to multiple subscribers. Ideal for notifying different services (e.g., Email, Analytics) about the same event.

# 2. PUB/SUB: Publish once, multiple services receive

# Subscribed Services:
# 1. EmailService -> Sends welcome email
# 2. AnalyticsService -> Tracks signup event

class PubSub:
    """Publish to topic, multiple subscribers receive"""
    def publish_event(self, event_type, data):
        # Publish to topic
        pubsub.publish("user.events", {
            "type": event_type,
            "data": data
        })

# Multiple services subscribe to same topic
class EmailService:
    def subscribe(self):
        pubsub.subscribe("user.events", self.on_user_event)

    def on_user_event(self, message):
        if message["type"] == "user.created":
            self.send_welcome_email(message["data"])

class AnalyticsService:
    def subscribe(self):
        pubsub.subscribe("user.events", self.on_user_event)

    def on_user_event(self, message):
        self.track_event(message["type"], message["data"])

3. Priority Queues

# 3. PRIORITY QUEUE: Process high-priority messages first
class PriorityQueue:
    def send_with_priority(self, message, priority):
        queue.send("tasks", message, priority=priority)

    # High priority processed first
    send_with_priority(
        {"task": "critical_alert"},
        priority=10
    )

    send_with_priority(
        {"task": "batch_report"},
        priority=1
    )

4. Delayed Queues

# 4. DELAYED/SCHEDULED MESSAGES
class ScheduledQueue:
    def schedule_reminder(self, user_id, minutes):
        queue.send("reminders", {
            "user_id": user_id,
            "message": "Your cart is waiting!"
        }, delay=minutes * 60)  # Deliver after delay

5. Dead Letter Queue (DLQ)

A safety net for "poison messages" that fail repeatedly. Instead of blocking the queue, they are moved to a DLQ for manual inspection.

# 5. DEAD LETTER QUEUE: Handle poison messages
class ResilientConsumer:
    def process(self, message):
        try:
            do_work(message)
        except Exception:
            message["retry_count"] += 1

            if message["retry_count"] > 3:
                # Move to dead letter queue for manual inspection
                queue.send("dead_letters", message)
            else:
                # Retry with exponential backoff
                delay = 2 ** message["retry_count"]
                queue.send("jobs", message, delay=delay)

Popular Message Queue Systems

# RabbitMQ - Full-featured message broker
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='orders')

# Publish
channel.basic_publish(
    exchange='',
    routing_key='orders',
    body='{"order_id": 123}'
)

# Consume
def callback(ch, method, properties, body):
    print(f"Received {body}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='orders', on_message_callback=callback)
channel.start_consuming()

# Apache Kafka - High-throughput distributed streaming
from kafka import KafkaProducer, KafkaConsumer

# Producer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer.send('orders', b'{"order_id": 123}')

# Consumer with consumer groups
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processors',  # Load balancing across group
    auto_offset_reset='earliest'
)

for message in consumer:
    process_order(message.value)

# AWS SQS - Managed queue service
import boto3

sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123/orders'

# Send message
sqs.send_message(
    QueueUrl=queue_url,
    MessageBody='{"order_id": 123}',
    DelaySeconds=10
)

# Receive messages
messages = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=20  # Long polling
)

for message in messages.get('Messages', []):
    process(message['Body'])
    # Delete after processing
    sqs.delete_message(
        QueueUrl=queue_url,
        ReceiptHandle=message['ReceiptHandle']
    )

Which one should you choose?

RabbitMQ: When you need complex routing, traditional messaging.
Kafka: When you have massive scale, event streaming, high throughput or need to "replay" events from the past.
SQS: When you want zero-maintenance and pay-as-you-go pricing.
Redis: When you need ultra-fast, simple, in-memory queues, good for rate limiting.

Distributed Caching: Speed at Scale

Databases are the bottleneck of most systems, to address this, stores frequently accessed data in RAM to provide sub-millisecond response times is a good alternative. Distributed caches store data in memory across multiple servers for fast access. They reduce database load, improve response times, and can serve as a shared state layer between services. Redis and Memcached are the most popular options.

Cache Patterns

1. CACHE-ASIDE

Lazy Loading.

class UserService:
    def __init__(self, cache, database):
        self.cache = cache
        self.db = database

    def get_user(self, user_id):
        # Try cache first
        cached = self.cache.get(f"user:{user_id}")
        if cached:
            return cached  # Cache hit

        # Cache miss - load from database
        user = self.db.query("SELECT * FROM users WHERE id = %s", user_id)

        # Store in cache for next time
        self.cache.set(f"user:{user_id}", user, ttl=3600)  # 1 hour

        return user

    def update_user(self, user_id, data):
        # Update database
        self.db.update("users", user_id, data)

        # Invalidate cache
        self.cache.delete(f"user:{user_id}")

Caching with TTL and refresh.

class CacheWithRefresh:
    def get_popular_products(self):
        products = self.cache.get("popular_products")

        if products:
            return products

        # Expensive query
        products = self.db.query("""
            SELECT * FROM products
            WHERE sales > 1000
            ORDER BY sales DESC
            LIMIT 20
        """)

        # Cache for 5 minutes
        self.cache.set("popular_products", products, ttl=300)

        return products

2. WRITE-THROUGH

Write to cache and DB together.

class WriteThrough:
    def update_user(self, user_id, data):
        # Write to database
        self.db.update("users", user_id, data)

        # Update cache immediately
        user = self.db.get_user(user_id)
        self.cache.set(f"user:{user_id}", user)

3. WRITE-BEHIND

Write to cache, async sync to database.

class WriteBehind:
    def update_user(self, user_id, data):
        # Write to cache immediately
        self.cache.set(f"user:{user_id}", data)

        # Queue database write for later
        self.queue.send("db_writes", {
            "table": "users",
            "id": user_id,
            "data": data
        })

Redis: The Speed Demon of Distributed Systems

Redis (Remote Dictionary Server) is an open-source, in-memory data structure store. While classified as a NoSQL Database, it is primarily used as a Distributed Cache or Message Broker. By operating in RAM, it handles millions of operations per second with sub-millisecond latency, helping to solve the "Database Bottleneck" by storing frequently accessed data and preventing your primary database from crashing under high load.

Extreme Speed

Sub-millisecond latency by avoiding slow Disk I/O operations.

Data Structures

Supports Lists, Sets, Hashes, and Sorted Sets, not just simple strings.

Atomic Ops

Prevents race conditions through single-threaded execution of commands.

Common Use Cases

1. Distributed Caching

The "Gold Standard" for caching. Sits between the app and DB to store sessions, product details, or configs.

2. Session Management

Acts as a centralized "Session Store" that all distributed web servers can access simultaneously.

3. Real-Time Leaderboards

Uses Sorted Sets to rank millions of users by score instantly, perfect for gaming and fintech.

4. Rate Limiting

Acts as a high-speed counter to track API requests per IP and block abuse in real-time.

Architect's Tip: While Redis is fast, RAM is expensive. Always set an TTL (Time To Live) for your keys to prevent the memory from filling up with "stale" data that is no longer needed.

The Cache-Aside Pattern

This is the most common caching strategy. The application checks the cache first, and only queries the database on a "miss."

import redis

# 1. Connect to Redis (In-Memory)
cache = redis.Redis(host='localhost', port=6379)

def get_user_profile(user_id):
    # 2. Try to fetch from Redis
    profile = cache.get(f"user:{user_id}")

    if profile:
        return profile # Cache Hit (Fast!)

    # 3. Cache Miss: Query Database (Slow)
    profile = db.query("SELECT * FROM users WHERE id = %s", user_id)

    # 4. Save to Redis for next time (expires in 1 hour)
    cache.setex(f"user:{user_id}", 3600, profile)

    return profile

Feature	Traditional DB (SQL)	Redis
Primary Storage	Disk (SSD/HDD)	RAM (Memory)
Latency	10ms - 100ms+	< 1ms
Best For	Complex Queries / Durability	Speed / Temporary Data

Architect's Tip: Use Redis for...

Session Management: Shared user state.
Rate Limiting: Blocking API abuse.

Leaderboards: Real-time sorted lists.
Pub/Sub: Lightweight messaging.

Other examples...

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 1. STRINGS - Basic key-value
r.set('user:123:name', 'Alice')
r.get('user:123:name')  # 'Alice'

# With expiration
r.setex('session:abc', 3600, 'session_data')  # Expires in 1 hour

# Increment (atomic)
r.incr('page:views')  # Atomic counter

# 2. HASHES - Store objects
r.hset('user:123', mapping={
    'name': 'Alice',
    'email': 'alice@example.com',
    'age': 30
})
r.hgetall('user:123')  # Get entire object

# 3. LISTS - Queues and stacks
r.lpush('queue:tasks', 'task1', 'task2')  # Push to queue
r.rpop('queue:tasks')  # Pop from queue (FIFO)

# Recent activity feed
r.lpush('user:123:activity', 'logged_in')
r.ltrim('user:123:activity', 0, 99)  # Keep only last 100

# 4. SETS - Unique collections
r.sadd('users:online', 'user1', 'user2', 'user3')
r.sismember('users:online', 'user1')  # Check membership
r.scard('users:online')  # Count members

# Set operations
r.sadd('users:premium', 'user1', 'user4')
r.sinter('users:online', 'users:premium')  # Intersection

# 5. SORTED SETS - Leaderboards, rankings
r.zadd('leaderboard', {'player1': 100, 'player2': 200, 'player3': 150})
r.zrevrange('leaderboard', 0, 9, withscores=True)  # Top 10

# Get user rank
r.zrevrank('leaderboard', 'player2')  # Returns rank

# 6. PUB/SUB - Real-time messaging
def message_handler(message):
    print(f"Received: {message['data']}")

pubsub = r.pubsub()
pubsub.subscribe(**{'notifications': message_handler})
pubsub.run_in_thread(sleep_time=0.001)

# Publish
r.publish('notifications', 'New message!')

# 7. DISTRIBUTED LOCKS
def acquire_lock(lock_name, timeout=10):
    """Distributed lock for coordinating across services"""
    lock_key = f"lock:{lock_name}"
    identifier = str(uuid.uuid4())

    # Try to acquire lock
    if r.set(lock_key, identifier, nx=True, ex=timeout):
        return identifier
    return None

def release_lock(lock_name, identifier):
    """Release lock only if we own it"""
    lock_key = f"lock:{lock_name}"

    # Lua script for atomic check-and-delete
    lua_script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """

    return r.eval(lua_script, 1, lock_key, identifier)

# Usage
lock_id = acquire_lock("process_orders")
if lock_id:
    try:
        # Critical section - only one service executes this
        process_orders()
    finally:
        release_lock("process_orders", lock_id)

# 8. RATE LIMITING
def is_rate_limited(user_id, max_requests=100, window=60):
    """Sliding window rate limiter"""
    key = f"rate_limit:{user_id}"
    now = time.time()

    # Remove old entries
    r.zremrangebyscore(key, 0, now - window)

    # Count requests in window
    count = r.zcard(key)

    if count < max_requests:
        # Add new request
        r.zadd(key, {str(uuid.uuid4()): now})
        r.expire(key, window)
        return False

    return True  # Rate limited

# 9. CACHING WITH STAMPEDE PROTECTION
def get_cached_data(key, fetch_fn, ttl=3600):
    """Cache with protection against thundering herd"""
    value = r.get(key)
    if value:
        return value

    # Try to acquire lock to fetch
    lock_key = f"lock:{key}"
    if r.set(lock_key, 1, nx=True, ex=10):
        try:
            # We got the lock - fetch the data
            value = fetch_fn()
            r.setex(key, ttl, value)
            return value
        finally:
            r.delete(lock_key)
    else:
        # Someone else is fetching - wait a bit and retry
        time.sleep(0.1)
        return get_cached_data(key, fetch_fn, ttl)

The "Hardest Part": Cache Invalidation

Caching is easy; keeping the cache consistent with the database is the challenge. If you don't invalidate correctly, users see old, stale data.

The "Cache Miss" Storm: When you invalidate a highly popular key (like "homepage_data"), thousands of requests will hit your database at the same time. This is called a Cache Stampede. Use "Mutex Locking" to ensure only one request regenerates the cache.

# 1. TIME-BASED (TTL)
cache.set("data", value, ttl=300)  # Expires after 5 minutes

# 2. EVENT-BASED
class EventBasedCache:
    def on_user_updated(self, event):
        user_id = event["user_id"]
        # Invalidate cache when user changes
        cache.delete(f"user:{user_id}")
        cache.delete(f"user:{user_id}:profile")

# 3. VERSIONED KEYS
def get_user_v2(user_id):
    version = cache.get("user_schema_version") or "v1"
    key = f"user:{user_id}:{version}"
    return cache.get(key)

# 4. CACHE TAGGING
class TaggedCache:
    def set_with_tags(self, key, value, tags):
        cache.set(key, value)
        for tag in tags:
            cache.sadd(f"tag:{tag}", key)

    def invalidate_tag(self, tag):
        keys = cache.smembers(f"tag:{tag}")
        for key in keys:
            cache.delete(key)
        cache.delete(f"tag:{tag}")

# Usage
cache.set_with_tags("product:123", product_data, ["products", "user:456"])
cache.invalidate_tag("user:456")  # Clears all user's cached data

Cache Gotchas

Cache Stampede: Occurs when many requests try to fetch the same missing key simultaneously, potentially crashing your database.
Stale Data: The risk that the Cache and DB diverge, leading to users seeing outdated information.
Memory Limits: RAM is finite. You must configure Eviction Policies (LRU, LFU) to decide which data to drop when the cache is full.
Network Overhead: Connecting to a cache takes time. The cache must be significantly faster than the DB to be worthwhile.

Load Balancing: Distributing Traffic

Load balancers distribute incoming requests across multiple server instances. Critical for scaling horizontally, improving availability (if one server fails, others handle traffic), and preventing any single server from being overwhelmed.

Load Balancing Algorithms

Round Robin

Distribute requests evenly in rotation. Simple, works well if all servers equal.

Server 1 → Server 2 → Server 3 → Server 1...

Least Connections

Send to server with fewest active connections. Better for long-lived connections.

Active connections: S1(5), S2(3), S3(7) → Send to S2

Weighted Round Robin

Servers with higher capacity get more requests. Use when servers have different specs.

S1(2x) → S1 → S2 → S1 → S3...

IP Hash / Consistent Hashing

Hash client IP to server. Same client always goes to same server (session affinity).

hash(client_ip) % num_servers

💡 Pro Tip: Use Least Connections for variable request durations (e.g., uploads). Use Consistent Hashing when you need session affinity or cache locality. Use Round Robinfor stateless, similar-duration requests.

Service Discovery: Finding Services Dynamically

In dynamic environments (containers, cloud), services start and stop frequently with changing IP addresses. Service discovery lets services find each other automatically without hardcoded IPs. Services register themselves, and clients query the registry to find available instances.

Service Discovery Patterns

Client-Side Discovery

How: Client queries service registry, selects instance, makes direct request

✅ Pros: No extra hop, client controls load balancing
❌ Cons: Couples client to discovery logic

Examples: Netflix Eureka, Consul

Server-Side Discovery

How: Client requests load balancer, which queries registry and routes request

✅ Pros: Clients stay simple, centralized logic
❌ Cons: Extra network hop, load balancer as bottleneck

Examples: AWS ELB, Kubernetes Services

Kubernetes Service Discovery: Kubernetes provides built-in DNS-based service discovery. Services are accessible at `service-name.namespace.svc.cluster.local`. Kubernetes automatically updates endpoints as pods start/stop.

Replication Strategies: Data Redundancy

Replication copies data across multiple nodes for availability and fault tolerance. If one node fails, others serve the data. Also enables read scaling by distributing read traffic across replicas.

Replication Approaches

Leader-Follower (Primary-Replica)

Writes: Only to leader
Reads: From any replica

Leader replicates changes to followers. Simple, widely used. Followers may lag (replication lag).

Multi-Leader (Multi-Master)

Writes: To any leader
Reads: From any leader

Multiple leaders accept writes, sync with each other. Better availability, but conflicts possible.

Leaderless (Dynamo-style)

Writes: To multiple nodes
Reads: From multiple nodes

No designated leader. Client writes to N nodes, reads from N nodes. Used in Cassandra, DynamoDB.

Synchronous vs Asynchronous

Sync: Wait for replicas before ack
Async: Ack immediately

Sync = stronger consistency, slower. Async = faster, but replicas may lag.

Quorum & Consistency: Tunable Guarantees

Quorum-based systems let you tune the trade-off between consistency and availability. Instead of requiring ALL nodes to agree, require a quorum (majority). Reads and writes contact multiple nodes.

Quorum Formula

# Quorum Configuration
N = Total number of replicas (nodes storing data)
W = Write quorum (number of nodes that must acknowledge write)
R = Read quorum (number of nodes to read from)

# Consistency guarantee:
# If W + R > N, then reads see latest write (strong consistency)

# Example: N=3, W=2, R=2
# - Write must succeed on 2 out of 3 nodes
# - Read queries 2 out of 3 nodes
# - W + R = 4 > N = 3 → Guaranteed to read latest write

# Common configurations:
# Strong consistency:   W=2, R=2, N=3  (majority)
# Read-optimized:       W=3, R=1, N=3  (wait for all writes, fast reads)
# Write-optimized:      W=1, R=3, N=3  (fast writes, read all for consistency)
# Eventual consistency: W=1, R=1, N=3  (fastest, weakest consistency)

Trade-offs: Higher W = slower writes, more durable. Higher R = slower reads, more consistent. Choose based on your application: bank account (high W+R), social media feed (low W+R).

Service Mesh: Infrastructure Layer for Microservices

A service mesh is a dedicated infrastructure layer that handles service-to-service communication. Instead of each service implementing its own retry logic, circuit breakers, and observability, the mesh provides this as a platform capability through sidecar proxies.

Architecture

# Without Service Mesh: Each service handles its own concerns
class OrderService:
    def call_payment_service(self, data):
        # Service must implement all this logic
        retries = 0
        while retries < 3:
            try:
                response = requests.post(
                    "http://payment-service/charge",
                    json=data,
                    timeout=5
                )

                # Manual metrics
                metrics.increment("payment_calls")

                # Manual logging
                logger.info(f"Called payment service: {response.status_code}")

                if response.status_code == 200:
                    return response.json()

                retries += 1
                time.sleep(2 ** retries)  # Exponential backoff

            except requests.exceptions.Timeout:
                retries += 1
                metrics.increment("payment_timeouts")

# With Service Mesh: Sidecar proxy handles cross-cutting concerns
class OrderService:
    def call_payment_service(self, data):
        # Simple call - mesh handles retries, timeouts, metrics, tracing
        response = requests.post(
            "http://payment-service/charge",
            json=data
        )
        return response.json()

# Service Mesh provides:
# - Automatic retries with exponential backoff
# - Circuit breakers
# - Load balancing
# - Mutual TLS (mTLS) for security
# - Distributed tracing
# - Metrics collection
# - Traffic management (canary deployments, A/B testing)

Istio: Popular Service Mesh

# Istio configuration (YAML)
# 1. RETRY POLICY
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: payment-service
spec:
  hosts:
    - payment-service
  http:
    - route:
        - destination:
            host: payment-service
      retries:
        attempts: 3
        perTryTimeout: 2s
        retryOn: 5xx,connect-failure,refused-stream

# 2. CIRCUIT BREAKER
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: payment-service
spec:
  host: payment-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 2
    outlierDetection:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50

# 3. LOAD BALANCING
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: order-service
spec:
  host: order-service
  trafficPolicy:
    loadBalancer:
      consistentHash:
        httpCookie:
          name: session
          ttl: 3600s

# 4. TRAFFIC SPLITTING (Canary Deployment)
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    - match:
        - headers:
            user-type:
              exact: beta
      route:
        - destination:
            host: order-service
            subset: v2
    - route:
        - destination:
            host: order-service
            subset: v1
          weight: 90
        - destination:
            host: order-service
            subset: v2
          weight: 10  # 10% of traffic to v2

# 5. MUTUAL TLS
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
spec:
  mtls:
    mode: STRICT  # All service-to-service traffic encrypted

# 6. RATE LIMITING
apiVersion: networking.istio.io/v1alpha3
kind: EnvoyFilter
metadata:
  name: filter-ratelimit
spec:
  configPatches:
    - applyTo: HTTP_FILTER
      match:
        context: SIDECAR_INBOUND
      patch:
        operation: INSERT_BEFORE
        value:
          name: envoy.filters.http.local_ratelimit
          typed_config:
            "@type": type.googleapis.com/udpa.type.v1.TypedStruct
            type_url: type.googleapis.com/envoy.extensions.filters.http.local_ratelimit.v3.LocalRateLimit
            value:
              stat_prefix: http_local_rate_limiter
              token_bucket:
                max_tokens: 100
                tokens_per_fill: 100
                fill_interval: 60s

Service Mesh Benefits

✅ Advantages

Uniform reliability patterns
Language-agnostic
Centralized security (mTLS)
Observability out-of-box
Traffic management without code changes

❌ Disadvantages

Operational complexity
Latency overhead (proxy hop)
Resource consumption (sidecars)
Debugging complexity
Learning curve

When to Use Service Mesh: Multiple microservices (10+), need consistent resilience patterns, security requirements, complex traffic routing. When to Skip:Few services, simple architecture, limited ops team.

Resilience Patterns: Handling Failures

In distributed systems, failures are inevitable. Networks fail, services crash, and latency spikes. Resilience patterns help systems gracefully handle these failures without cascading to other services.

1. Circuit Breaker

🎯 Purpose: Prevent cascading failures by failing fast when a downstream service is unhealthy.

The Problem: When a service fails, clients keep sending requests, wasting resources and creating timeouts. This overwhelms the failing service and blocks threads in the caller, potentially causing cascading failures.

The Solution: Circuit breaker monitors failures. After a threshold, it "opens" and immediately rejects requests without calling the service. After a timeout, it enters "half-open" to test if service recovered. Like an electrical circuit breaker protecting your house from electrical fires.

💡 When to Use: Calling external services or microservices that may fail. Especially important when failures are expensive (long timeouts) or when you have fallback options available.

Implementation

class CircuitBreaker:
    """Prevent cascading failures by failing fast"""

    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise CircuitBreakerOpen("Circuit breaker is open")

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

    def _on_success(self):
        self.failure_count = 0
        if self.state == "HALF_OPEN":
            self.state = "CLOSED"

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

# Usage
payment_breaker = CircuitBreaker(failure_threshold=5, timeout=30)

def process_payment(amount):
    return payment_breaker.call(payment_service.charge, amount)

try:
    result = process_payment(100)
except CircuitBreakerOpen:
    # Use fallback
    result = use_backup_payment_method()

# States:
# CLOSED: Normal operation, requests go through
# OPEN: Too many failures, reject immediately (fail fast)
# HALF_OPEN: Testing if service recovered, allow one request

2. Retry with Exponential Backoff

🎯 Purpose: Automatically retry failed operations with increasing delays to handle transient failures.

The Problem: Networks experience temporary blips, services restart, databases have momentary connection issues. These transient failures resolve quickly, but immediate retry can overwhelm a recovering service (thundering herd).

The Solution: Retry with exponentially increasing delays (1s, 2s, 4s, 8s...). This gives the service time to recover while avoiding synchronized retry storms. Add jitter (randomness) to prevent all clients retrying simultaneously.

💡 When to Use: Transient network errors, connection timeouts, 503 Service Unavailable, rate limit errors (429).⚠️ Don't Use: For non-idempotent operations without idempotency keys, or validation errors (4xx except 429).

Implementation

def retry_with_backoff(func, max_retries=3, base_delay=1, max_delay=60):
    """Retry failed operations with increasing delays"""

    for attempt in range(max_retries):
        try:
            return func()
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries - 1:
                raise  # Last attempt, give up

            # Exponential backoff with jitter
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            time.sleep(delay + jitter)

            logging.warning(f"Retry {attempt + 1}/{max_retries} after {delay}s")

# Usage
def fetch_user_data(user_id):
    response = requests.get(f"http://user-service/users/{user_id}", timeout=5)
    return response.json()

user = retry_with_backoff(lambda: fetch_user_data(123))

# Advanced: Decorator pattern
def retry(max_attempts=3, backoff=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return retry_with_backoff(
                lambda: func(*args, **kwargs),
                max_retries=max_attempts,
                base_delay=backoff
            )
        return wrapper
    return decorator

@retry(max_attempts=3, backoff=2)
def call_external_api(data):
    return requests.post("https://api.example.com/endpoint", json=data)

3. Bulkhead Pattern

🎯 Purpose: Isolate resources (threads, connections) so failure in one area doesn't exhaust resources for everything.

The Problem: One slow/failing service consumes all available threads/connections. Now even healthy services can't process requests because the thread pool is exhausted. A single failure takes down your entire application.

The Solution: Partition resources into isolated pools (bulkheads). Critical services get their own thread pool, preventing low-priority services from starving high-priority ones. Named after ship bulkheads that contain flooding to one compartment.

💡 When to Use: Multiple dependencies with different SLAs, mix of critical and non-critical services, risk of resource exhaustion from slow services. Essential for preventing cascading failures in microservices.

Implementation

# Isolate resources to prevent total system failure

from concurrent.futures import ThreadPoolExecutor
import threading

class Bulkhead:
    """Isolate different workloads with separate thread pools"""

    def __init__(self):
        # Separate pools for different services
        self.critical_pool = ThreadPoolExecutor(max_workers=10)
        self.normal_pool = ThreadPoolExecutor(max_workers=5)
        self.background_pool = ThreadPoolExecutor(max_workers=2)

        self.semaphores = {
            "critical": threading.Semaphore(10),
            "normal": threading.Semaphore(5),
            "background": threading.Semaphore(2)
        }

    def execute(self, priority, func, *args):
        if priority == "critical":
            pool = self.critical_pool
            semaphore = self.semaphores["critical"]
        elif priority == "normal":
            pool = self.normal_pool
            semaphore = self.semaphores["normal"]
        else:
            pool = self.background_pool
            semaphore = self.semaphores["background"]

        # Acquire semaphore
        if not semaphore.acquire(blocking=False):
            raise ResourceExhausted(f"{priority} pool exhausted")

        future = pool.submit(func, *args)
        # Release semaphore when task completes, not immediately
        future.add_done_callback(lambda f: semaphore.release())
        return future

# Usage
bulkhead = Bulkhead()

# Critical operations get more resources
bulkhead.execute("critical", process_payment, order_id)

# Background tasks use separate pool
bulkhead.execute("background", generate_report, user_id)

# If background tasks exhaust their pool, critical operations unaffected

4. Timeout Pattern

🎯 Purpose: Prevent requests from hanging indefinitely by setting maximum wait times.

The Problem: Services can hang forever waiting for responses. A slow database query, unresponsive API, or network partition can block threads indefinitely. Without timeouts, you'll run out of threads/connections as they all wait forever.

The Solution: Set aggressive timeouts on all I/O operations (database queries, HTTP calls, cache reads). Fail fast with timeout errors rather than blocking forever. Typically: 2-5 seconds for APIs, 50-100ms for databases, 1-2 seconds for cache.

💡 When to Use: ALWAYS. Every network call, database query, external API, cache operation needs a timeout. Default is often infinite or too long (30+ seconds). Make timeouts explicit and aggressive.

Implementation

# Always set timeouts to prevent hanging
class TimeoutManager:
    """Manage different timeout levels"""

    TIMEOUTS = {
        "database": 5,      # Database queries should be fast
        "cache": 1,         # Cache should be very fast
        "external_api": 30, # External APIs can be slower
        "internal_api": 10  # Internal services moderate timeout
    }

    @staticmethod
    def call_with_timeout(service_type, func, *args, **kwargs):
        timeout = TimeoutManager.TIMEOUTS.get(service_type, 10)

        try:
            # Using signal for timeout (Unix only)
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(timeout)

            result = func(*args, **kwargs)
            signal.alarm(0)  # Cancel alarm
            return result

        except TimeoutError:
            logging.error(f"{service_type} timeout after {timeout}s")
            raise

# Better: Use async with asyncio timeout
import asyncio

async def fetch_with_timeout(url, timeout=10):
    try:
        timeout_config = aiohttp.ClientTimeout(total=timeout)
        async with aiohttp.ClientSession(timeout=timeout_config) as session:
            async with session.get(url) as response:
                return await response.json()
    except asyncio.TimeoutError:
        logging.error(f"Request to {url} timed out")
        raise

# Usage
result = await fetch_with_timeout("http://api.example.com/data", timeout=5)

5. Rate Limiting

🎯 Purpose: Control the rate of incoming requests to prevent resource exhaustion and ensure fair usage.

The Problem: Malicious users can overwhelm your service with requests (DDoS), one customer can monopolize resources starving others, bugs can create infinite request loops, or sudden traffic spikes can take down your service.

The Solution: Limit requests per user/IP/API key to a maximum rate (e.g., 100 requests/minute). Reject excess requests with 429 Too Many Requests. Common algorithms: Token Bucket (allows bursts), Leaky Bucket (fixed rate), Sliding Window (accurate but complex).

💡 When to Use: Public APIs, multi-tenant systems, protecting expensive operations (search, reports), preventing abuse. Rate limit per user/tenant, not globally. Return 429 with Retry-After header.

Implementation

# Token Bucket Algorithm
class TokenBucket:
    """Rate limiter using token bucket"""

    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate  # Tokens per second
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        tokens_to_add = elapsed * self.refill_rate

        self.tokens = min(self.capacity, self.tokens + tokens_to_add)
        self.last_refill = now

    def consume(self, tokens=1):
        with self.lock:
            self._refill()

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

# Usage
limiter = TokenBucket(capacity=100, refill_rate=10)  # 100 burst, 10/sec

def handle_request():
    if not limiter.consume():
        raise RateLimitExceeded("Rate limit exceeded")

    # Process request
    return process()

# Sliding Window Rate Limiter (Redis)
def rate_limit_sliding_window(user_id, limit=100, window=60):
    """Sliding window with Redis sorted sets"""

    key = f"rate_limit:{user_id}"
    now = time.time()

    # Remove old entries
    redis.zremrangebyscore(key, 0, now - window)

    # Count requests in window
    count = redis.zcard(key)

    if count >= limit:
        return False  # Rate limited

    # Add new request
    redis.zadd(key, {str(uuid.uuid4()): now})
    redis.expire(key, window)

    return True

# Distributed rate limiting with leaky bucket
class LeakyBucket:
    """Rate limiter that processes at fixed rate"""

    def __init__(self, capacity, leak_rate):
        self.capacity = capacity
        self.leak_rate = leak_rate
        self.queue = []
        self.last_leak = time.time()

    def add_request(self, request):
        self._leak()

        if len(self.queue) >= self.capacity:
            raise RateLimitExceeded("Bucket full")

        self.queue.append(request)

    def _leak(self):
        now = time.time()
        elapsed = now - self.last_leak
        requests_to_process = int(elapsed * self.leak_rate)

        for _ in range(min(requests_to_process, len(self.queue))):
            request = self.queue.pop(0)
            process_request(request)

        self.last_leak = now

6. Fallback Pattern

🎯 Purpose: Provide degraded but functional responses when primary service fails.

The Problem: When external services fail (ML recommendation engine, payment gateway, search service), users see errors or blank pages. This creates terrible UX and lost revenue, even though you could provide simpler alternatives.

The Solution: Define fallback strategies - cached data, default values, simpler alternatives, or static content. If primary fails, automatically switch to fallback. Better to show "popular items" than no recommendations at all.

💡 When to Use: Services with acceptable degraded functionality (recommendations → popular items, ML → rules, real-time → cached), external dependencies you don't control. Often combined with circuit breaker (when circuit opens, use fallback).

Implementation

class FallbackHandler:
    """Provide degraded functionality when primary fails"""

    def __init__(self, primary, fallback):
        self.primary = primary
        self.fallback = fallback

    def execute(self, *args, **kwargs):
        try:
            return self.primary(*args, **kwargs)
        except Exception as e:
            logging.warning(f"Primary failed: {e}, using fallback")
            return self.fallback(*args, **kwargs)

# Example: Recommendation system
def get_recommendations_ml(user_id):
    """ML-based recommendations (primary)"""
    return ml_service.get_recommendations(user_id)

def get_recommendations_popular(user_id):
    """Popular items (fallback)"""
    return cache.get("popular_items", [])

recommendations_service = FallbackHandler(
    primary=get_recommendations_ml,
    fallback=get_recommendations_popular
)

recommendations = recommendations_service.execute(user_id)

# Multi-level fallback
class MultiLevelFallback:
    def __init__(self, *strategies):
        self.strategies = strategies

    def execute(self, *args, **kwargs):
        for strategy in self.strategies:
            try:
                return strategy(*args, **kwargs)
            except Exception as e:
                logging.warning(f"Strategy {strategy.__name__} failed: {e}")
                continue

        raise AllStrategiesFailed("All fallback strategies failed")

# Usage
service = MultiLevelFallback(
    get_from_cache,           # Try cache first
    get_from_database,        # Then database
    get_from_backup_db,       # Then backup
    get_default_value         # Finally, default
)

data = service.execute(key)

7. Health Checks

Health checks are endpoints that report the status of a service. Load balancers, orchestrators (Kubernetes), and service meshes use health checks to route traffic only to healthy instances and restart unhealthy ones.

# Health check types
from datetime import datetime
from enum import Enum


class HealthStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

class HealthCheck:
    """Service health monitoring"""

    def __init__(self):
        self.checks = {}

    def register_check(self, name, check_func):
        """Register a health check function"""
        self.checks[name] = check_func

    def check_health(self):
        """Run all health checks"""
        results = {}
        overall_status = HealthStatus.HEALTHY

        for name, check_func in self.checks.items():
            try:
                status, details = check_func()
                results[name] = {
                    "status": status.value,
                    "details": details
                }

                # Determine overall status
                if status == HealthStatus.UNHEALTHY:
                    overall_status = HealthStatus.UNHEALTHY
                elif status == HealthStatus.DEGRADED and overall_status != HealthStatus.UNHEALTHY:
                    overall_status = HealthStatus.DEGRADED

            except Exception as e:
                results[name] = {
                    "status": HealthStatus.UNHEALTHY.value,
                    "error": str(e)
                }
                overall_status = HealthStatus.UNHEALTHY

        return {
            "status": overall_status.value,
            "checks": results,
            "timestamp": datetime.now(timezone.utc).isoformat()
        }

# Example health checks
def check_database():
    """Check database connectivity"""
    try:
        db.execute("SELECT 1")
        return HealthStatus.HEALTHY, "Database connected"
    except Exception as e:
        return HealthStatus.UNHEALTHY, f"Database error: {e}"

def check_redis():
    """Check Redis cache"""
    try:
        redis.ping()
        return HealthStatus.HEALTHY, "Redis connected"
    except Exception as e:
        return HealthStatus.DEGRADED, f"Cache unavailable: {e}"

def check_disk_space():
    """Check available disk space"""
    usage = disk_usage("/")
    percent_used = usage.percent

    if percent_used > 90:
        return HealthStatus.UNHEALTHY, f"Disk {percent_used}% full"
    elif percent_used > 80:
        return HealthStatus.DEGRADED, f"Disk {percent_used}% full"
    else:
        return HealthStatus.HEALTHY, f"Disk {percent_used}% used"

def check_dependencies():
    """Check external service dependencies"""
    try:
        response = requests.get("http://payment-service/health", timeout=2)
        if response.status_code == 200:
            return HealthStatus.HEALTHY, "Payment service healthy"
        else:
            return HealthStatus.DEGRADED, f"Payment service status {response.status_code}"
    except Exception as e:
        return HealthStatus.DEGRADED, f"Payment service unreachable: {e}"

# Flask/FastAPI health endpoint example
@app.route("/health")
def health_endpoint():
    """
    Health check endpoint for load balancers
    Returns 200 if healthy, 503 if unhealthy
    """
    health_checker = HealthCheck()
    health_checker.register_check("database", check_database)
    health_checker.register_check("redis", check_redis)
    health_checker.register_check("disk", check_disk_space)
    health_checker.register_check("payment_service", check_dependencies)

    result = health_checker.check_health()

    status_code = 200 if result["status"] == "healthy" else 503
    return jsonify(result), status_code

# Kubernetes liveness vs readiness probes
@app.route("/health/liveness")
def liveness_probe():
    """
    Liveness: Is the service running?
    If fails, Kubernetes restarts the pod
    """
    # Basic check - is process alive?
    return {"status": "alive"}, 200

@app.route("/health/readiness")
def readiness_probe():
    """
    Readiness: Is the service ready to handle traffic?
    If fails, Kubernetes removes from service endpoints
    """
    # Comprehensive checks - can we actually serve traffic?
    health_checker = HealthCheck()
    health_checker.register_check("database", check_database)
    health_checker.register_check("cache", check_redis)

    result = health_checker.check_health()

    if result["status"] in ["healthy", "degraded"]:
        return result, 200
    else:
        return result, 503

# Graceful shutdown pattern
import signal
import sys

class GracefulShutdown:
    def __init__(self):
        self.is_shutting_down = False
        self.active_requests = 0
        signal.signal(signal.SIGTERM, self.handle_shutdown)

    def handle_shutdown(self, signum, frame):
        """Handle graceful shutdown signal"""
        logging.info("Shutdown signal received, draining connections...")
        self.is_shutting_down = True

        # Wait for active requests to complete
        while self.active_requests > 0:
            logging.info(f"Waiting for {self.active_requests} requests to complete...")
            time.sleep(1)

        logging.info("Shutdown complete")
        sys.exit(0)

    @app.before_request
    def before_request(self):
        if self.is_shutting_down:
            return {"error": "Service shutting down"}, 503
        self.active_requests += 1

    @app.after_request
    def after_request(self, response):
        self.active_requests -= 1
        return response

shutdown_handler = GracefulShutdown()

💡 Best Practices: Separate liveness (is it alive?) from readiness (can it serve traffic?). Liveness failures trigger restarts. Readiness failures remove from load balancer but don't restart the service. This prevents cascading failures during temporary issues.

Combining Resilience Patterns: Defense in Depth

These patterns work best when combined. A resilient service typically uses multiple patterns together:

Typical Production Setup

Layer 1: Prevent Overload

Rate Limiting: Reject excess requests early
Bulkhead: Isolate different workloads
Health Checks: Remove unhealthy instances

Layer 2: Handle Transient Failures

Timeout: Don't wait forever for responses
Retry: Automatically retry transient errors
Circuit Breaker: Fail fast when service is down

Layer 3: Graceful Degradation

Fallback: Provide degraded but functional responses (cached data, defaults, simpler alternatives)
Health Checks: Signal when service can't handle traffic (readiness = false)

🔁 Example Flow: Request comes in → Rate limiter checks quota →Timeout set to 5s → Circuit breaker checks if service healthy → If open, use fallback immediately → Otherwise, call service via bulkhead thread pool → If fails, retry with backoff → If still failing, open circuit and use fallback

Pattern Dependencies

Circuit Breaker + Fallback

When circuit opens, fallback provides degraded functionality instead of errors.

Timeout + Retry

Timeout detects hung requests. Retry with backoff handles transient failures.

Bulkhead + Circuit Breaker

Bulkhead limits blast radius. Circuit breaker stops calling failing services.

Rate Limiting + Health Checks

Rate limit prevents overload. Health checks remove overloaded instances from rotation.

💡 Implementation Priority: Start with 1) Timeouts (prevent hung requests),2) Health Checks (remove unhealthy instances), 3) Circuit Breaker (fail fast). Then add 4) Retry, 5) Rate Limiting, 6) Bulkhead,7) Fallback as needed. Don't implement all at once - add patterns as you encounter specific problems.

Idempotency: Safe Retries

An idempotent operation produces the same result no matter how many times it's executed. Critical for distributed systems where network failures cause retries. Without idempotency, retries can cause duplicate charges, double inventory deductions, or inconsistent state.

💡 The Problem: Client sends request → Server processes it → Response lost in network → Client retries → Server processes again → Duplicate operation!

# NOT idempotent - multiple calls create multiple charges
def charge_customer(user_id, amount):
    transaction_id = uuid.uuid4()
    payment_gateway.charge(user_id, amount, transaction_id)
    db.insert("transactions", {
        "transaction_id": transaction_id,
        "user_id": user_id,
        "amount": amount
    })
    return transaction_id

# If client retries due to timeout, customer charged twice!

# IDEMPOTENT - using idempotency key
def charge_customer_idempotent(user_id, amount, idempotency_key):
    """Client provides idempotency_key (e.g., UUID)"""

    # Check if we've already processed this request
    existing = db.query(
        "SELECT * FROM transactions WHERE idempotency_key = ?",
        idempotency_key
    )

    if existing:
        # Already processed - return existing result
        return existing["transaction_id"]

    # First time seeing this key - process normally
    transaction_id = uuid.uuid4()
    payment_gateway.charge(user_id, amount, transaction_id)

    db.insert("transactions", {
        "transaction_id": transaction_id,
        "user_id": user_id,
        "amount": amount,
        "idempotency_key": idempotency_key  # Store the key
    })

    return transaction_id

# Client usage
idempotency_key = str(uuid.uuid4())  # Client generates unique key
charge_customer_idempotent(user_id, 100.00, idempotency_key)
# If this times out and client retries with SAME key, no duplicate charge

# Idempotent API design pattern
@app.route("/api/orders", methods=["POST"])
def create_order():
    idempotency_key = request.headers.get("Idempotency-Key")

    if not idempotency_key:
        return {"error": "Idempotency-Key header required"}, 400

    # Check if order already exists for this key
    existing_order = Order.query.filter_by(
        idempotency_key=idempotency_key
    ).first()

    if existing_order:
        # Return existing order instead of creating new one
        return jsonify(existing_order.to_dict()), 200

    # Create new order
    order = Order(
        id=uuid.uuid4(),
        user_id=request.json["user_id"],
        items=request.json["items"],
        idempotency_key=idempotency_key
    )
    db.session.add(order)
    db.session.commit()

    return jsonify(order.to_dict()), 201

# HTTP Methods and Idempotency:
# GET     - Idempotent (read only)
# PUT     - Idempotent (replace entire resource)
# DELETE  - Idempotent (deleting twice = same result)
# POST    - NOT idempotent (creates new resource)
# PATCH   - May or may not be idempotent (depends on operation)

Idempotency Strategies

Client-Generated IDs

Client generates unique request ID (UUID). Server checks if ID was already processed.

Natural Idempotency

Design operations to be naturally idempotent (SET value = X, DELETE resource, UPDATE).

Distributed Cache

Store idempotency keys in Redis with TTL (24 hours). Fast lookup, automatic cleanup.

Database Unique Constraint

UNIQUE constraint on idempotency key. Database prevents duplicates automatically.

⚠️ Critical: Always implement idempotency for operations that modify state in distributed systems. Network failures WILL happen, clients WILL retry. Without idempotency, you'll have data corruption, duplicate charges, and angry customers.

Distributed Locks: Coordination Primitive

Distributed locks ensure only one process performs a critical operation across multiple nodes. Essential for preventing race conditions, duplicate processing, and maintaining consistency in distributed systems. Common use cases: cron jobs, batch processing, leader election.

# Redis-based distributed lock (simple version)
import redis
import time
import uuid

class DistributedLock:
    """Simple distributed lock using Redis"""

    def __init__(self, redis_client, lock_name, timeout=10):
        self.redis = redis_client
        self.lock_name = f"lock:{lock_name}"
        self.timeout = timeout
        self.lock_id = str(uuid.uuid4())  # Unique ID for this lock holder

    def acquire(self, blocking=True, timeout=None):
        """Acquire the lock"""
        end_time = time.time() + timeout if timeout else None

        while True:
            # Try to set key with NX (only if not exists) and EX (expiration)
            acquired = self.redis.set(
                self.lock_name,
                self.lock_id,
                nx=True,  # Only set if key doesn't exist
                ex=self.timeout  # Auto-release after timeout
            )

            if acquired:
                return True

            if not blocking:
                return False

            if end_time and time.time() > end_time:
                return False

            time.sleep(0.1)  # Wait before retry

    def release(self):
        """Release the lock (only if we own it)"""
        # Lua script ensures atomic check-and-delete
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(lua_script, 1, self.lock_name, self.lock_id)

    def __enter__(self):
        self.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

# Usage: Prevent duplicate cron job execution
redis_client = redis.Redis(host='localhost', port=6379)

def run_daily_report():
    """This runs on multiple servers, but should only run once"""

    lock = DistributedLock(redis_client, "daily_report_lock", timeout=3600)

    if not lock.acquire(blocking=False):
        print("Another instance already running, skipping...")
        return

    try:
        # Only one server executes this
        generate_report()
        send_to_users()
    finally:
        lock.release()

# Context manager pattern
def process_payment(order_id):
    """Prevent duplicate payment processing"""

    with DistributedLock(redis_client, f"payment:{order_id}", timeout=30):
        # Only one process can execute this at a time
        charge_customer(order_id)
        update_order_status(order_id)

# Redlock Algorithm (robust distributed lock)
from redlock import Redlock

# Connect to multiple Redis instances for fault tolerance
redlock = Redlock([
    {"host": "redis1", "port": 6379},
    {"host": "redis2", "port": 6379},
    {"host": "redis3", "port": 6379},
])

lock = redlock.lock("resource_name", 10000)  # 10 second TTL

if lock:
    try:
        # Critical section
        perform_critical_operation()
    finally:
        redlock.unlock(lock)
else:
    print("Could not acquire lock")

# Database-based distributed lock (PostgreSQL)
def acquire_pg_lock(lock_id, timeout=10):
    """
    Advisory locks in PostgreSQL
    - Fast (in-memory)
    - Automatically released on connection close
    """
    try:
        # Try to acquire lock with timeout
        result = db.execute(
            "SELECT pg_try_advisory_lock(%s)",
            (lock_id,)
        ).fetchone()

        return result[0]  # True if acquired, False otherwise
    except Exception as e:
        logging.error(f"Lock acquisition failed: {e}")
        return False

def release_pg_lock(lock_id):
    db.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))

# Usage
REPORT_LOCK_ID = 12345

if acquire_pg_lock(REPORT_LOCK_ID):
    try:
        generate_report()
    finally:
        release_pg_lock(REPORT_LOCK_ID)
else:
    print("Report already being generated")

Lock Implementations Comparison

Implementation	Pros	Cons	Best For
Redis Single Instance	Simple, fast	Single point of failure	Non-critical locks, development
Redlock (Multi-Redis)	Fault tolerant, proven	Complex, needs 3+ Redis instances	Production critical locks
PostgreSQL Advisory	If already using Postgres, fast	Tied to DB connection	App using Postgres
ZooKeeper	Highly reliable, battle-tested	Heavy, operational complexity	Large-scale coordination
etcd	Kubernetes-native, reliable	Requires etcd cluster	Kubernetes environments

⚠️ Lock Pitfalls: Always set timeouts (prevent deadlocks). Always release locks in finally blocks. Use unique lock IDs to prevent accidental release by another process. Test lock timeout scenarios - what happens if holder crashes?

Distributed Tracing: Debugging Across Services

Distributed tracing tracks requests as they flow through multiple services. Each service adds timing information and context, creating a complete picture of the request lifecycle. Essential for debugging performance issues in microservices.

# OpenTelemetry: Standard for distributed tracing
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Setup tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Export via OTLP (works with Jaeger, Zipkin, and other backends)
otlp_exporter = OTLPSpanExporter(endpoint="localhost:4317")
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(otlp_exporter)
)

# Instrument your code
class OrderService:
    @tracer.start_as_current_span("create_order")
    def create_order(self, user_id, items):
        span = trace.get_current_span()
        span.set_attribute("user_id", user_id)
        span.set_attribute("item_count", len(items))

        # Each operation creates a child span
        with tracer.start_as_current_span("validate_order"):
            self.validate(items)

        with tracer.start_as_current_span("call_payment_service"):
            payment_result = self.payment_service.charge(user_id, total)
            span.set_attribute("payment_status", payment_result["status"])

        with tracer.start_as_current_span("save_to_database"):
            order = self.db.save_order(user_id, items)

        return order

# Trace context propagated automatically across services
# Service A creates trace → Service B continues trace → Service C continues trace
# View entire request flow in Jaeger UI

# Trace example:
# Span 1: API Gateway (100ms)
#   Span 2: Order Service (80ms)
#     Span 3: Validate Order (5ms)
#     Span 4: Call Payment Service (50ms)
#       Span 5: Payment Service Process (45ms)
#         Span 6: Database Query (40ms) ← Bottleneck!
#     Span 7: Save to Database (20ms)

# Benefits:
# - See exact timing of each operation
# - Identify slow services/operations
# - Track errors across services
# - Understand service dependencies

Consensus Algorithms: Agreement in Distributed Systems

Consensus algorithms allow distributed systems to agree on shared state despite failures. Critical for leader election, distributed coordination, and maintaining consistency.

# Raft Consensus (simplified conceptual example)
class RaftNode:
    """Simplified Raft consensus algorithm"""
    def __init__(self, node_id, peers):
        self.node_id = node_id
        self.peers = peers
        self.state = "FOLLOWER"  # FOLLOWER, CANDIDATE, LEADER
        self.current_term = 0
        self.voted_for = None
        self.log = []
        self.commit_index = 0

    def start_election(self):
        """Become candidate and request votes"""
        self.state = "CANDIDATE"
        self.current_term += 1
        self.voted_for = self.node_id

        votes = 1  # Vote for self

        for peer in self.peers:
            if peer.request_vote(self.current_term, self.node_id):
                votes += 1

        # Majority wins
        if votes > len(self.peers) / 2:
            self.become_leader()

    def become_leader(self):
        """Elected as leader"""
        self.state = "LEADER"
        # Send heartbeats to maintain leadership
        self.send_heartbeats()

    def append_entry(self, entry):
        """Leader appends to log and replicates"""
        if self.state != "LEADER":
            raise NotLeaderError()

        self.log.append(entry)

        # Replicate to followers
        acks = 1  # Leader acks itself
        for peer in self.peers:
            if peer.append_entries(self.log):
                acks += 1

        # Commit when majority acknowledges
        if acks > len(self.peers) / 2:
            self.commit_index = len(self.log) - 1

# Zookeeper: Distributed coordination
from kazoo.client import KazooClient

zk = KazooClient(hosts='localhost:2181')
zk.start()

# 1. LEADER ELECTION
election = zk.Election("/election", "service-1")

def run_as_leader():
    """Called when elected as leader"""
    print("I am the leader!")
    # Do leader things

election.run(run_as_leader)

# 2. DISTRIBUTED LOCK
lock = zk.Lock("/locks/resource")

with lock:
    # Only one node executes this at a time
    process_shared_resource()

# 3. SERVICE DISCOVERY
@zk.ChildrenWatch("/services")
def watch_services(children):
    """Called when services join/leave"""
    print(f"Available services: {children}")

# Register this service
zk.create("/services/service-1", ephemeral=True)

# 4. CONFIGURATION MANAGEMENT
@zk.DataWatch("/config/database")
def watch_config(data, stat):
    """Called when config changes"""
    config = json.loads(data)
    update_database_config(config)

# etcd: Another popular coordination service
import etcd3

etcd = etcd3.client()

# Leader election with etcd
def leader_election():
    lease = etcd.lease(60)  # 60 second lease

    # Try to become leader (atomic compare-and-set)
    success, _ = etcd.transaction(
        compare=[etcd.transactions.create('/leader') == 0],
        success=[etcd.transactions.put('/leader', 'service-1', lease)],
        failure=[]
    )

    if success:
        # We're the leader!
        while lease.ttl > 0:
            do_leader_work()
            lease.refresh()
    else:
        # Watch for leader change
        watch_leader()

Consensus Use Cases: Leader election: Ensure one instance performs task. Distributed locks: Coordinate access to resources.Service discovery: Track available services. Configuration:Distribute config changes.

Pattern Comparison: Choosing the Right Tool

Different distributed systems patterns solve different problems. Here's a guide to choosing the right pattern for your situation.

Communication Patterns

Pattern	Best For	Avoid When
Synchronous REST/RPC	Immediate response needed, simple request-reply	Long-running operations, high decoupling needed
Message Queue	Async processing, decoupling, load smoothing	Need immediate response, simple architecture
Pub/Sub	Broadcasting events, multiple consumers	Point-to-point communication, guaranteed order
Event Streaming	Real-time data pipelines, event sourcing	Simple task queues, low throughput

Resilience Patterns - When to Use

Use Circuit Breaker When:

Calling external services that may fail
Want to fail fast instead of waiting
Need to prevent cascading failures
Have fallback options available

Use Retry When:

Transient failures are common (network blips)
Operation is idempotent
Can afford to wait for retry delays
Alternative: NOT when operation is expensive

Use Bulkhead When:

Different workloads with different priorities
One slow service shouldn't block others
Resource exhaustion is a risk

Use Rate Limiting When:

Protecting against abuse or DDoS
Enforcing API quotas
Preventing resource exhaustion
Fair usage across tenants

Caching Strategy Selector

Use Cache-Aside when: Read-heavy workload, data rarely changes

Use Write-Through when: Consistency critical, can afford write latency

Use Write-Behind when: High write throughput, eventual consistency OK

Use Refresh-Ahead when: Predictable access patterns, proactive freshness needed

💡 Golden Rule: Start simple. Add complexity (message queues, caching, service mesh) only when you have specific problems to solve. Premature optimization and over-engineering cause more issues than the problems they try to prevent.

Troubleshooting Distributed Systems

Distributed systems fail in unique and interesting ways. Here's how to diagnose and fix common issues.

Common Problems & Solutions

Problem: Service Timeouts

Symptoms: Requests hang, timeout errors, slow responses

Possible Causes:

Network latency or packet loss
Downstream service overloaded
Database connection pool exhausted
Infinite retries causing cascading failures

Solutions:

Add circuit breakers to fail fast
Set aggressive timeouts (2-5s for most APIs)
Check distributed tracing for bottlenecks
Monitor connection pool metrics

Problem: Data Inconsistency

Symptoms: Different services see different data, stale reads

Possible Causes:

Replication lag between primary and replicas
Cache invalidation not working
Race conditions in distributed updates
Event ordering issues in message queues

Solutions:

Read from primary for critical reads
Implement proper cache invalidation strategy
Use distributed locks for critical sections
Design for idempotency

Problem: Message Queue Backlog Growing

Symptoms: Queue depth increasing, messages delayed, eventual OOM

Possible Causes:

Consumers can't keep up with producers
Consumer crashes or poison messages
Database/downstream service slow

Solutions:

Scale up number of consumers (horizontal scaling)
Implement dead letter queue for poison messages
Add backpressure to slow down producers
Optimize consumer processing logic

Problem: Cascading Failures

Symptoms: One service fails, then others fail, entire system down

Possible Causes:

No circuit breakers - requests pile up
Synchronous calls create tight coupling
Shared resources (DB) become bottleneck

Solutions:

Implement circuit breakers everywhere
Use bulkhead pattern to isolate failures
Degrade gracefully with fallback responses
Use async communication (message queues)

Debugging Checklist

✅ Check Metrics

Error rates (4xx, 5xx)
Latency percentiles (p50, p95, p99)
Throughput (requests/sec)
Resource utilization (CPU, memory, connections)

✅ Check Logs

Correlation IDs to trace requests
Error messages and stack traces
Timing information
Business logic flow

✅ Check Distributed Tracing

Service dependency graph
Slow spans (bottlenecks)
Error propagation
Cross-service latency

✅ Check Infrastructure

Network connectivity
DNS resolution
Load balancer health checks
Database connection pools

⚠️ Pro Tip: When debugging, start with the metrics. Metrics tell you WHAT is broken. Logs tell you WHY. Distributed tracing tells you WHERE. Use all three together for fastest root cause analysis.