System Design at Scale

The Pendulum Has Swung Back

In 2015, every engineering blog post was about how your monolith was holding you back and microservices were the path to salvation. By 2023, those same companies were quietly publishing post-mortems about how microservices made their systems 10x harder to operate, debug, and deploy. The truth, as always, is that the right architecture depends on your context: team size, deployment maturity, traffic patterns, and organizational structure.

The core insight, articulated by Melvin Conway in 1967 as Conway's Law, is that systems tend to mirror the communication structure of the organizations that build them. Amazon famously restructured into small, autonomous "two-pizza teams" before decomposing their system into services. The organizational change came first. Companies that try to adopt microservices without first restructuring their organization end up with a distributed monolith — the worst of both worlds.

When Monoliths Win

Shopify ran on a Rails monolith for 15 years and scaled to $6B in GMV before undertaking their current modularization effort. Stack Overflow serves millions of developers from a surprisingly small cluster of servers with a monolith. Basecamp, Stack Exchange, and many other highly successful companies have consciously chosen monoliths and are thriving. The reasons monoliths win in many scenarios:

• Developer velocity: A single codebase means you can refactor across module boundaries in one commit. In a microservices world, cross-service changes require coordinating deployments across multiple repos, sometimes with API versioning gymnastics.
• Debugging: A stack trace in a monolith tells you exactly what happened. In a microservices world, you need distributed tracing (Jaeger, Zipkin, OpenTelemetry), and even then, reconstructing the call chain across 12 services is painful.
• Transactions: ACID transactions within a monolith are trivially easy. Distributed transactions across microservices require saga patterns, 2PC (with all its failure modes), or accepting eventual consistency — none of which are trivial.
• Operational overhead: A monolith needs one deployment pipeline, one monitoring dashboard, one alerting setup. Microservices multiply each of these by the number of services.

The Modular Monolith Pattern

A modular monolith is a single deployable unit where code is organized into well-isolated modules with explicit public interfaces and no direct cross-module database access. The key discipline is treating module boundaries as if they were service boundaries — even though at runtime it's all one process.

ecommerce-monolith/
├── modules/
│   ├── orders/
│   │   ├── public/          # The ONLY interface other modules may use
│   │   │   ├── orders_api.rb
│   │   │   └── order_placed_event.rb
│   │   ├── internal/        # Private — no cross-module imports allowed
│   │   │   ├── order_repository.rb
│   │   │   ├── payment_processor.rb
│   │   │   └── fraud_checker.rb
│   │   └── db/
│   │       └── migrations/  # Only orders module touches orders_* tables
│   ├── inventory/
│   │   ├── public/
│   │   │   ├── inventory_api.rb
│   │   │   └── stock_reserved_event.rb
│   │   └── internal/
│   └── customers/
│       └── public/
├── shared/
│   ├── event_bus.rb         # In-process pub/sub for inter-module comms
│   └── domain_event.rb
└── config/
    └── module_boundaries.rb # Zeitwerk + custom loader enforces isolation

The crucial enforcement mechanism is build-time isolation checking. In Ruby, tools like Packwerk from Shopify analyze constant references and flag violations. In Java, modules via Project Jigsaw or ArchUnit tests can enforce that com.acme.orders.internal is never imported by any class outside the orders package. In Go, internal packages provide this natively.

The Strangler Fig Pattern

Martin Fowler named this after the strangler fig tree, which grows around an existing tree, eventually replacing it. The pattern describes how to incrementally migrate from a monolith to microservices (or from one system to another) without a big-bang rewrite.

The key to a successful strangler fig migration is the proxy/facade layer. In practice, this is often an API gateway (Kong, Nginx, or a thin custom service) that routes traffic based on URL path or request attributes. You never have to make a big-bang cutover — you can migrate one endpoint at a time, verify with feature flags, and roll back if something goes wrong.

Service Boundaries: Domain-Driven Design

The hardest part of microservices is figuring out where to draw the boundaries. Chop too fine and you get hundreds of nanoservices with massive operational overhead. Chop too coarse and services become coupled "distributed monoliths." Domain-Driven Design (DDD) gives us the tool: bounded contexts.

A bounded context is a subsystem where a particular domain model applies consistently. The word "order" means something specific in the context of an e-commerce system: it's a cluster of order lines, shipping details, and payment information. But to the warehouse management system, an "order" is a pick list. These two uses of the word "order" represent different bounded contexts, and trying to share one "Order" model across both leads to a God Object that satisfies neither.

Data Ownership: The Hardest Problem

If there is one single rule for microservices that teams violate most often, it is this: each service owns its data exclusively. No service may directly read from or write to another service's database. All access goes through the service's API or event stream. Violate this and you've built a distributed monolith — all the operational complexity of microservices with the coupling of a monolith.

Why is this so hard? Because it removes the ability to do multi-table JOINs. The Customer Service owns customer data. The Order Service owns order data. To show an order with customer details, the Order Service must either: (1) call the Customer Service via API and join in application code, (2) subscribe to customer events and maintain a local read model with relevant customer fields, or (3) accept denormalization and store customer name/email in the order at creation time.

Option 3 — denormalization at write time — is what most high-scale systems actually do. When you place an order on Amazon, your shipping address is copied into the order record. This means the order record remains accurate even if you later change your address. It also means the Order Service never needs to call the Customer Service to display an order's shipping address.

The Distributed Monolith Anti-Pattern

You've deployed 50 microservices. They all share the same database (or each service's code directly queries another service's database tables). Deployments require a specific order because Service A depends on Schema version 4.2 of Service B's tables. You've achieved the latency of microservices, the operational overhead of microservices, but none of the independence or scalability benefits. Congratulations, you've built a distributed monolith.

Netflix: The Microservices Benchmark

Netflix is the canonical microservices success story — but the full context is usually omitted. Netflix made the transition to microservices starting around 2009, running the migration over nearly 7 years. By 2016, they had over 700 microservices. The key enablers were: a world-class platform engineering team that built Hystrix (circuit breaker), Eureka (service discovery), Ribbon (load balancing), and Zuul (API gateway) — all open-sourced; a chaos engineering practice (Chaos Monkey, Chaos Kong) that continuously validated resilience; and a strong culture of operational ownership where service teams were on-call for their own services 24/7.

Netflix runs roughly 150,000 instances in AWS. Their edge traffic goes through Open Connect CDN nodes, then into their API gateway, then fans out to dozens of microservices per request. The device team might call 20 microservices to render a single home screen, with aggressive parallelism and graceful degradation (if the personalization service is slow, show popular titles instead). This architecture works for Netflix because they have the platform engineering depth to support it. Most companies don't.

When Your Database Becomes the Bottleneck

A well-tuned PostgreSQL instance on modern hardware (64 cores, 256GB RAM, NVMe SSDs) can handle roughly 50,000–100,000 transactions per second for mixed OLTP workloads, with a dataset size limited to what fits within your storage tier. For the majority of applications, this is more than enough. Instagram served 300 million daily users for years on a relatively small PostgreSQL cluster with aggressive read replicas and caching.

But when you cross certain thresholds — write throughput that saturates a single master, dataset sizes that no longer fit on one server, or regulatory requirements for data locality across geographies — sharding becomes necessary. Sharding is the practice of partitioning data across multiple database instances, where each instance owns a distinct subset of the data. It is complex, operationally demanding, and introduces fundamental limitations on the queries you can efficiently run. You should delay sharding as long as architecturally possible.

Vertical vs Horizontal Partitioning

Vertical partitioning splits a table by columns — moving less-frequently-accessed columns to a separate table or storage tier. For example, separating a user's large bio text blob from their core profile fields (id, email, name) means the hot-path query that loads a user profile doesn't waste I/O reading the bio every time. This is essentially normalization taken to an operational extreme.

