System Design & Scalability

The first question in any scaling conversation is: when your system can no longer handle its load, what do you do? The two fundamental strategies are vertical scaling (scale up) and horizontal scaling (scale out).

Vertical scaling means upgrading to a bigger machine — more CPU cores, more RAM, faster disks. It is simple: your code does not change, your architecture does not change. The problem is that it has hard limits. You cannot buy a machine with a terabyte of RAM and 10,000 CPU cores for any practical price, and even if you could, a single machine is a single point of failure.

Horizontal scaling means adding more machines and distributing the load across them. It is how every large system in the world works. Google does not run on one very large computer; it runs on millions of commodity servers. The tradeoff is complexity: you now have to coordinate multiple machines, handle partial failures, and deal with the fact that data can exist on multiple nodes.

The key insight is that stateless services scale horizontally almost for free — you just add more instances behind a load balancer. Stateful services (especially databases) are much harder to scale horizontally, because you have to solve data consistency across nodes. This is why most architectures separate compute (stateless, easy to scale) from storage (stateful, hard to scale).

A useful mental model: vertical scaling is like replacing your car with a bigger truck. Horizontal scaling is like building a fleet of delivery vans. The fleet can handle much more total volume, but you need a dispatcher (load balancer), a routing system, and a way to handle when one van breaks down.

# Stateless service — trivially horizontally scalable
# Every request carries its own context (JWT token)
from flask import Flask, request, jsonify
import jwt

app = Flask(__name__)

@app.route('/api/data')
def get_data():
    # No server-side session state — any instance can handle this
    token = request.headers.get('Authorization')
    user = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
    return jsonify(fetch_from_db(user['id']))

# Contrast: stateful service — hard to scale out
class SessionServer:
    def __init__(self):
        self.sessions = {}  # Lives on ONE machine — breaks horizontal scaling
    
    def store_session(self, session_id, data):
        self.sessions[session_id] = data  # Sticky sessions or shared store needed

A load balancer sits in front of a pool of servers and distributes incoming requests among them. It is the traffic cop of horizontal scaling. Without a load balancer, you can have ten servers but every client would still need to know which server to talk to — which defeats the purpose.

Algorithms for distributing load:

1. Round Robin: requests go to servers in rotation (1, 2, 3, 1, 2, 3...). Simple, works well when all requests are similar in cost.
2. Weighted Round Robin: servers with more capacity get a proportionally larger share.
3. Least Connections: the next request goes to whichever server has the fewest active connections. Better when requests have variable durations (e.g., some API calls take 5ms, others take 500ms).
4. IP Hash / Consistent Hashing: the client's IP (or some request attribute) is hashed to pick a server. The same client always hits the same server — useful for caching session state locally, but can create hot spots.
5. Random with two choices: pick two servers at random, send to the one with fewer connections. Surprisingly close to "least connections" in practice with much less coordination overhead.

