Sharding, hashing & queues

1 · Sharding — when one DB isn't enough essential

Here's the wall you eventually hit. You added a read replica when the dashboards got slow (that's SD6), and it worked beautifully — reads flew. Then the writes started backing up. Every replica in the world is a copy of the same single primary, and every insert, update, and delete still funnels through that one machine. One night the write queue grows faster than the disk can flush it, and inserts start timing out. You can't add another replica to fix this, because more copies don't give you more places to write. That is the moment you need sharding.

The everyday analogy is a single overflowing filing cabinet. Buying a photocopier and making duplicate cabinets (replication) lets more people read the files at once — but there's still one cabinet everyone must file into, and it's full. Sharding is the other move: you buy three cabinets and agree a rule — "surnames A–H go here, I–P there, Q–Z in the third." Now three people can file at the same time, and you have three times the drawer space. The rule that decides which cabinet a file belongs in is the shard key, and picking a good one is the whole art: choose "surname" and a run of Smiths overloads one cabinet; choose something that spreads evenly and every cabinet fills at the same rate.

The mistake almost everyone makes first is confusing sharding with replication — treating "add more servers" as one undifferentiated idea. They solve opposite problems. Replication answers "too many readers" by copying the same data everywhere; sharding answers "too many writes, too much data" by splitting different data across machines. A replica holds all the rows; a shard holds only its slice. Get this backwards in an interview and the whole design falls apart — and in production you'll scale the wrong axis and wonder why the write bottleneck never budges. The second mistake, which we're about to watch happen, is reaching for the obvious shard rule (hash(key) % N) without realising it quietly bakes the number of servers into every key's address — so the day you add a machine, almost every key needs to move.

Replication copies data (scales reads); sharding splits it (scales writes + storage). Each shard holds a slice, chosen by a shard key. The naive approach: hash(key) % N.

modulo_shard.pydef shard_of(key, n):
    return hash(key) % n

keys = ["alice", "bob", "carol", "dave", "eve"]
before = {k: shard_of(k, 3) for k in keys}
print("with 3 shards:", before)

after = {k: shard_of(k, 4) for k in keys}       # add ONE shard...
moved = sum(before[k] != after[k] for k in keys)
print(f"added a shard -> {moved}/{len(keys)} keys moved")   # almost all! disaster

with 3 shards: {'alice': 0, 'bob': 2, 'carol': 2, 'dave': 0, 'eve': 1}
added a shard -> 4/5 keys moved

2 · Consistent hashing — shard without chaos intermediate

Picture the pain the modulo scheme just showed you, but at 3am and at real scale. You have a 100-node cache serving live traffic, one node dies, and your code does hash(key) % N to find things. The instant N drops from 100 to 99, the remainder changes for almost every key — so nearly every lookup now points at the wrong node, finds nothing, and falls through to the database. All of them. At once. The database, sized to handle the trickle of cache misses, gets the entire read load in one wave and falls over too. One dead cache node just took down your database. That cascade has a name — a cache stampede — and it's exactly what consistent hashing was invented to prevent.

Here's the mental shift that fixes it. Instead of a formula that depends on how many servers exist, imagine a giant clock face — a ring of numbers that wraps around from the top back to zero. You hash each server to a spot on that clock, and you hash each key to a spot too. To find where a key lives, you start at the key's position and walk clockwise until you bump into a server; that server owns the key. The beauty is what happens when a server vanishes: only the keys sitting in the arc just before it need a new home (they walk clockwise to the next server), and every other key on the ring is completely undisturbed. Add a server and it's the mirror image — it slots into one arc and steals only the keys in that slice, roughly 1/N of them, instead of shuffling the whole deck.

The subtle trap is assuming a plain ring automatically balances load. It doesn't — if you drop three servers onto random clock positions, one might land right after a huge empty arc and get swamped while another sits nearly idle. The fix is virtual nodes (vnodes): instead of placing each physical server at one point, you place it at many points around the ring (100 in the lab below), under names like s1:0, s1:1, and so on. Now each server's territory is chopped into 100 small arcs scattered evenly around the circle, so the law of averages gives everyone a fair share — and when a node dies, its keys get redistributed across all the survivors rather than dumped entirely on its single clockwise neighbour. This is not academic: it's the literal backbone of Amazon DynamoDB, Apache Cassandra, and most distributed caches. Watch the numbers in the lab — adding a server moves ~1/4 of keys, not ~all of them.