Horizontal partitioning (sharding) splits a table by rows, distributing subsets of rows across multiple servers. This is what most people mean when they say "sharding." All shards have the same schema; they just hold different rows. The challenge is choosing the partition key and partition strategy wisely — a bad choice here is extremely painful to fix later.

Consistent Hashing

The classic hash-based sharding formula shard = hash(key) % N has a fatal flaw: when N changes (you add or remove a shard), almost every key remaps to a different shard, requiring a massive data migration. Consistent hashing solves this by placing both keys and nodes on a conceptual ring, so adding or removing a node only remaps the keys that were assigned to that node — approximately 1/N of the total keys.

import hashlib
from sortedcontainers import SortedDict

class ConsistentHashRing:
    def __init__(self, virtual_nodes=150):
        # virtual_nodes: how many points per physical node on the ring
        # Higher = more even distribution, more memory
        self.virtual_nodes = virtual_nodes
        self.ring = SortedDict()  # hash_position -> node_id
        self.nodes = set()

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node_id: str):
        self.nodes.add(node_id)
        for i in range(self.virtual_nodes):
            vnode_key = f"{node_id}-vnode-{i}"
            pos = self._hash(vnode_key)
            self.ring[pos] = node_id

    def remove_node(self, node_id: str):
        self.nodes.discard(node_id)
        for i in range(self.virtual_nodes):
            vnode_key = f"{node_id}-vnode-{i}"
            pos = self._hash(vnode_key)
            self.ring.pop(pos, None)

    def get_node(self, key: str) -> str:
        if not self.ring:
            raise ValueError("No nodes in ring")
        pos = self._hash(key)
        # Find the first node clockwise from this position
        idx = self.ring.bisect_left(pos)
        if idx == len(self.ring):
            idx = 0  # Wrap around the ring
        return self.ring.values()[idx]

# Usage: add 3 nodes, each with 150 virtual nodes = 450 ring positions
ring = ConsistentHashRing(virtual_nodes=150)
ring.add_node("db-shard-1")
ring.add_node("db-shard-2")
ring.add_node("db-shard-3")
# Adding db-shard-4 only remaps ~25% of keys, not 75%

Rebalancing Strategies

When you add capacity to a sharded cluster, you need to move data from existing shards to the new one. This process — rebalancing — must happen without taking the cluster offline and without causing read/write errors for clients. The practical approaches:

Fixed number of partitions: Kafka and Redis Cluster use a fixed number of logical partitions (16,384 for Redis). Nodes are assigned ownership of partitions. Adding a node means reassigning some partitions to it — no key remapping needed, just partition ownership changes. The data moves, but the partition key assignment is stable.

Dynamic partitioning: HBase and MongoDB automatically split a partition when it grows beyond a configured size (default 256MB in HBase). A single large partition becomes two smaller ones. This is transparent to clients but requires coordination to avoid split storms.

Manual rebalancing with hot key awareness: In many real production scenarios, you identify hot keys (shards with disproportionate traffic) and manually move them to dedicated shards. Instagram identified that celebrity accounts represented a small number of users but an enormous fraction of reads, and gave them dedicated infrastructure.

Cross-Shard Queries: The Achilles' Heel

Sharding's greatest weakness becomes apparent the moment you need to query across shard boundaries. If users are sharded by user_id and you need to find all users who signed up in the last 7 days, you must query ALL shards in parallel and merge the results in your application. This is called a scatter-gather query and it's expensive both in latency and load on your database tier.

import asyncio
from typing import List

async def scatter_gather_query(shards: List, query: str, params: dict):
    # Fan out to all shards in parallel
    tasks = [shard.execute(query, params) for shard in shards]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    merged = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            # Log but don't fail the entire request on one shard outage
            log_shard_error(shards[i], result)
            continue
        merged.extend(result)

    # Re-sort if query has ORDER BY (each shard sorted independently)
    merged.sort(key=lambda r: r['created_at'], reverse=True)
    return merged[:100]  # Apply global LIMIT after merge

# The problem: if you LIMIT 20 per shard across 10 shards,
# you fetch 200 rows to return 20. Deep pagination is catastrophic:
# OFFSET 10000 LIMIT 20 across 10 shards = 100,020 rows fetched.

The standard mitigation is to maintain a separate search index (Elasticsearch, Solr) for cross-shard queries. Your application writes to the shard for writes and reads, but complex search queries go to Elasticsearch which has its own index across all data. This is exactly how Instagram handles search — Postgres shards for user data, Elasticsearch for username/bio search.

Vitess: MySQL Sharding at YouTube Scale

Vitess was built at YouTube in 2010 to solve their MySQL scaling problems and is now a CNCF graduated project used by Slack, GitHub, Pinterest, and many others. Vitess sits between your application and MySQL, providing connection pooling, query routing, and transparent sharding with a MySQL-compatible wire protocol. Your application thinks it's talking to one MySQL instance; Vitess handles the sharding complexity.

Vitess handles resharding transparently: it can double the number of shards by splitting each shard in half, using a VReplication mechanism that streams changes from the old shard to the new ones. The cutover window is typically under 5 seconds. YouTube went from 1 to 4 to 8 to 16 shards over several years, each time using Vitess's online resharding.

When NOT to Shard

Before sharding, exhaust every other option: connection pooling (PgBouncer can 10x your effective connection capacity), read replicas (offload 80%+ of traffic which is typically reads), caching (eliminate database round-trips entirely for hot data), query optimization (a missing index can make a query 100x slower), and vertical scaling (a 128-core machine with 2TB RAM can run PostgreSQL in a surprising number of use cases).

The Numbers That Make Caching Non-Negotiable

RAM latency: ~100 nanoseconds. NVMe SSD latency: ~100 microseconds. Network round-trip to a Redis node in the same datacenter: ~200 microseconds. Disk-backed database query (uncached): ~10 milliseconds. These numbers represent 100x differences between tiers. When you're serving 100,000 requests per second and each request needs 10 data lookups, the difference between a 60% cache hit rate and a 99% cache hit rate is the difference between needing 10 database servers and needing 1.

Facebook reported that their Memcached tier handles over 1 billion requests per second across their fleet, with cache hit rates above 99% for most access patterns. This is not an optimization — it's a foundational architectural requirement. Without caching, their database infrastructure would need to be orders of magnitude larger and still couldn't keep up with the read load.

Cache-Aside (Lazy Loading)

The most common caching pattern. The application checks the cache first; on a miss, it reads from the database and populates the cache. The cache is only populated with data that's actually requested, so you don't waste memory on cold data.

import redis
import json
from typing import Optional

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

def get_user(user_id: int) -> Optional[dict]:
    cache_key = f"user:{user_id}"

    # 1. Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache HIT — fast path

    # 2. Cache MISS — go to database
    user = db.query("SELECT * FROM users WHERE id = %s", [user_id])
    if user is None:
        # Cache negative results too! Prevents cache bypass attacks.
        # TTL of 60s so legitimate creates eventually show up.
        redis_client.setex(cache_key, 60, "__NULL__")
        return None

    # 3. Populate cache with TTL of 3600s (1 hour)
    redis_client.setex(cache_key, 3600, json.dumps(user))
    return user

def update_user(user_id: int, updates: dict):
    # Write to database first
    db.update("UPDATE users SET ... WHERE id = %s", [user_id])
    # Invalidate cache — don't update it (avoids race conditions)
    redis_client.delete(f"user:{user_id}")

Cache-aside is simple but has two failure modes worth knowing: cache stampede (many requests simultaneously hit a cold key and all go to the database) and stale data (the cache holds a value that the database has since updated). Cache stampede is addressed with probabilistic early expiration or a mutex lock on the first miss.

Read-Through and Write-Through

In read-through caching, the cache itself is responsible for fetching from the database on a miss — the application always talks to the cache, never directly to the database for reads. This simplifies application code and ensures the cache is always warm, but requires the cache to understand your data model.