Layer 4 vs. Layer 7 balancing: A Layer 4 load balancer operates at the TCP level — it sees source/destination IP and port but not the HTTP content. Fast, but dumb. A Layer 7 load balancer understands HTTP: it can route /api/* to one pool and /static/* to another, inspect cookies, terminate SSL. Modern load balancers (NGINX, HAProxy, AWS ALB) are typically Layer 7.

Health checks: A load balancer periodically probes each backend (e.g., GET /health). If a server stops responding, the load balancer removes it from rotation automatically. This is the foundation of graceful degradation under partial failure.

Session stickiness is a common trap: if your application stores session state on the server (not in a shared cache or database), the load balancer must route each client to the same server every time. This works, but it undermines your ability to scale freely and makes deployments messy. The clean solution is to keep servers stateless.

# Consistent hashing — a smarter load balancing/sharding scheme
import hashlib
from bisect import bisect, insort

class ConsistentHashRing:
    """Maps keys to nodes such that adding/removing a node
       only remaps ~1/N of keys (vs. all keys in naive modulo hashing)."""
    
    def __init__(self, replicas=150):
        self.replicas = replicas  # virtual nodes per real node
        self.ring = []            # sorted list of (hash, node)
        self.node_map = {}
    
    def add_node(self, node):
        for i in range(self.replicas):
            h = self._hash(f"{node}:{i}")
            insort(self.ring, (h, node))
            self.node_map[h] = node
    
    def remove_node(self, node):
        self.ring = [(h, n) for h, n in self.ring if n != node]
    
    def get_node(self, key):
        if not self.ring:
            return None
        h = self._hash(key)
        # Find first virtual node with hash >= key's hash
        idx = bisect(self.ring, (h, ''))
        idx = idx % len(self.ring)  # wrap around
        return self.ring[idx][1]
    
    def _hash(self, s):
        return int(hashlib.md5(s.encode()).hexdigest(), 16)

Caching is the single most effective tool for improving system performance. The core idea is simple: if computing or fetching a value is expensive, store the result somewhere fast and reuse it. The complexity is in the details of what to cache, where, and when to invalidate.

Where caches live:

• Client-side (browser cache, CDN): reduces network round trips entirely. Headers like Cache-Control: max-age=3600 and ETags are the HTTP protocol for this.
• CDN (Content Delivery Network): a globally distributed cache of static assets and sometimes dynamic content. A request from a user in Tokyo hits a CDN edge server in Tokyo rather than your origin server in Virginia.
• Application-level cache (Redis, Memcached): a fast in-memory store shared across your application servers. Your app checks Redis before hitting the database.
• Database query cache: some databases (MySQL) have built-in query caches. Generally less useful than application-level caching because the database is often the bottleneck.

Cache-aside (lazy loading): The most common pattern. On a cache miss, the application fetches from the database and writes to the cache. On subsequent requests, the cache hit is served directly. Simple and resilient — if the cache is down, the app still works (just slower).

Write-through: Every write goes to the cache AND the database synchronously. Keeps cache consistent but adds write latency.

Write-behind (write-back): Writes go to the cache first; the database is updated asynchronously. Faster writes, but risk of data loss if the cache crashes before the write is persisted.

Cache eviction policies:

• LRU (Least Recently Used): evict the item that was least recently accessed. Works well for temporal locality.
• LFU (Least Frequently Used): evict the item accessed least often. Better for popularity-skewed workloads.
• TTL (Time-To-Live): items expire after a set time. Simple and predictable; necessary for data that changes.

Cache invalidation is notoriously hard. Phil Karlton's famous quote: "There are only two hard things in Computer Science: cache invalidation and naming things." The challenge is that when the underlying data changes, all cache entries that reflect that data must be updated or invalidated — and in a distributed system with many cache nodes, this is a distributed coordination problem.

Cache stampede (thundering herd): When a popular cache entry expires, many requests simultaneously miss the cache and all try to fetch from the database at once. Solutions include probabilistic early expiry, lock-based regeneration (only one process regenerates, others wait), and background refresh.

import redis
import json
from functools import wraps

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

def cache_aside(ttl=300):
    """Decorator implementing cache-aside pattern with Redis."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args):
            key = f"{fn.__name__}:{':'.join(str(a) for a in args)}"
            
            # 1. Check cache
            cached = r.get(key)
            if cached:
                return json.loads(cached)  # cache HIT
            
            # 2. Cache MISS: fetch from source
            result = fn(*args)
            
            # 3. Populate cache for next time
            r.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@cache_aside(ttl=60)
def get_user_profile(user_id: int) -> dict:
    # Expensive DB query — only runs on cache miss
    return db.query("SELECT * FROM users WHERE id = %s", user_id)

# LRU cache in pure Python (for illustration)
from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()  # insertion-ordered dict
        self.capacity = capacity
    
    def get(self, key):
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)  # mark as recently used
        return self.cache[key]
    
    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # evict LRU item

When a single database cannot handle your read or write throughput — no matter how many read replicas you add — you need to partition your data across multiple database instances. This is sharding (also called horizontal partitioning).

The fundamental idea: instead of one database holding all rows of the users table, you have 8 databases (shards), each holding 1/8 of the rows. A request for user #5,234,718 goes to the shard responsible for that user ID range.

Sharding strategies:

1. Range-based sharding: users 1–1M go to shard 1, 1M–2M to shard 2, etc. Simple, but creates hot spots — if most activity involves recent users (the highest IDs), all load concentrates on the last shard.

2. Hash-based sharding: shard = hash(user_id) % num_shards. Distributes load evenly, but range queries across shards become expensive (you have to query all shards and merge results).

3. Directory-based sharding: a lookup service (the shard map) records which shard holds which key. Flexible — you can move data between shards. But the lookup service itself is a bottleneck and single point of failure.

4. Consistent hashing (covered under Load Balancing): especially useful for sharding because adding a new shard only remaps ~1/N of the keys.

The costs of sharding:

• Cross-shard queries are expensive: a JOIN across users on different shards requires querying multiple shards and merging in the application. This is why you often denormalize data before sharding.
• Cross-shard transactions are very hard: you need distributed transactions (2PC, saga pattern) to maintain atomicity across shards.
• Rebalancing is painful: as data grows, you may need to reshard. This is a major operational event.
• No global uniqueness for auto-increment IDs: each shard auto-increments independently. Solutions include UUID, Snowflake IDs (Twitter), or a central ID service.

Sharding vs. Replication: These are often confused. Replication means copying the same data to multiple nodes (for fault tolerance and read scaling). Sharding means splitting different data across nodes (for write scaling). A well-designed system often uses both: 8 shards, each with 3 replicas.

-- Snowflake ID scheme (Twitter): 64-bit IDs that embed timestamp,
-- datacenter, machine, and sequence — globally unique without coordination
--
-- Bit layout:
--  0         [1]   timestamp_ms [41]   datacenter [5]   machine [5]   seq [12]
--  sign bit  epoch-relative ms  datacenter ID      worker ID        per-ms seq

-- In SQL, the shard for a user is: shard_id = user_id % NUM_SHARDS
-- Route query to the right DB connection:

# Python: route to the correct shard
NUM_SHARDS = 8

shard_connections = {
    i: create_db_connection(f"db-shard-{i}.internal") 
    for i in range(NUM_SHARDS)
}

def get_shard(user_id: int):
    return shard_connections[user_id % NUM_SHARDS]

def get_user(user_id: int) -> dict:
    conn = get_shard(user_id)
    return conn.execute(
        "SELECT * FROM users WHERE id = %s", 
        (user_id,)
    ).fetchone()

# Cross-shard query: expensive — must fan out to all shards
def count_users_in_country(country_code: str) -> int:
    total = 0
    for conn in shard_connections.values():
        row = conn.execute(
            "SELECT COUNT(*) FROM users WHERE country = %s",
            (country_code,)
        ).fetchone()
        total += row[0]
    return total  # O(num_shards) DB queries

The CAP theorem, proven by Eric Brewer and Seth Gilbert/Nancy Lynch, states that a distributed data store can provide at most two of the following three guarantees simultaneously:

• Consistency (C): Every read receives the most recent write or an error. All nodes see the same data at the same time.
• Availability (A): Every request receives a (non-error) response — though it might not be the most recent data.
• Partition Tolerance (P): The system continues to operate even when network partitions (message loss between nodes) occur.

The critical insight: in any real distributed system, network partitions will happen. You cannot build a network that never drops a packet. Therefore P is not really optional — you must tolerate partitions. The real choice is between C and A when a partition occurs:

• CP systems (e.g., HBase, Zookeeper, etcd): When a partition occurs, the system refuses to answer rather than returning potentially stale data. Banks, inventory systems, coordination services.
• AP systems (e.g., Cassandra, CouchDB, DynamoDB in default mode): When a partition occurs, the system keeps answering, but different nodes may return different (stale) values. The system is eventually consistent. Social media feeds, product catalogs, DNS.

PACELC extends CAP: Even when there is no partition, there is a tradeoff between Latency and Consistency. A strongly consistent system requires coordination (round trips between nodes) even for reads, adding latency. PACELC acknowledges this is often the more important tradeoff in practice — partitions are rare, but latency affects every request.

Consistency models (from strong to weak):

1. Linearizability (strongest): operations appear to execute instantaneously at some point between their start and end. The strongest guarantee; requires synchronous replication.
2. Sequential consistency: all operations appear to execute in some sequential order that respects each process's local order. Weaker than linearizability.
3. Causal consistency: causally related operations are seen in the correct order by all nodes. Unrelated operations may be seen in different orders.
4. Eventual consistency (weakest): if no new updates are made, all replicas will eventually converge to the same value. Amazon's Dynamo paper popularized this model.

Understanding which consistency model your data store provides is not academic — it determines whether you can safely read from a replica, whether you need distributed locks, and what invariants your application code can assume.

# Illustrating eventual consistency: two replicas diverge and converge
# This models what Cassandra or Dynamo does internally

import time, threading

class Replica:
    def __init__(self, name):
        self.name = name
        self.data = {}        # key -> (value, timestamp)
        self.peers = []
    
    def write(self, key, value):
        ts = time.time()
        self.data[key] = (value, ts)
        # In an AP system, we ACK immediately WITHOUT waiting for peers
        print(f"{self.name}: wrote {key}={value} at {ts:.3f}")
        # Async gossip to peers
        threading.Thread(target=self._gossip, args=(key,)).start()
    
    def read(self, key):
        # May return STALE data if gossip hasn't propagated yet
        return self.data.get(key, (None, 0))[0]
    
    def _gossip(self, key):
        time.sleep(0.1)  # simulate network delay
        for peer in self.peers:
            peer._merge(key, self.data[key])
    
    def _merge(self, key, incoming):
        # Last-write-wins conflict resolution using timestamp
        if key not in self.data or incoming[1] > self.data[key][1]:
            self.data[key] = incoming
            print(f"{self.name}: converged {key}={incoming[0]}")

r1, r2 = Replica('R1'), Replica('R2')
r1.peers = [r2]; r2.peers = [r1]

r1.write('user:42:name', 'Alice')
print(f"R2 reads immediately: {r2.read('user:42:name')}")  # None (stale!)
time.sleep(0.2)
print(f"R2 reads after gossip: {r2.read('user:42:name')}")  # Alice (converged)

A monolith is a single deployable unit containing all the application's functionality. A microservices architecture decomposes that monolith into small, independent services, each responsible for a single bounded context, communicating over well-defined APIs (typically HTTP/REST or gRPC over a network, or asynchronous message queues).

Why decompose? The monolith works fine until it doesn't. The problems appear as the team and codebase grow: a change to the recommendation engine requires redeploying the entire application. The team touching checkout and the team touching user profiles need to coordinate releases. A memory leak in the search service takes down the entire product. Microservices address these problems by making services independently deployable and isolating failures.

The real benefits:

• Independent deployability: the checkout team can ship 20 times a day without coordinating with the video streaming team.
• Technology heterogeneity: the ML recommendation service can be Python, the payment processor can be Java, the real-time notifications service can be Go.
• Fault isolation: a crashed recommendation service degrades the experience ("recommended for you" section is empty) but does not prevent users from checking out.
• Scaling granularity: you can run 100 instances of the search service and 3 instances of the rarely-used admin service.

The real costs:

• Network calls replace function calls: a function call takes nanoseconds; an HTTP call takes milliseconds and can fail. Every inter-service call adds latency and a new failure mode.
• Distributed transactions: if placing an order requires updating inventory (service A) and creating an order record (service B), you no longer have ACID transactions. You must use the saga pattern or two-phase commit.
• Operational complexity: 50 services means 50 deployment pipelines, 50 sets of logs to aggregate, 50 services to monitor. You need a service mesh (Istio, Linkerd), distributed tracing (Jaeger, Zipkin), and centralized logging.
• Data consistency: each service owns its own database (the "database per service" pattern). Keeping data consistent across these databases requires careful event-driven design.

When to use microservices: Start with a monolith. Extract services when you have clear bounded contexts, independent scaling needs, or independent team ownership. "Microservices-first" is almost always a mistake for early-stage products.

Service communication patterns:

• Synchronous (HTTP/gRPC): the caller waits for the response. Simple, but creates tight coupling — if service B is slow or down, service A is slow or down.
• Asynchronous (message queue: Kafka, RabbitMQ, SQS): the caller sends a message and moves on. Decouples services in time; enables retry and backpressure. The tradeoff is that the caller cannot immediately know the result.

# Saga pattern: coordinating a distributed transaction without 2PC
# Order service must update inventory AND charge payment — both must succeed or both rollback

# CHOREOGRAPHY-based saga: services react to events
# (no central coordinator — each service knows what to do next)

from dataclasses import dataclass
from enum import Enum

class Event(Enum):
    ORDER_CREATED      = "order.created"
    INVENTORY_RESERVED = "inventory.reserved"
    INVENTORY_FAILED   = "inventory.failed"
    PAYMENT_CHARGED    = "payment.charged"
    PAYMENT_FAILED     = "payment.failed"
    ORDER_COMPLETED    = "order.completed"
    ORDER_CANCELLED    = "order.cancelled"  # compensating transaction

# Order Service
def handle_event(event, payload):
    if event == Event.PAYMENT_FAILED:
        # Compensating transaction: undo the inventory reservation
        publish(Event.INVENTORY_RESERVED, {'action': 'RELEASE', **payload})
        publish(Event.ORDER_CANCELLED, payload)
    elif event == Event.INVENTORY_RESERVED and payload['action'] == 'CONFIRM':
        publish('payment.charge_requested', payload)

# Inventory Service
def on_order_created(payload):
    if reserve_stock(payload['item_id'], payload['qty']):
        publish(Event.INVENTORY_RESERVED, {**payload, 'action': 'CONFIRM'})
    else:
        publish(Event.INVENTORY_FAILED, payload)

# Each service does its work and emits an event.
# If any step fails, compensating events undo prior steps.
# No distributed lock held; no 2PC coordinator needed.

A message queue is a durable, ordered buffer between a producer and one or more consumers. The producer writes a message to the queue and moves on immediately — it does not wait for the consumer to process it. The consumer reads from the queue at its own pace.

Message queues solve several critical problems in distributed systems:

1. Traffic spikes and backpressure: If your API suddenly receives 100x its normal traffic, a message queue absorbs the spike. The consumers process messages at their sustainable rate; the queue grows temporarily. Without a queue, the spike either crashes your backend or requires you to provision 100x capacity for a temporary burst.

2. Decoupling: The service sending emails does not need to know about or depend on the email delivery service. It just publishes {"type": "send_email", "to": "..."} to a queue. If the email service is redeployed, the queue holds the messages until it comes back.

3. Reliability through retries: If a consumer crashes while processing a message, the queue (configured with acknowledgements) will redeliver the message to another consumer. This gives you at-least-once delivery semantics.

4. Fan-out / pub-sub: One producer can publish a message that is consumed by multiple independent consumer groups. A user.signed_up event might trigger the welcome email service, the analytics service, and the onboarding service all independently.

Key concepts:

• Acknowledgement (ACK): the consumer tells the queue "I successfully processed this message." Until the consumer ACKs, the message is not deleted — it will be redelivered if the consumer crashes.
• Dead Letter Queue (DLQ): messages that fail repeatedly (e.g., after 5 retries) are moved to a DLQ for manual inspection rather than blocking the main queue.
• Idempotency: because of redelivery, a consumer may process the same message more than once. Your consumer must be idempotent — processing the same message twice has the same effect as processing it once. This usually means keeping a processed-message-ID set, or using database upserts.
• Ordering: most queues guarantee ordered delivery within a partition (Kafka) but not globally across partitions. If strict ordering matters (e.g., financial transactions for one account), you must route related messages to the same partition.

Kafka vs. RabbitMQ: Kafka is a distributed log — it retains messages for a configurable time and allows consumers to replay from any offset. It is designed for high-throughput streaming and event sourcing. RabbitMQ is a traditional message broker with richer routing rules but less durability and lower throughput. Use Kafka when you need high throughput, replay, or multiple independent consumer groups; use RabbitMQ for task queues with complex routing.

# Producer: publish order events to Kafka
from kafka import KafkaProducer, KafkaConsumer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode()
)

def place_order(order: dict):
    # Returns immediately — does NOT wait for email/inventory services
    producer.send(
        topic='orders',
        key=str(order['user_id']).encode(),  # same user -> same partition -> ordered
        value=order
    )
    return {'status': 'accepted'}  # 202 Accepted

# Consumer: inventory service processes order events
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['kafka:9092'],
    group_id='inventory-service',  # each service has its own consumer group
    value_deserializer=lambda v: json.loads(v.decode()),
    auto_offset_reset='earliest',
    enable_auto_commit=False  # manual commit for reliability
)

for message in consumer:
    order = message.value
    try:
        if not already_processed(order['order_id']):  # idempotency check
            reserve_inventory(order['item_id'], order['qty'])
            mark_processed(order['order_id'])
        consumer.commit()  # ACK: tell Kafka we're done with this message
    except Exception as e:
        # Don't commit — Kafka will redeliver to this or another consumer
        log_error(e)

Replication means maintaining copies of the same data on multiple database nodes. The goals are fault tolerance (if the primary fails, a replica can take over), read scaling (route read queries to replicas), and geographic distribution (put replicas close to users in different regions).

Primary-replica (leader-follower) replication is the most common model. All writes go to the primary, which propagates changes to one or more replicas. Reads can go to any node.

Synchronous vs. asynchronous replication:

• Synchronous: the primary waits for at least one replica to acknowledge the write before returning success to the client. Strong consistency guarantee — a promoted replica has all committed writes. But write latency increases because a network round-trip to the replica is on the critical path.
• Asynchronous: the primary returns success as soon as it has written locally. Replicas catch up in the background. Lower write latency, but if the primary crashes before a replica catches up, committed data can be lost (replication lag).

Replication lag is a subtle source of bugs. If you write a value and then immediately read from a replica, you might get the old value. Applications must either read from the primary for consistency-sensitive queries, or use read-your-writes consistency (route reads for a user to the same node they just wrote to).

Multi-leader (multi-master) replication: Multiple nodes can accept writes, which they replicate to each other. Enables multi-region writes with low latency. The hard problem is write conflicts: if two leaders accept concurrent writes to the same key, you have a conflict that must be resolved. Strategies include last-write-wins (lossy), application-level resolution, or CRDTs (Conflict-free Replicated Data Types — data structures that merge automatically without conflicts).

Leaderless replication (Dynamo/Cassandra model): any node can accept a write, and the system uses quorums to achieve consistency. For N replicas, writes that succeed on W nodes and reads that check R nodes are consistent when W + R > N. With N=3, W=2, R=2: every read sees at least one node that has the latest write.

Write-ahead log (WAL) shipping: PostgreSQL and MySQL replicate by sending the binary WAL (the log of all changes) to replicas. Replicas replay the log to stay in sync. This is how logical replication, change data capture (CDC), and tools like Debezium work — they consume the WAL to propagate changes to other systems (search indexes, caches, data warehouses).

-- Quorum reads/writes in a Dynamo-style system (pseudocode/Python)
-- N=3 replicas, W=2 write quorum, R=2 read quorum
-- Guarantees: any read will see the latest write because W+R > N

N, W, R = 3, 2, 2  # W + R > N ensures overlap

def dynamo_write(key: str, value: str, replicas: list) -> bool:
    successes = 0
    for replica in replicas:
        try:
            replica.put(key, value, vector_clock=next_clock())
            successes += 1
        except NodeDown:
            pass
    # Only ACK client if quorum reached
    return successes >= W  # True = durable write

def dynamo_read(key: str, replicas: list):
    responses = []
    for replica in replicas:
        try:
            responses.append(replica.get(key))
        except NodeDown:
            pass
    if len(responses) < R:
        raise QuorumNotReached()
    # Resolve conflicts using vector clocks
    return reconcile(responses)  # picks highest vector clock value

# Vector clock for conflict detection (not last-write-wins)
from collections import defaultdict

class VectorClock:
    def __init__(self):
        self.clock = defaultdict(int)
    
    def increment(self, node_id: str):
        self.clock[node_id] += 1
    
    def dominates(self, other: 'VectorClock') -> bool:
        """True if self is causally after other."""
        return all(self.clock[k] >= other.clock[k] for k in other.clock) \
               and any(self.clock[k] > other.clock[k] for k in other.clock)

Rate limiting controls how often a client can call your API. It protects your system from abuse (DDoS, scraping), ensures fair usage across clients, and prevents cascading failures caused by runaway retry loops.

Why it's a systems design problem: A trivial implementation — a counter per user in memory — breaks immediately at scale. You have multiple API servers; a counter on one server knows nothing about requests hitting the others. You need a shared, fast store (Redis is the canonical choice).

Algorithms:

1. Fixed window counter: count requests in a fixed time window (e.g., 100 requests per minute). Simple, but has a boundary burst problem: a client can make 100 requests at 12:00:59 and 100 more at 12:01:01 — 200 requests in 2 seconds.

2. Sliding window log: maintain a sorted set of request timestamps for each client. Count entries in [now - window, now]. Accurate, but memory-intensive for high-request-rate clients.

3. Sliding window counter: approximation of sliding window using two fixed windows and a weighted interpolation. Memory-efficient and accurate enough for most uses.

4. Token bucket: a bucket holds up to N tokens. Tokens refill at rate r per second. Each request consumes one token. If the bucket is empty, the request is rejected. Allows short bursts (up to N tokens at once) while enforcing a long-run average rate r. Used by AWS, Stripe.

5. Leaky bucket: requests enter a queue (the bucket) at any rate and are processed at a fixed rate (the leak). Enforces a strict constant output rate. Good for smoothing traffic to a downstream system.

Distributed rate limiting with Redis: The standard approach uses Redis atomic operations (INCR + EXPIRE for fixed window, or sorted sets for sliding window). Redis's single-threaded execution model makes these operations race-condition free. For even higher throughput, you can use Redis's Lua scripts to execute multi-step logic atomically.

Response codes and headers: A rate-limited response should return HTTP 429 (Too Many Requests) with a Retry-After header indicating when the client can retry. Well-behaved clients use this to implement exponential backoff.

import redis
import time

r = redis.Redis()

def token_bucket_check(user_id: str, capacity: int = 100,
                       refill_rate: float = 10.0) -> bool:
    """
    Token bucket rate limiter using Redis.
    capacity: max burst size
    refill_rate: tokens added per second
    Returns True if request is allowed.
    """
    key = f"rate:{user_id}"
    now = time.time()
    
    # Lua script for atomicity — runs as a single Redis command
    lua_script = """
    local key = KEYS[1]
    local now = tonumber(ARGV[1])
    local capacity = tonumber(ARGV[2])
    local refill_rate = tonumber(ARGV[3])
    
    -- Get current state
    local data = redis.call('HMGET', key, 'tokens', 'last_refill')
    local tokens = tonumber(data[1]) or capacity
    local last_refill = tonumber(data[2]) or now
    
    -- Refill tokens based on elapsed time
    local elapsed = now - last_refill
    tokens = math.min(capacity, tokens + elapsed * refill_rate)
    
    -- Check and consume
    if tokens >= 1 then
        tokens = tokens - 1
        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
        redis.call('EXPIRE', key, 3600)
        return 1  -- allowed
    else
        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
        return 0  -- rejected
    end
    """
    result = r.eval(lua_script, 1, key, now, capacity, refill_rate)
    return bool(result)

The system design interview is a 45-60 minute conversation where you are asked to design a large-scale system (e.g., "Design Twitter", "Design a URL shortener", "Design Dropbox"). It is fundamentally a test of structured thinking under ambiguity, not a test of memorized architectures.

The framework (use this every time):

1. Clarify requirements (5 min) Never assume. Ask: Who are the users? What are the core features (scope carefully — you cannot design all of Twitter in 45 minutes)? What are the scale targets? (Daily active users? Requests per second? Data volume? Latency requirements?) What consistency/availability tradeoffs are acceptable?

2. Capacity estimation (5 min) Back-of-envelope math to understand the problem's scale. For Twitter:

• 300M daily active users, each reading ~100 tweets/day → 300M × 100 / 86,400 ≈ 350K reads/sec
• 50M tweets/day → ~600 writes/sec
• Each tweet is ~300 bytes → 50M × 300B = 15 GB/day of new tweet data These numbers tell you what kind of infrastructure you need and where the bottlenecks will be.

3. High-level design (15 min) Draw the big components: clients, load balancers, application servers, databases, caches, CDN, message queues. Do not go deep yet. Focus on the data flow for the core use cases.

4. Deep dive (20 min) Pick the most interesting/challenging components and go deep. The interviewer will guide this. Common areas: database schema, sharding strategy, cache design, how to handle the "hot key" problem (a celebrity's tweet creates a spike), feed generation (fan-out on write vs. fan-out on read).

5. Tradeoffs and bottlenecks (5 min) Proactively identify what will break first, what you have simplified, and what you would do differently with more time.

The "Twitter feed" tradeoff (canonical interview question):

• Fan-out on write (push model): when a user tweets, immediately push it to all followers' feed caches. Read is O(1) — just read your pre-computed feed. But write is O(followers) — a celebrity with 100M followers creates 100M write operations.
• Fan-out on read (pull model): when a user wants their feed, fetch the latest tweets from everyone they follow and merge/sort. Read is O(following) — expensive for active users but no write amplification.
• Hybrid (what Twitter actually does): fan-out on write for normal users, fan-out on read for celebrities ("verified" accounts above a threshold). Combine the feeds at read time.

What interviewers evaluate: structured thinking, knowledge of distributed systems primitives, ability to make and defend tradeoffs, awareness of failure modes, and communication clarity. There is no single right answer.

# Back-of-envelope estimation: always start here
# Example: URL shortener

# Assumptions (clarify with interviewer):
DAILY_NEW_URLS = 100_000_000   # 100M new short URLs per day
READ_WRITE_RATIO = 100         # 100 reads per write
URL_TTL_YEARS = 5

# QPS estimates
write_qps = DAILY_NEW_URLS / 86_400          # ~1,160 writes/sec
read_qps  = write_qps * READ_WRITE_RATIO     # ~116,000 reads/sec
# Peak is typically 2-3x average: ~350K reads/sec at peak

# Storage estimates  
bytes_per_url = 500   # original URL (200B) + short code (7B) + metadata
total_urls_5yr = DAILY_NEW_URLS * 365 * URL_TTL_YEARS   # 182.5B URLs
storage_tb = total_urls_5yr * bytes_per_url / 1e12       # ~91 TB

# Cache estimate: 80-20 rule — 20% of URLs get 80% of traffic
cache_entries = total_urls_5yr * 0.20                   # 36.5B entries
cache_gb = cache_entries * bytes_per_url / 1e9           # ~18 TB in cache
# Infeasible to cache all hot URLs in RAM — use tiered caching

print(f"Writes: {write_qps:.0f} QPS")
print(f"Reads: {read_qps:.0f} QPS (peak ~3x)")
print(f"Storage: {storage_tb:.0f} TB over {URL_TTL_YEARS} years")
# -> NoSQL (Cassandra/DynamoDB) for main store
# -> Redis cluster for hot URL cache
# -> Multiple read replicas or consistent hashing across nodes