Consistent hashing places servers and keys on a ring; a key belongs to the next server clockwise. Add/remove a server and only the keys near it move — the backbone of distributed caches and databases (Cassandra, DynamoDB).

consistent_hash.pyimport hashlib, bisect

class HashRing:
    def __init__(self, nodes=(), vnodes=100):
        self.vnodes = vnodes; self._ring = {}; self._sorted = []
        for n in nodes: self.add(n)
    def _h(self, key): return int(hashlib.md5(key.encode()).hexdigest(), 16)
    def add(self, node):
        for i in range(self.vnodes):
            h = self._h(f"{node}:{i}"); self._ring[h] = node; bisect.insort(self._sorted, h)
    def remove(self, node):
        for i in range(self.vnodes):
            h = self._h(f"{node}:{i}"); del self._ring[h]; self._sorted.remove(h)
    def get(self, key):
        h = self._h(key); i = bisect.bisect(self._sorted, h) % len(self._sorted)
        return self._ring[self._sorted[i]]

ring = HashRing(["s1", "s2", "s3"])
keys = [f"user{i}" for i in range(1000)]
before = {k: ring.get(k) for k in keys}
ring.add("s4")                                   # scale out
moved = sum(before[k] != ring.get(k) for k in keys)
print(f"added s4 -> {moved}/1000 keys moved (~1/4, not all)")

added s4 -> 251/1000 keys moved (~1/4, not all)

3 · Message queues — decouple & absorb bursts advanced

Think about the last time you uploaded a photo and the app said "we'll email you when it's ready." It didn't freeze your screen for thirty seconds while it resized the image, ran it through a model, and generated thumbnails. It took your file, said "got it," and let you get on with your life — the actual heavy work happened later, somewhere else. That "got it, we'll handle it" moment is a message queue doing its job. The alternative — the app calling the image service directly and making you wait for it — is how you build something that feels slow when it works and collapses entirely when the image service has a bad minute.

The analogy that makes it click is a restaurant. Waiters (producers) don't cook — they take orders, clip the tickets to a rail in the kitchen window, and immediately go serve the next table. Cooks (consumers) pull tickets off the rail at their own pace. That humble ticket rail is a queue, and it buys the restaurant three things it couldn't get if waiters had to stand at the stove until each dish was done. First, decoupling: waiters and cooks work independently — neither waits on the other. Second, absorbing bursts: when a coach party arrives and forty orders hit at once, the tickets simply stack up on the rail instead of the dining room grinding to a halt. Third, retries and safety: if a cook drops a plate, the ticket is still there — the order isn't lost, it just gets remade.

The mistake to unlearn is thinking a queue is "just a list you push to." The value isn't the buffer itself — it's the durability and the delivery guarantees wrapped around it. A real queue (Kafka, RabbitMQ, SQS) persists messages to disk so a crash doesn't vaporise the backlog, tracks which messages have been acknowledged so a consumer dying mid-job doesn't silently drop work, and re-delivers anything that wasn't confirmed done. That last property is why queues give you at-least-once delivery — and its flip side, the trap juniors fall into: because a message can be delivered more than once (a consumer processed it but crashed before acknowledging), your consumers must be idempotent — handling the same message twice must be safe. "Charge the customer" as a queue handler, written naively, will double-bill someone. Build the retry in from the start, and route the message that fails every retry to a dead-letter queue so one poison message can't jam the line forever — you'll build exactly that in the practice ladder. The lab below is this whole shape in ~20 lines of standard-library Python.

Instead of A calling B directly (and failing if B is down/slow), A puts a message on a queue and B consumes it when ready. This decouples producer from consumer, absorbs spikes, and enables retries.

queue_demo.pyimport queue, threading

work = queue.Queue()                 # stands in for Kafka/RabbitMQ
processed, failed = [], []

def producer():
    for i in range(6): work.put(f"job-{i}")
    work.put(None)                    # sentinel = done

def consumer():
    while True:
        job = work.get()
        if job is None: break
        try:
            if job == "job-3": raise RuntimeError("transient")   # simulate a failure
            processed.append(job)
        except Exception:
            work.put(job.replace("job", "retry"))                # re-queue for retry
            failed.append(job)
        finally:
            work.task_done()