Write-through caching updates both the cache and the database on every write, synchronously. This ensures the cache is always consistent but doubles write latency. Write-behind (or write-back) caching updates the cache immediately and writes to the database asynchronously, improving write latency but risking data loss if the cache node dies before the write is flushed.

Redis Architecture Deep Dive

Redis is single-threaded for command processing (by design) which eliminates locking overhead and makes its performance extremely predictable. Modern Redis (7.x) uses I/O threading for network reads/writes while keeping command execution single-threaded. On a single Redis node, you can expect 100,000–500,000 operations per second for simple get/set commands, with sub-millisecond P99 latency under normal conditions.

Redis Cluster (added in Redis 3.0) provides horizontal scaling via hash slots. The keyspace is divided into 16,384 slots. Each cluster node is responsible for a subset of slots. When a client sends a command for a key, the cluster topology (returned during initial CLUSTER NODES handshake) tells the client which node holds that key's slot. Smart clients (redis-py, Jedis, ioredis) cache this topology and route directly, avoiding an extra hop.

Redis persistence options matter for your durability/performance trade-off. RDB (Redis Database) snapshots fork the process and write a point-in-time snapshot to disk. Fast to restore but you can lose up to save 900 1 seconds of data. AOF (Append Only File) logs every write command. With appendfsync everysec, you risk losing at most 1 second of data. With appendfsync always, you're fully durable but cut throughput by roughly 10x. Most production setups use both: AOF for durability, RDB for fast restores after a complete failure.

Memcached vs Redis: The Honest Comparison

Memcached was the dominant caching solution before Redis. It's simpler, faster in some benchmarks for pure key-value workloads, and natively multi-threaded. But Redis has won the practical battle for most use cases due to its richer data structures (lists, sorted sets, hyperloglog, streams), persistence, pub/sub, Lua scripting, and cluster mode. The only scenario where Memcached still wins clearly is when you need multi-threaded cache performance at extreme scale and have only simple key-value needs — LinkedIn's cali team still uses Memcached for some hot tiers where Redis's single-threaded nature becomes a bottleneck.

Cache Stampede Prevention

Cache stampede (also called thundering herd or dog-pile effect) occurs when a popular cache key expires and hundreds or thousands of concurrent requests simultaneously go to the database to recompute the value. The database gets slammed with N identical queries when it would have been fine serving 1.

import random
import time

def get_with_probabilistic_early_expiry(key: str, ttl: int, beta: float = 1.0):
    """
    XFetch algorithm (Vattani et al., 2015).
    Probabilistically recompute the cache before it expires,
    so we always have a warm value ready.
    beta: higher = recompute sooner. Default 1.0.
    """
    cached_value, expiry_time, compute_time = redis_client.hmget(
        key, 'value', 'expiry', 'delta'
    )

    if cached_value:
        # time_to_expire is negative when already expired
        time_to_expire = float(expiry_time) - time.time()
        # early_recompute probability increases as expiry approaches
        early_recompute = (-float(compute_time) * beta *
                           random.uniform(0, float(compute_time)))
        if time_to_expire > early_recompute:
            return cached_value  # Use cached value

    # Recompute (one request at a time via distributed lock)
    lock_key = f"lock:{key}"
    if redis_client.set(lock_key, 1, nx=True, ex=30):
        start = time.time()
        new_value = expensive_computation()
        delta = time.time() - start
        expiry = time.time() + ttl
        redis_client.hset(key, mapping={
            'value': new_value, 'expiry': expiry, 'delta': delta
        })
        redis_client.expire(key, ttl + 30)
        redis_client.delete(lock_key)
        return new_value
    else:
        # Another request is recomputing — return stale value
        return cached_value or wait_and_retry(key)

CDN Caching and Multi-Tier Architecture

A production caching architecture is not a single Redis cluster — it's a hierarchy of caches, each serving a different purpose and trading off cost vs. freshness vs. capacity.

Cloudflare's CDN handles over 100 million requests per second globally, serving from 285+ edge locations. When a user in Tokyo requests a cacheable resource, Cloudflare's Tokyo PoP serves it directly — no round trip to your origin server in Virginia. Cache hit rates for static assets commonly exceed 95%. For API responses, you need careful cache key design (include user auth tier, locale, A/B test bucket in the cache key) to avoid serving wrong content.

L4 vs L7 Load Balancing

Load balancers operate at different layers of the network stack, and the choice fundamentally changes what they can do with your traffic. Understanding this distinction is critical because the wrong choice can either leave performance on the table or add unnecessary latency.

L4 (Transport Layer) load balancers work at the TCP/UDP level. They see IP addresses and port numbers, but not the content of the packets. They forward TCP connections to backend servers based on connection state. This makes them extremely fast — they can handle millions of connections per second with microsecond-level overhead. AWS Network Load Balancer (NLB) is L4. HAProxy in TCP mode is L4. F5 BIG-IP operating in NAT mode is L4. Use L4 when you need minimal latency overhead and don't need to inspect request content.

L7 (Application Layer) load balancers terminate the TCP connection, parse the HTTP request, and make routing decisions based on URL paths, headers, cookies, or request body content. This allows URL-based routing, header injection, A/B testing, SSL termination, and authentication. AWS Application Load Balancer (ALB), Nginx in proxy mode, and Envoy are L7. The overhead versus L4 is typically 1-3ms of additional latency but the routing flexibility is usually worth it for HTTP workloads.

Load Balancing Algorithms

The Power of Two Choices algorithm is worth understanding in depth because it elegantly solves the "join the shortest queue" problem in a distributed way. Instead of maintaining a global view of all backend loads (expensive), you pick two backends at random and route to whichever has fewer connections. This achieves near-optimal load distribution with O(1) complexity per routing decision. It's the algorithm used by Nginx Plus's least_conn in its upstream load balancing and by Envoy's LEAST_REQUEST policy.

Health Checks and Graceful Degradation

A load balancer without good health checks is worse than no load balancer at all — it will happily route traffic to dead backends. Health checks come in three tiers, each catching different failure modes:

• TCP health check: Can I open a TCP connection to port 8080? This catches total node failures but misses application-level errors (the process is running but returning 503s).
• HTTP health check: Does GET /health return 200 within 2 seconds? This catches application crashes and startup issues. Your /health endpoint should check critical dependencies (database connection pool health, not just "is the HTTP server up").
• Deep health check: Does the application successfully complete a representative operation? Write a test record to the database, read it back, delete it. This catches cases where the app is running but can't actually serve traffic.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio

app = FastAPI()

class HealthStatus(BaseModel):
    status: str
    database: str
    cache: str
    version: str

@app.get("/health")
async def health_check() -> HealthStatus:
    # Run checks in parallel with timeout
    try:
        db_ok, cache_ok = await asyncio.gather(
            asyncio.wait_for(check_database(), timeout=2.0),
            asyncio.wait_for(check_redis(), timeout=1.0),
        )
    except asyncio.TimeoutError:
        # Don't let health check block indefinitely
        raise HTTPException(status_code=503, detail="Health check timeout")

    if not db_ok:
        raise HTTPException(status_code=503, detail="Database unhealthy")

    return HealthStatus(
        status="healthy",
        database="ok" if db_ok else "degraded",
        cache="ok" if cache_ok else "degraded",
        version="2.4.1"
    )

@app.get("/health/live")
async def liveness():
    # Kubernetes liveness: just "is the process alive?"
    # Never check external dependencies here — causes cascading restarts
    return {"status": "alive"}

Global Server Load Balancing (GSLB)

When you have datacenters in multiple regions, GSLB is the mechanism that routes users to the nearest (or healthiest) datacenter. It typically operates via DNS: a user's DNS lookup returns different IP addresses depending on their location, the health of each datacenter, and your traffic routing policy.

Cloudflare's Anycast routing is a more sophisticated approach: the same IP address is announced from multiple physical locations. BGP routing naturally directs users to the nearest Cloudflare PoP. This is how Cloudflare achieves 30ms average time-to-first-byte from most global locations — the DNS resolution itself points you close to where you are.

Envoy Proxy Deep Dive

Envoy is the backbone of modern service mesh architectures. Originally built at Lyft, it's now the data plane for Istio, AWS App Mesh, and many custom service meshes. What makes Envoy special is its xDS API — a standardized, dynamic configuration protocol that allows a control plane (like Istio's Pilot) to push new configuration to all Envoy instances simultaneously without a restart.

# Envoy: Circuit breaker + retry policy for an upstream cluster
static_resources:
  clusters:
  - name: order_service
    type: STRICT_DNS
    lb_policy: LEAST_REQUEST
    circuit_breakers:
      thresholds:
      - priority: DEFAULT
        max_connections: 1000       # Max concurrent TCP connections
        max_pending_requests: 100   # Max requests waiting for conn
        max_requests: 1000          # Max active requests
        max_retries: 3
    upstream_http_filters:
    - name: retry_policy
      typed_config:
        retry_on: "5xx,connect-failure,reset"
        num_retries: 3
        retry_back_off:
          base_interval: 0.025s    # 25ms initial backoff
          max_interval: 0.250s     # 250ms max backoff
        per_try_timeout: 0.5s

Rate Limiting Implementations

Rate limiting protects your system from abuse and ensures fair resource allocation. The four main algorithms:

Fixed window: Count requests in a fixed time window (e.g., 100 requests per minute). Simple but has a boundary problem: if a user sends 100 requests at 12:00:59 and 100 at 12:01:01, they've made 200 requests in a 2-second window while the counter thinks they're within limits.

Sliding window log: Store a timestamp log of each request. Count only requests within the last N seconds. Precise but memory-intensive (stores one entry per request per user).

Token bucket: Each user has a bucket with a capacity of N tokens. Tokens are added at a rate of R per second. Each request consumes 1 token. Allows bursting up to bucket capacity. This is what AWS API Gateway uses.

Leaky bucket: Requests enter a queue; they're processed at a fixed rate regardless of arrival pattern. Smooths burst traffic but introduces queuing latency.

import time

# Redis Lua script for atomic token bucket rate limiting
RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local rate = tonumber(ARGV[1])      -- tokens per second
local capacity = tonumber(ARGV[2])  -- bucket capacity
local now = tonumber(ARGV[3])       -- current timestamp in ms
local requested = tonumber(ARGV[4]) -- tokens requested (usually 1)

local last_tokens = tonumber(redis.call('hget', key, 'tokens') or capacity)
local last_refill = tonumber(redis.call('hget', key, 'last_refill') or now)

-- Refill tokens based on time elapsed
local elapsed = math.max(0, now - last_refill)
local new_tokens = math.min(capacity, last_tokens + (elapsed * rate / 1000))

if new_tokens >= requested then
    -- Allow the request
    redis.call('hset', key, 'tokens', new_tokens - requested, 'last_refill', now)
    redis.call('expire', key, math.ceil(capacity / rate) + 1)
    return {1, math.floor(new_tokens - requested)}
else
    -- Reject — return wait time in ms
    local wait_ms = math.ceil((requested - new_tokens) * 1000 / rate)
    return {0, wait_ms}