t = threading.Thread(target=consumer); t.start()
producer(); t.join()
print("processed:", processed)
print("retried:", failed)

processed: ['job-0', 'job-1', 'job-2', 'job-4', 'job-5', 'retry-3']
retried: ['job-3']

4 · Expert — SQL vs NoSQL, decided by data expert

Somewhere right now a team is rewriting a perfectly good PostgreSQL app onto a NoSQL database because a conference talk said SQL "doesn't scale." Six months later they're hand-rolling joins in application code, discovering their "eventually consistent" store let a customer's balance go negative, and quietly missing the transactions they threw away. This is the single most expensive mistake in this chapter — not a bug, but a whole architecture chosen for the wrong reason. SQL versus NoSQL is not a fight where one side wins; it's a fit question, and the answer is written in your data, not in the hype cycle.

The honest way to see it: these are two different tools that gave up different things to be good at different jobs. A SQL database is like a meticulous accountant — it enforces relationships between tables, lets you ask complex questions across them with joins, and wraps changes in ACID transactions so "move $100 from A to B" either fully happens or fully doesn't, never half. That rigour is priceless for money, orders, and anything where a wrong answer is a real-world disaster. A NoSQL store is more like a vast, simple set of labelled lockers — hand it a key, it hands back a blob, blindingly fast, and you can add lockers (shards) forever. It got that horizontal scale and schema flexibility by giving up the accountant's guarantees: cross-locker joins and multi-key transactions largely aren't on the menu. Neither is "better." They're bets on what your workload needs most.

The mistake, spelled out so you never make it: choosing by fashion or by a single scary word ("scale") instead of by your actual access pattern — the concrete way your app reads and writes. Ask what you truly need: Do rows relate to each other and get queried together (a payments ledger, an inventory system)? Lean SQL. Are you firehosing simple, independent records at enormous volume and only ever looking them up by one key (event logs, session stores, feature flags)? That's the NoSQL sweet spot. And know the default: start with SQL until a specific pattern forces you off it — it scales far further than its reputation suggests (read replicas from SD6, plus the sharding from §1), and losing joins and transactions is a cost you pay forever. Most real systems land on "SQL for the core data that must be correct, plus a key-value cache for the hot lookups." The lab makes the NoSQL shape concrete: a tiny key-value store sharded across nodes by the very ring you built in §2 — fast, horizontally scalable, and pointedly unable to do a cross-shard join, which is precisely the trade you're agreeing to.

Not a rivalry — a fit question decided by access pattern. SQL gives relations, joins, and ACID; NoSQL trades some of those for scale and flexible schemas. Let's make the trade-off concrete with a tiny key-value store (the NoSQL shape) sharded by the ring above.

sharded_kv.pyimport hashlib, bisect

class HashRing:
    def __init__(self, nodes=(), vnodes=50):
        self.vnodes=vnodes; self._ring={}; self._sorted=[]
        for n in nodes: self.add(n)
    def _h(self,k): return int(hashlib.md5(k.encode()).hexdigest(),16)
    def add(self,node):
        for i in range(self.vnodes):
            h=self._h(f"{node}:{i}"); self._ring[h]=node; bisect.insort(self._sorted,h)
    def get(self,key):
        h=self._h(key); i=bisect.bisect(self._sorted,h)%len(self._sorted)
        return self._ring[self._sorted[i]]

class ShardedKV:
    """Fast key-lookup store, horizontally sharded — the NoSQL sweet spot.
    No joins/transactions across shards; that's the trade for scale."""
    def __init__(self, nodes):
        self.ring = HashRing(nodes)
        self.shards = {n: {} for n in nodes}
    def put(self, key, value): self.shards[self.ring.get(key)][key] = value
    def get(self, key): return self.shards[self.ring.get(key)].get(key)
    def where(self, key): return self.ring.get(key)

kv = ShardedKV(["db0", "db1", "db2"])
for i in range(6): kv.put(f"user:{i}", {"name": f"u{i}"})
print("user:3 ->", kv.get("user:3"), "on", kv.where("user:3"))
print("distribution:", {n: len(s) for n, s in kv.shards.items()})

user:3 -> {'name': 'u3'} on db1
distribution: {'db0': 2, 'db1': 2, 'db2': 2}

def shard_of(key, n_shards):
    return hash(key) % n_shards