end
"""

def check_rate_limit(user_id: str, rate: int = 100, capacity: int = 200) -> tuple:
    return redis_client.eval(
        RATE_LIMIT_SCRIPT, 1,
        f"ratelimit:{user_id}",
        rate, capacity,
        int(time.time() * 1000), 1
    )

Backpressure: When to Say No

Backpressure is the mechanism by which an overwhelmed service signals to its callers to slow down, rather than accepting requests it can't handle and failing them. Without backpressure, an overloaded service degrades gracefully into a spiral: it accepts more requests than it can handle, latency climbs, timeouts cascade upstream, callers retry (making things worse), and eventually the service becomes completely unresponsive.

Envoy's circuit breaker (shown above) implements backpressure at the proxy level. When max_pending_requests is exceeded, Envoy returns 503 immediately rather than queuing more work. This is the correct behavior — failing fast is better than failing slow. The caller sees the 503, applies its own circuit breaker or retry-with-backoff, and the overall system stabilizes.

Why Async Is Not Optional at Scale

When a user places an order on an e-commerce platform, the following things need to happen: inventory must be reserved, a payment must be charged, a confirmation email must be sent, fraud scoring must run, the warehouse must receive a pick list, and analytics must record the event. If you do all of this synchronously in the HTTP request handler, you are coupling the response time to your slowest dependency, creating a system that fails entirely if any one step is unavailable, and preventing independent scaling of each concern.

Message queues and event streams decouple producers from consumers. The order service publishes an OrderPlaced event and immediately responds to the user. Downstream services — inventory, payments, email, fraud, analytics — each subscribe to this event and process it independently, at their own pace, in their own infrastructure. If the email service is slow, it doesn't affect the user's checkout experience. If the fraud scoring service goes down, you can replay events when it comes back up.

Kafka Architecture

Apache Kafka is the dominant event streaming platform at scale, used by LinkedIn (where it was created), Uber, Netflix, Airbnb, and thousands of other companies. Understanding Kafka's architecture is essential because its design choices — log-based storage, consumer groups, partitions — are what make it capable of sustained throughput that simple message queues cannot match.

The key insight in Kafka's design is that it stores messages durably on disk in an ordered, immutable log. Consumers track their own position (offset) in the log. This means: multiple consumer groups can read the same topic independently, consumers can replay historical messages by seeking to an earlier offset, and the broker doesn't need to track which messages have been "consumed" for each consumer (that's each consumer group's responsibility). Traditional message queues like RabbitMQ delete messages after successful consumption — Kafka retains them for a configurable period (default 7 days).

from confluent_kafka import Producer, Consumer, KafkaError
import json

# Producer: publish order events
producer = Producer({
    'bootstrap.servers': 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    'acks': 'all',          # Wait for all ISR replicas to confirm
    'retries': 5,
    'enable.idempotence': True,  # Exactly-once delivery
    'compression.type': 'snappy',  # 4-5x compression on JSON
    'batch.size': 65536,   # 64KB batch size
    'linger.ms': 5,        # Wait 5ms to batch messages
})

def publish_order_event(order: dict):
    event = {
        'event_type': 'OrderPlaced',
        'order_id': order['id'],
        'user_id': order['user_id'],
        'total': order['total'],
        'timestamp': order['created_at'],
    }
    producer.produce(
        topic='order.events',
        key=str(order['user_id']),  # Key determines partition
        value=json.dumps(event).encode(),
        callback=delivery_report
    )
    producer.poll(0)  # Trigger callbacks, non-blocking

# Consumer: inventory service subscribes
consumer = Consumer({
    'bootstrap.servers': 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    'group.id': 'inventory-service-v2',  # Consumer group
    'auto.offset.reset': 'earliest',     # Read from start if new group
    'enable.auto.commit': False,          # Manual commit for at-least-once
})
consumer.subscribe(['order.events'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None: continue
    if msg.error(): continue
    event = json.loads(msg.value())
    try:
        reserve_inventory(event['order_id'])
        consumer.commit(asynchronous=False)  # Commit only after success
    except Exception as e:
        # Don't commit — message will be redelivered
        log_error(e)

Kafka vs RabbitMQ: The Honest Comparison

Event Sourcing and CQRS

Event sourcing stores the history of state changes as an ordered sequence of events, rather than the current state. Instead of storing "User John's balance is $500", you store "AccountOpened($1000), Withdrawal($200), Withdrawal($100), Deposit($300) → derived balance: $500". The current state is derived by replaying events. This gives you a complete audit log, the ability to time-travel to any historical state, and a natural foundation for event-driven architectures.

CQRS (Command Query Responsibility Segregation) separates the write model (commands that change state) from the read model (queries that read state). In a traditional CRUD app, the same data model serves both writes and reads. In CQRS, you might write order events to an event store and maintain multiple read models optimized for different query patterns — a denormalized "order summary" table optimized for the order list page, a separate "order analytics" table for reporting, and a search index for order search.

The Saga Pattern for Distributed Transactions

In a microservices world, you cannot use a single database transaction to span multiple services. The saga pattern implements distributed transactions as a sequence of local transactions, each publishing events that trigger the next step. If any step fails, compensating transactions undo the previous steps.

Consider an order placement flow: Create Order → Reserve Inventory → Charge Payment → Send Confirmation. If Payment fails, you need to release the inventory reservation (compensating transaction) and mark the order as failed. There are two saga implementations:

• Choreography-based: Each service listens for events and reacts. No central coordinator. Simple for small sagas but debugging event chains becomes complex at scale.
• Orchestration-based: A central saga orchestrator (often implemented with Temporal or Conductor) explicitly tells each service what to do next. Easier to reason about and debug, but the orchestrator becomes a bottleneck and single point of failure if not designed carefully.

Idempotency: The Key to Reliable Messaging

In at-least-once delivery systems (which includes most Kafka consumer setups), you will receive duplicate messages. A consumer might process a message, commit the result to its database, then crash before committing its offset to Kafka. On restart, it reprocesses the message. Your message handlers must be idempotent — processing the same message twice produces the same result as processing it once.

def handle_order_placed(event: dict):
    order_id = event['order_id']

    # Idempotency check: have we already processed this?
    # Use the event's unique ID, not just order_id
    event_id = event['event_id']  # Must be included in every event

    with db.transaction():
        # INSERT ... ON CONFLICT DO NOTHING
        inserted = db.execute("""
            INSERT INTO processed_events (event_id, processed_at)
            VALUES (%s, NOW())
            ON CONFLICT (event_id) DO NOTHING
            RETURNING event_id
        """, [event_id])

        if not inserted:
            # Already processed — this is a duplicate. Skip safely.
            return

        # Process the event — safe because we hold the idempotency record
        db.execute("""
            UPDATE inventory SET reserved = reserved + %s
            WHERE sku_id = %s
        """, [event['quantity'], event['sku_id']])

# LinkedIn processes 7 trillion messages per day through Kafka.
# Uber's Kafka cluster handles 1 trillion+ messages per day across 4000+ brokers.

REST Maturity Model

Leonard Richardson's REST Maturity Model describes four levels of REST API design. Most APIs that claim to be "RESTful" are actually at Level 2. True Level 3 (HATEOAS) is rarely implemented in practice but worth understanding to know what you're trading away.

• Level 0 (RPC over HTTP): Single endpoint, verbs in the URL. POST /getUser, POST /updateUser. This is what SOAP-based systems and many early APIs look like. The HTTP method is meaningless — it's always POST.
• Level 1 (Resources): Multiple endpoints, one per resource. POST /users/get, POST /users/update. Better URL structure but still misusing HTTP.
• Level 2 (HTTP Verbs): Resources + correct HTTP verbs. GET /users/{id}, PUT /users/{id}, DELETE /users/{id}. This is what most people mean by "REST" and what Stripe, GitHub, and Twilio implement. Use the right status codes too: 201 for created, 404 for not found, 409 for conflict.
• Level 3 (HATEOAS): Responses include links to related actions. A GET /orders/123 response includes links to /orders/123/cancel, /orders/123/ship, etc. Theoretically decouples clients from API structure but adds complexity that most teams don't find worth it.

# Creating an order
POST /v1/orders
Content-Type: application/json
Authorization: Bearer sk_live_xxxx
Idempotency-Key: 7f3b2a91-8c4d-4e6f-a9b0-123456789abc

{
  "user_id": "usr_abc123",
  "items": [{"sku": "SHOE-001", "qty": 2, "price_cents": 4999}],
  "shipping_address_id": "addr_xyz789"
}

HTTP/1.1 201 Created
Location: /v1/orders/ord_def456
X-Request-Id: req_ghi789
Retry-After: null

{
  "id": "ord_def456",
  "status": "pending",
  "created_at": "2026-06-29T10:30:00Z",
  "total_cents": 9998,
  "_links": {
    "self": "/v1/orders/ord_def456",
    "cancel": "/v1/orders/ord_def456/cancel"
  }
}

# Idempotency: same Idempotency-Key → same response, no duplicate order
# Use Stripe's approach: store key → response for 24h

GraphQL Federation

GraphQL solves a real problem with REST: client-specified queries eliminate over-fetching (getting fields you don't need) and under-fetching (needing multiple round trips to get all required data). But naive GraphQL implementation creates the "N+1 query problem" and can expose your database to expensive, hard-to-predict queries from clients.

Apollo Federation enables you to compose a single GraphQL schema from multiple downstream microservices, each owning a slice of the schema. The gateway receives a client query, decomposes it into sub-queries for the relevant services, fetches them in parallel, and assembles the final response. This is how large GraphQL deployments (Netflix, Airbnb, GitHub) manage schema ownership across teams.

gRPC and Protocol Buffers

gRPC is Google's open-source RPC framework built on HTTP/2 and Protocol Buffers. It's the default choice for service-to-service communication in performance-sensitive systems. The advantages over JSON/REST are substantial:

• Binary serialization: Protobuf messages are 3-10x smaller than equivalent JSON and 5-10x faster to serialize/deserialize. At 100K RPS, this difference in CPU cost is significant.
• Strongly typed contracts: The .proto schema is the contract. Client and server code is generated from it. Schema violations are caught at compile time, not runtime.
• HTTP/2 multiplexing: Multiple RPCs can share a single connection, eliminating head-of-line blocking and reducing connection overhead.
• Streaming: gRPC supports server-side streaming, client-side streaming, and bidirectional streaming — use cases that are awkward with REST.

// order_service.proto
syntax = "proto3";
package order;

service OrderService {
  rpc GetOrder (GetOrderRequest) returns (Order);
  rpc ListOrders (ListOrdersRequest) returns (stream Order);
  rpc CreateOrder (CreateOrderRequest) returns (Order);
}

message Order {
  string id = 1;
  string user_id = 2;
  int64 total_cents = 3;
  OrderStatus status = 4;
  int64 created_at_unix = 5;
  repeated LineItem items = 6;
}

enum OrderStatus {
  PENDING = 0;
  CONFIRMED = 1;
  SHIPPED = 2;
  DELIVERED = 3;
  CANCELLED = 4;
}

---
# Python server implementation
class OrderServicer(order_pb2_grpc.OrderServiceServicer):
    async def GetOrder(self, request, context):
        order = await db.get_order(request.order_id)
        if not order:
            context.abort(grpc.StatusCode.NOT_FOUND, "Order not found")
        return order_pb2.Order(**order)

    async def ListOrders(self, request, context):
        # Server-side streaming: yield orders as they're found
        async for order in db.stream_orders(request.user_id):
            yield order_pb2.Order(**order)

API Gateway Patterns

An API gateway is the single entry point for all client traffic into your system. It handles concerns that would otherwise need to be implemented in every microservice: authentication, rate limiting, SSL termination, request logging, response caching, and routing. Think of it as the "reverse proxy with business logic" layer.

Popular choices and their sweet spots: Kong (open-source, plugin ecosystem, good for complex routing), AWS API Gateway (zero ops, deep AWS integration, expensive at scale), Nginx (highest raw performance, great for simple routing), Envoy (best for service mesh integration, programmatic configuration), Traefik (cloud-native, automatic service discovery in Kubernetes).

Backend for Frontend (BFF) Pattern

The BFF pattern creates dedicated API gateway instances for each client type — mobile, web, third-party. Instead of one general-purpose API that serves all clients, you have a Mobile BFF that returns data pre-optimized for mobile (smaller payloads, aggregated responses), a Web BFF optimized for the web UI's data needs, and a Partner API with its own authentication and rate limiting regime.

SoundCloud pioneered this pattern. Their mobile clients needed different data shapes than their web clients, and a single general API was either over-fetching or requiring too many round trips. Netflix uses BFF extensively — their device gateway is customized per device type (iOS, Android, smart TV, game console) because each has different UI capabilities and network characteristics.

API Versioning Strategies

Stripe's date-based versioning is the most sophisticated approach. Every API change is gated behind a date. Customers pin their API calls to a specific date when they integrated, and that version's behavior is frozen for them forever. When Stripe changes behavior, new customers get the new behavior; old customers keep the old. This is operationally complex (you're maintaining many versions simultaneously) but produces the best developer experience because old integrations never break.

Cursor-Based Pagination

Never use OFFSET-based pagination for large datasets. SELECT * FROM orders ORDER BY created_at DESC LIMIT 20 OFFSET 10000 requires the database to scan 10,020 rows to return 20 — this gets slower as users page deeper. Cursor-based pagination uses an opaque cursor (usually a base64-encoded value of the last seen primary key or sort key) to efficiently resume where you left off:

# Cursor-based: O(1) regardless of depth
# First page:
SELECT id, created_at, total FROM orders
WHERE user_id = 123
ORDER BY created_at DESC, id DESC
LIMIT 21  -- fetch N+1 to know if there's a next page

# Subsequent pages (cursor = last item's (created_at, id)):
SELECT id, created_at, total FROM orders
WHERE user_id = 123
  AND (created_at, id) < ('2026-06-01T10:00:00Z', 98765)
ORDER BY created_at DESC, id DESC
LIMIT 21

# Python: encode/decode cursor
import base64, json

def encode_cursor(created_at: str, order_id: int) -> str:
    payload = json.dumps({'ts': created_at, 'id': order_id})
    return base64.urlsafe_b64encode(payload.encode()).decode()

def decode_cursor(cursor: str) -> dict:
    payload = base64.urlsafe_b64decode(cursor.encode())
    return json.loads(payload)

CAP Theorem: What It Actually Says

The CAP theorem (Brewer, 2000; formally proven by Gilbert and Lynch, 2002) states that a distributed data store can provide at most two of three guarantees simultaneously: Consistency (every read receives the most recent write or an error), Availability (every request receives a non-error response, though possibly stale), and Partition Tolerance (the system continues to operate when network partitions occur).

The critical — and often misunderstood — point is that partition tolerance is not optional. Networks partition. If you claim to build a distributed system without handling partitions, you're just not thinking about it when it happens. This means the real choice is: when a partition occurs, do you prioritize consistency or availability?

CP systems (Consistent + Partition Tolerant) choose to refuse requests rather than return potentially stale data during a partition. HBase, Zookeeper, and etcd are CP systems. During a partition that prevents a quorum, they become unavailable. AP systems (Available + Partition Tolerant) continue serving requests during a partition, potentially returning stale data. Cassandra, CouchDB, and DynamoDB (in eventual consistency mode) are AP systems.

PACELC: The More Useful Model

Daniel Abadi's PACELC theorem extends CAP to cover normal operation (not just partitions): "If there is a Partition, how do you trade off Availability vs Consistency? Else, even without partitions, how do you trade off Latency vs Consistency?" This is much more relevant to daily engineering decisions because partitions are relatively rare but the latency-consistency trade-off happens on every read.

Consistency Models Spectrum

Consistency is not binary. There's a rich spectrum from strongest (most correct but slowest) to weakest (fastest but most surprising behavior):

Linearizability (Strict Consistency): The strongest consistency model. Every operation appears to take effect instantaneously at some point between its invocation and completion, and all operations are globally ordered. If you write a value and immediately read it, you'll see your write. etcd, ZooKeeper, and Google Spanner provide linearizability. The cost: every read must go through the leader and verify no newer write exists.

Sequential Consistency: All operations appear to execute in some total order consistent with each process's program order. Weaker than linearizability — the global order doesn't need to respect real-time ordering. Rarely implemented in distributed databases; mostly relevant in CPU memory models.

Causal Consistency: Operations that are causally related are seen by all nodes in causal order. Concurrent (causally unrelated) operations may be seen in different orders by different nodes. MongoDB's causal consistency sessions implement this. This is often the "sweet spot" for social applications — you see your own writes immediately, and replies always appear after the message they reply to.

Eventual Consistency: If no new updates are made, all replicas will eventually converge to the same value. This says nothing about how long "eventually" takes or what you see in the interim. Cassandra and DynamoDB use this model. The practical implication: a user updates their profile and immediately visits their profile page — they might see the old version. This is usually acceptable if the convergence time is sub-second.

Read Your Own Writes (RYOW): A session-level guarantee that you'll always see your own writes, even in an eventually consistent system. Implemented by routing reads to the replica that handled your writes, or by passing a "read-after-write" token (the write's timestamp or log position) with subsequent reads.

Quorum Reads and Writes: Dynamo Style

Amazon Dynamo (the internal system, not DynamoDB) introduced a tunable consistency model using quorum reads and writes. With N replicas, a write is acknowledged after W replicas confirm it. A read queries R replicas and returns the most recent value. Consistency is guaranteed when R + W > N (read and write quorums overlap, ensuring at least one replica has the latest value).

# Cassandra: N=3 replicas
# Common configurations:

# High availability, eventual consistency:
# W=1 (write to 1 replica), R=1 (read from 1 replica)
# Fast but no consistency guarantees (R+W=2, not > 3)
consistency_level_write = ONE
consistency_level_read = ONE

# Balanced: strong consistency, some availability
# W=2, R=2  (R+W=4 > 3=N → consistent)
consistency_level_write = QUORUM   # ceil(N/2)+1 = 2
consistency_level_read = QUORUM

# Strongest: write to all, read from one
# W=3, R=1 (R+W=4 > 3=N → consistent, great read perf)
consistency_level_write = ALL
consistency_level_read = ONE

# DynamoDB strongly consistent reads:
# Always reads from the leader replica (not eventual quorum)
# 2x cost, ~double the latency vs eventually consistent
response = dynamodb.get_item(
    TableName='Orders',
    Key={'order_id': {'S': order_id}},
    ConsistentRead=True   # Linearizable read
)

CRDTs: Conflict-Free Replicated Data Types

CRDTs are data structures designed to be safely merged across replicas without coordination. Each CRDT operation is designed such that merging concurrent updates always produces a deterministic result, regardless of the order in which updates are applied. This makes CRDTs ideal for use in eventually-consistent systems where you want to avoid conflicts entirely rather than detect and resolve them.

Common CRDTs: G-Counter (grow-only counter — each node has its own counter, total is the sum; used by Riak for view counts), PN-Counter (increment and decrement — two G-counters, total = increment sum - decrement sum), LWW-Element-Set (Last-Write-Wins set — each element has a timestamp; the replica with the highest timestamp wins), OR-Set (Observed-Remove Set — tracks which node added an element with a unique tag; safe concurrent add/remove).

Google Spanner: Having Your Cake

Google Spanner achieves global, linearizable consistency across geographically distributed data while maintaining competitive performance. The trick is TrueTime — atomic clocks and GPS receivers in every Google datacenter, providing a globally synchronized time with a bounded uncertainty interval (typically ±7ms). Spanner uses this to implement external consistency: if transaction T1 commits before T2 starts (in real time), then T1's commit timestamp is lower than T2's. This globally orders all transactions without coordination across datacenters.

Spanner's read-write transactions use two-phase locking across shards. Cross-shard commit uses two-phase commit (2PC), which Spanner makes safe by using Paxos groups rather than a single coordinator. The result: serializable, globally-consistent transactions with ~5–10ms P50 latency within a region and ~100ms cross-continent. CockroachDB implements a similar architecture using HLC (Hybrid Logical Clocks) as an open-source TrueTime analog.

Resilience Is an Architecture Decision

Resilience is not something you add to a system after it's built — it's a property that emerges from design decisions made at every layer. A system is resilient when it continues to provide an acceptable level of service in the presence of faults and challenges to normal operation. Not perfect service — acceptable service. The goal is graceful degradation, not perfection.

Netflix coined the term "chaos engineering" precisely because resilience can't be assumed — it must be continuously validated. They've found that systems they believed were resilient would fail in unexpected ways when a single dependency was removed. The only way to know your system is resilient is to deliberately break it in production and observe what happens. Organizations that do this proactively have vastly fewer unplanned outages than those that discover failure modes during incidents.

Circuit Breaker Pattern

The circuit breaker pattern (named after electrical circuit breakers) prevents cascading failures by stopping calls to a failing dependency. When a dependency starts failing, you don't want your service to keep sending requests that will timeout — each timed-out request holds threads/connections and increases latency. The circuit breaker detects the failure rate and "opens" the circuit, returning errors immediately without attempting the call. After a configured time, it enters a "half-open" state, allowing a test request through. If that succeeds, it closes the circuit; if not, it stays open.

import time
from enum import Enum
from collections import deque

class State(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self,
                 failure_threshold: float = 0.5,  # 50% failure rate
                 window_seconds: int = 10,
                 open_duration: int = 30):
        self.failure_threshold = failure_threshold
        self.window = deque()
        self.window_seconds = window_seconds
        self.open_duration = open_duration
        self.state = State.CLOSED
        self.opened_at = None

    def _prune_window(self):
        cutoff = time.time() - self.window_seconds
        while self.window and self.window[0][0] < cutoff:
            self.window.popleft()

    def call(self, fn, *args, **kwargs):
        if self.state == State.OPEN:
            if time.time() - self.opened_at > self.open_duration:
                self.state = State.HALF_OPEN
            else:
                raise CircuitOpenError("Circuit is open")

        try:
            result = fn(*args, **kwargs)
            self.window.append((time.time(), True))
            if self.state == State.HALF_OPEN:
                self.state = State.CLOSED
            return result
        except Exception as e:
            self.window.append((time.time(), False))
            self._prune_window()
            failures = sum(1 for _, ok in self.window if not ok)
            rate = failures / len(self.window) if self.window else 0
            if rate >= self.failure_threshold:
                self.state = State.OPEN
                self.opened_at = time.time()
            raise

Bulkhead Isolation

The bulkhead pattern (named after the watertight compartments in ship hulls) isolates components into "pools" so that failure in one pool doesn't drain resources from another. In practice, this means: separate thread pools for different downstream services, separate connection pools for different databases, separate rate limit buckets per customer tier.

Netflix's Hystrix (now in maintenance mode, superseded by Resilience4j) implements bulkheads via thread pools. Each upstream service dependency gets its own thread pool with a configurable size. If the payment service starts timing out and exhausts its 10-thread pool, requests queue or fail fast — but the inventory service's 10-thread pool is unaffected. Without bulkheads, a slow dependency can consume all your application threads and bring down the entire service.

Retry with Exponential Backoff and Jitter

Retries are essential for handling transient failures, but naive retries can make an already-overloaded system catastrophically worse. If 10,000 clients all retry at the same time after a 503, the server gets a synchronized thundering herd. The solution is exponential backoff with jitter.

import time, random
from typing import Callable, TypeVar

T = TypeVar('T')

def retry_with_backoff(
    fn: Callable[[], T],
    max_attempts: int = 5,
    base_delay: float = 0.1,   # 100ms
    max_delay: float = 30.0,   # 30 seconds max
    jitter: bool = True
) -> T:
    for attempt in range(max_attempts):
        try:
            return fn()
        except TransientError as e:
            if attempt == max_attempts - 1:
                raise  # Exhausted retries

            # Exponential backoff: 100ms, 200ms, 400ms, 800ms, 1600ms
            delay = min(base_delay * (2 ** attempt), max_delay)

            if jitter:
                # "Full jitter" (AWS recommendation)
                # Spreads retries across [0, delay] uniformly
                # Eliminates synchronized retry bursts
                delay = random.uniform(0, delay)

            time.sleep(delay)

# What NOT to do: fixed sleep between retries
# time.sleep(1.0)  # Every client retries at t+1, t+2, t+3...
# This creates synchronized retry waves that hammer your service

Timeout Budgets

Every outbound call must have a timeout. "No timeout" is the single most common cause of thread starvation in production systems. A call to a slow service without a timeout will hold a thread forever, and 10,000 slow calls will exhaust your thread pool. But timeout values also need to be thoughtful — too aggressive (50ms) and you'll retry operations that would have succeeded in 100ms; too loose (30 seconds) and you hold threads unnecessarily long.

The more sophisticated approach is deadline propagation: each incoming request has a total deadline budget. As it calls downstream services, each call gets a portion of the remaining budget. If only 50ms remains and the downstream call typically takes 100ms, you can skip the call and return a degraded response rather than waiting and failing. Google's gRPC context propagation implements this. The idea: the user's patience is finite; make every system component aware of the remaining time budget and fail fast when it's exhausted.

Feature Flags for Graceful Degradation

Feature flags (or feature toggles) allow you to disable non-critical features at runtime without a deployment. During an incident where a downstream service is failing, you can immediately disable the feature that depends on it, serve a degraded but functional response, and re-enable it when the dependency recovers. This is vastly faster than deploying a code change during an incident.

Tools: LaunchDarkly (the enterprise standard, with targeting rules and gradual rollouts), Unleash (open-source), Flagsmith (open-source with cloud option). Implement flags as a simple boolean at first; the infrastructure cost is low and the operational benefit during incidents is enormous.

Chaos Engineering

Chaos engineering is the practice of deliberately injecting failures into a production system to discover weaknesses before they cause real incidents. Netflix's Chaos Monkey randomly terminates EC2 instances. Chaos Kong simulates the failure of an entire AWS availability zone. These run continuously in production — not occasionally in staging.

The chaos engineering process: define steady state (normal behavior metrics — error rate, latency P99, throughput), form a hypothesis ("if we kill one pod in the order-service deployment, the error rate should stay below 0.1% because we have 3 replicas"), run the experiment, verify the system behaves as hypothesized or fix the gap. Start small (single pod in staging) and expand to production when you have confidence.

Service Mesh for Resilience

A service mesh (Istio, Linkerd, Consul Connect) deploys a sidecar proxy (typically Envoy) alongside every service instance. This proxy intercepts all network traffic and can implement circuit breaking, retries, timeouts, mutual TLS, and distributed tracing without any changes to application code. The appeal is enormous: every service gets production-grade resilience primitives "for free," configured centrally by the platform team rather than reimplemented by every service team.

The reality is more nuanced. Istio adds ~15% latency overhead from the sidecar proxy hop. The control plane (Istiod) is a complex distributed system in its own right. Debugging networking issues in a service mesh is harder, not easier, because traffic flows through an extra layer. LinkedIn found that Istio at their scale added enough overhead to require dedicated platform teams just to manage it. Evaluate carefully based on your team's operational sophistication.

How to Approach System Design Interviews

System design interviews are not about having the "right answer" — there is no single correct architecture. They're about demonstrating that you can think through a complex problem systematically, make explicit trade-offs, and communicate your reasoning clearly. The best answers acknowledge limitations, ask clarifying questions, and adapt when the interviewer steers you in a different direction.

A reliable framework: (1) Clarify requirements — functional (what the system does) and non-functional (scale, latency, availability). (2) Estimate scale — QPS, storage, bandwidth. (3) High-level design — draw the major components. (4) Deep dive — focus on the hardest or most interesting components. (5) Address bottlenecks — what breaks first? how do you fix it? Spend about 5 minutes on 1-2, 10 minutes on 3, 20 minutes on 4-5 in a 45-minute interview.

Problem 1: Design Twitter/X Feed

Requirements clarification: 500M daily active users, each with ~200 followers on average, celebrities with up to 100M followers. Users post ~500M tweets/day. Feed shows tweets from followed users, reverse-chronological (or ranked). Read-heavy system — reads outnumber writes 100:1. Latency target: feed loads in <200ms P99.

Scale estimation: 500M tweets/day = ~5,800 tweets/second. Feed generation: 500M users × 5 feed loads/day = ~29,000 feed loads/second. Storage: avg tweet = 200 bytes, 500M/day × 200B = 100GB/day of tweet text.

The fundamental architectural choice: push vs pull for feed generation.

Pull model (fanout-on-read): When a user requests their feed, query all accounts they follow, get their recent tweets, merge and sort. Simple to implement, no write amplification. But at scale: a user following 2,000 accounts = 2,000 database queries per feed load. For 29,000 feed loads/second, that's 58 million queries/second to your tweet store. Prohibitive.

Push model (fanout-on-write): When a user tweets, immediately push the tweet ID into the home timeline cache of each follower. Feed load = single Redis ZSET lookup. Extremely fast reads. Problem: Lady Gaga has 100M followers. A single tweet triggers 100 million cache writes. At even 100K writes/second, that takes 16 minutes to complete — followers see the tweet hours later.

Twitter's actual solution (hybrid): Push for normal users (under ~10K followers). Pull for celebrities. When rendering a feed, merge the pre-computed timeline (from push) with tweets from any celebrities you follow (fetched on-demand). This caps write amplification at ~10K per tweet for 99.9% of tweets while giving celebrities real-time distribution.

Problem 2: Design Uber Ride Matching

Requirements: Match riders to the nearest available driver in <10 seconds. Handle 10M rides/day globally. Driver location updates every 4 seconds (GPS pings). City-level deployment, thousands of drivers online simultaneously per major city.

The core problem: Efficiently finding the nearest available drivers to a given location. This is a geospatial nearest-neighbor query at high update frequency. Standard SQL SELECT ... ORDER BY distance LIMIT 5 with an index is O(n) in the worst case and too slow at scale.

Geohashing: Encode latitude/longitude into a string where the prefix determines geographic proximity. A geohash of length 6 covers an ~1.2km × 0.6km cell. Drivers in the same cell share a geohash prefix. Finding nearby drivers = query drivers in the same geohash cell and adjacent cells (8 neighbors).

import geohash2 as geohash
import redis

r = redis.Redis()

def update_driver_location(driver_id: str, lat: float, lng: float):
    # Redis GEO commands handle geohashing internally
    r.geoadd("drivers:available", [lng, lat, driver_id])
    # Also store driver metadata for fast retrieval
    r.hset(f"driver:{driver_id}", mapping={
        "lat": lat, "lng": lng,
        "status": "available",
        "updated_at": int(time.time())
    })
    r.expire(f"driver:{driver_id}", 30)  # Expire if no update in 30s

def find_nearby_drivers(rider_lat: float, rider_lng: float,
                         radius_km: float = 2.0,
                         max_results: int = 10) -> list:
    # Redis GEOSEARCH: O(N+log(M)) where N is results, M is set size
    nearby = r.geosearch(
        "drivers:available",
        longitude=rider_lng, latitude=rider_lat,
        radius=radius_km, unit="km",
        sort="ASC",     # Nearest first
        count=max_results,
        withcoord=True,
        withdist=True
    )
    return [{
        "driver_id": item[0],
        "distance_km": item[1],
        "location": item[2]
    } for item in nearby]

# At 10,000 drivers online in NYC, Redis GEOSEARCH is ~1ms
# Uber partitions by city: each city has its own Redis instance

Problem 3: Design YouTube Video Processing

Requirements: 500 hours of video uploaded per minute. Each video must be transcoded into multiple resolutions (360p, 720p, 1080p, 4K) and formats (MP4/H.264, WebM/VP9, HLS for adaptive streaming). Processing must complete within minutes of upload. Store petabytes of video data.

High-level pipeline: Upload → object storage → message queue → transcoding workers → output object storage → CDN. The transcoding is CPU-intensive and perfectly parallelizable: different resolution outputs are independent jobs.

Key design decisions: Use spot/preemptible instances for transcoding (70–90% cost savings; checkpointing handles interruptions). FFmpeg is the standard transcoding tool — wrap it in a worker that pulls jobs from SQS. HLS (HTTP Live Streaming) output splits video into 2–10 second segments, enabling adaptive bitrate streaming where the client switches between quality levels based on available bandwidth. YouTube serves video from a custom CDN (Open Connect) deployed in ISP data centers. Netflix does the same — 15TB of content prefetched into an ISP's servers means video bytes never traverse the public internet for most users.

Problem 4: Design a Distributed Cache

Requirements: 1 million QPS. Sub-millisecond P99 read latency. 10TB of cached data. High availability (99.99% uptime). Support for TTL-based expiration. Key-value store with string values up to 1MB.

Architecture: Consistent hashing ring with 16 nodes (each handling ~625GB, well within 1TB NVMe capacity). Each node is a primary + 1 replica (synchronous replication for consistency). Nodes expose a binary protocol (like Memcached's binary protocol) for minimal parsing overhead. Connection pooling in the client library avoids TCP handshake overhead.

Eviction policy: LRU (Least Recently Used) eviction using an O(1) implementation: a hash map for O(1) lookup combined with a doubly-linked list where the head is MRU and tail is LRU. When the cache is full, evict the tail. Redis uses an approximated LRU that samples N random keys and evicts the least recently used among them — cheaper than maintaining a true LRU structure at scale.

from collections import OrderedDict
import time

class LRUCacheWithTTL:
    """O(1) get/put with TTL expiration"""
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = OrderedDict()  # key → (value, expiry_time)

    def get(self, key: str):
        if key not in self.cache:
            return None
        value, expiry = self.cache[key]
        if expiry and time.time() > expiry:
            del self.cache[key]
            return None
        self.cache.move_to_end(key)  # Mark as recently used
        return value

    def put(self, key: str, value: str, ttl_seconds: int = None):
        expiry = time.time() + ttl_seconds if ttl_seconds else None
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = (value, expiry)
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # Evict LRU (first item)

Problem 5: Design a URL Shortener

Requirements: 100M URLs shortened per day (1,157/second). 10B redirects per day (115,000 reads/second). URLs must not expire. Short codes should be ~7 characters. Custom aliases optional. Analytics: track click counts per URL.

Short code generation: The core challenge. Options: (1) MD5 hash of the long URL, take first 7 characters — collision risk, same URL might generate different codes. (2) Base62 encode a distributed counter (using a database sequence or Redis INCR) — unique by construction, no collisions, but the counter is a hotspot. (3) Pre-generate a pool of 7-character codes and distribute them — eliminates real-time generation latency.

The base62 counter approach is the most common in practice. With 62 characters (a-z, A-Z, 0-9) and 7 characters, you get 62^7 = 3.5 trillion unique codes — more than enough. The counter hotspot problem is solved by pre-allocating ranges: each application server claims a range of 1,000 IDs from the database and hands them out locally until exhausted, then claims another range. No per-request database write needed.

Caching strategy: 20% of URLs generate 80% of traffic (Pareto principle). Cache those hot URLs in Redis with an LRU policy. At 100 bytes per URL × 10M hot URLs = 1GB — easily fits in a single Redis instance. Add a CDN layer in front: Cloudflare can cache 301 redirects and serve them from the edge, eliminating your servers entirely for hot URLs. Use 302 (temporary redirect) if you want to maintain control and analytics; use 301 (permanent redirect) only if you want browsers to permanently cache it (you lose click analytics).

Common Scaling Interview Anti-Patterns

• Jumping straight to microservices without discussing scale requirements first
• Adding Kafka to every design "for scalability" without explaining what the consumer does
• Saying "use a distributed database" without specifying which one and why
• Ignoring the read/write ratio when choosing between SQL and NoSQL
• Not considering failure modes — what happens when the cache goes down? When the database master fails?
• Designing to serve 1B users from day one instead of showing how you'd evolve the design incrementally