from collections import Counter
counts = Counter(shard_of(f"user{i}", 4) for i in range(10000))
print(dict(sorted(counts.items())))
# roughly even, e.g. {0: 2504, 1: 2470, 2: 2512, 3: 2514}

keys = [f"user{i}" for i in range(10000)]
def where(key, n): return hash(key) % n

moved = sum(1 for k in keys if where(k, 4) != where(k, 5))
print(f"{moved}/{len(keys)} keys move = {moved/len(keys):.0%}")
# ~8000/10000 = ~80% move  <-- almost everything reshuffles

import hashlib, bisect

class HashRing:
    def __init__(self, nodes, vnodes=100):
        self.vnodes = vnodes
        self.ring = {}      # hash -> node
        self.sorted = []
        for n in nodes: self.add(n)
    def _h(self, s):
        return int(hashlib.md5(s.encode()).hexdigest(), 16)
    def add(self, node):
        for v in range(self.vnodes):
            h = self._h(f"{node}#{v}")
            self.ring[h] = node
            bisect.insort(self.sorted, h)
    def get(self, key):
        h = self._h(key)
        i = bisect.bisect(self.sorted, h) % len(self.sorted)
        return self.ring[self.sorted[i]]

keys = [f"user{i}" for i in range(10000)]
r1 = HashRing(["a", "b", "c", "d"])
before = {k: r1.get(k) for k in keys}
r1.add("e")
moved = sum(1 for k in keys if before[k] != r1.get(k))
print(f"{moved}/{len(keys)} keys move = {moved/len(keys):.0%}")
# ~1900/10000 = ~19% (near the ideal 1/5), vs ~80% for modulo

from collections import deque

class Queue:
    def __init__(self, max_retries=3):
        self.q = deque(); self.dlq = []; self.max_retries = max_retries
    def publish(self, msg): self.q.append({"msg": msg, "attempts": 0})
    def consume(self, handler):
        while self.q:
            item = self.q.popleft()
            item["attempts"] += 1
            try:
                handler(item["msg"])
            except Exception:
                if item["attempts"] < self.max_retries:
                    self.q.append(item)          # retry later
                else:
                    self.dlq.append(item)        # give up -> DLQ

q = Queue(max_retries=2)
for m in ["ok1", "BAD", "ok2"]: q.publish(m)
def handler(m):
    if m == "BAD": raise ValueError("boom")
q.consume(handler)
print("dlq:", [i["msg"] for i in q.dlq], "attempts:", q.dlq[0]["attempts"])
# dlq: ['BAD'] attempts: 2

import math

def consumers_needed(arrival_rps, service_time_s, target_util=0.7):
    # each consumer handles 1/service_time msgs/s; keep utilization under target
    per_consumer = 1 / service_time_s
    return math.ceil(arrival_rps / (per_consumer * target_util))

def drain_time_s(backlog, arrival_rps, consumers, service_time_s):
    capacity = consumers / service_time_s          # msgs/s the pool can process
    net = capacity - arrival_rps                    # drain rate
    return float("inf") if net <= 0 else backlog / net

n = consumers_needed(arrival_rps=500, service_time_s=0.05)
print("consumers:", n)                              # 36
print("drain 10k backlog:",
      round(drain_time_s(10000, 500, n, 0.05), 1), "s")   # ~44.9 s

def choose_store(needs):
    reasons = []
    score_sql = 0
    if needs.get("multi_row_transactions"): score_sql += 2; reasons.append("ACID txns -> SQL")
    if needs.get("complex_joins"):          score_sql += 2; reasons.append("relational joins -> SQL")
    if needs.get("flexible_schema"):        score_sql -= 1; reasons.append("schema churn -> NoSQL")
    if needs.get("massive_write_scale"):    score_sql -= 2; reasons.append("write scale -> NoSQL")
    if needs.get("simple_key_access"):      score_sql -= 1; reasons.append("key lookups -> NoSQL")
    verdict = "SQL" if score_sql >= 0 else "NoSQL"
    return verdict, reasons

print(choose_store({"multi_row_transactions": True, "complex_joins": True}))
# ('SQL', ['ACID txns -> SQL', 'relational joins -> SQL'])
print(choose_store({"massive_write_scale": True, "simple_key_access": True,
                    "flexible_schema": True}))
# ('NoSQL', ['schema churn -> NoSQL', 'write scale -> NoSQL', 'key lookups -> NoSQL'])

Knowledge check check yourself