8. Project: a sharding and replica plan for Enlace

First, the calculator that produces the plan, including the remapping demonstration from lessons 6-7:

# enlace_scaling_plan.py — the scaling plan for Enlace's database.
import bisect
import hashlib
import math
from collections import Counter

# ---------- Enlace's anchor numbers ----------
TOTAL_TB = 6.0              # ~6 TB at 5 years (module 2)
qps_read = 3_858           # ~4,000 reads/s
qps_write = 39             # ~40 writes/s
hit_ratio = 0.90           # module 4 cache
tb_per_shard = 1.0         # DECISION: data budget per shard (comfortable)
reads_per_replica = 2_000  # DECISION: reads/s a replica handles

# ---------- Step 1: how many shards for the 6 TB ----------
shards_min = math.ceil(TOTAL_TB / tb_per_shard)
SHARDS = 8                                    # DECISION: 8 (power of 2, margin)
print(f"Data: {TOTAL_TB} TB / {tb_per_shard} TB per shard -> minimum {shards_min} shards")
print(f"We choose SHARDS = {SHARDS}  (~{TOTAL_TB/SHARDS*1000:.0f} GB per shard, with margin)\n")

# ---------- Step 2: load per shard ----------
reads_to_db = qps_read * (1 - hit_ratio)      # the cache absorbs 90%
reads_per_shard = reads_to_db / SHARDS
writes_per_shard = qps_write / SHARDS
print(f"Reads to the DB (after 90% cache): {reads_to_db:,.0f}/s -> {reads_per_shard:,.1f}/s per shard")
print(f"Writes: {qps_write}/s -> {writes_per_shard:.1f}/s per shard primary\n")

# ---------- Step 3: replicas per shard ----------
replicas_needed = max(1, math.ceil(reads_per_shard / reads_per_replica))
replicas_prod = max(2, replicas_needed + 1)   # +1 for redundancy (hold up N-1)
total_nodes = SHARDS * (1 + replicas_prod)    # 1 primary + R replicas per shard
print(f"Replicas per shard: minimum {replicas_needed}, in production {replicas_prod} (redundancy)")
print(f"Total nodes: {SHARDS} x (1 primary + {replicas_prod} replicas) = {total_nodes}\n")

# ---------- Step 4: growing without a storm (lessons 6-7) ----------
def make_keys(k):
    alphabet = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
    out = []
    for i in range(k):
        num, s = 100_000_000 + i, ""
        while num:
            s, num = alphabet[num % 62] + s, num // 62
        out.append(s.rjust(7, "0"))
    return out

def ring_point(text):
    return int(hashlib.md5(text.encode()).hexdigest(), 16) % (2 ** 32)

def node_modn(code, n):
    return int(hashlib.md5(code.encode()).hexdigest(), 16) % n

def build_ring(nodes, vnodes=100):
    ring = {}
    for node in nodes:
        for i in range(vnodes):
            ring[ring_point(f"{node}#{i}")] = node
    return ring

def node_ring(code, ring, pts):
    p = ring_point(code)
    idx = bisect.bisect(pts, p)
    if idx == len(pts):
        idx = 0
    return ring[pts[idx]]

K, N = 1_000_000, SHARDS
keys = make_keys(K)

# mod-N: remapping when going from N to N+1
moved_modn = sum(node_modn(k, N) != node_modn(k, N + 1) for k in keys)

# consistent hashing: same experiment over the ring
nodes = [f"db-{i}" for i in range(N)]
ring = build_ring(nodes)
pts = sorted(ring)
before = {k: node_ring(k, ring, pts) for k in keys}
for i in range(100):                           # we add shard 9 (db-8)
    ring[ring_point(f"db-{N}#{i}")] = f"db-{N}"
pts = sorted(ring)
moved_ring = sum(node_ring(k, ring, pts) != before[k] for k in keys)

print(f"Adding shard {N+1} (from {N} to {N+1}), {K:,} keys:")
print(f"  mod-N            remaps {moved_modn:>9,}  ({100*moved_modn/K:5.1f}%)  ~{moved_modn/K*TOTAL_TB:.2f} TB moved")
print(f"  consistent hash  remaps {moved_ring:>9,}  ({100*moved_ring/K:5.1f}%)  ~{moved_ring/K*TOTAL_TB:.2f} TB moved")
print(f"  improvement: {moved_modn/moved_ring:.1f}x less data moved\n")

# even spread among the 9 shards after growing (vnodes do their job)
dist = Counter(node_ring(k, ring, pts) for k in keys)
print("Spread among the 9 shards (vnodes=100):", [dist[f"db-{i}"] for i in range(N + 1)])

What to expect. Running python enlace_scaling_plan.py gives, exactly:

Data: 6.0 TB / 1.0 TB per shard -> minimum 6 shards
We choose SHARDS = 8  (~750 GB per shard, with margin)

Reads to the DB (after 90% cache): 386/s -> 48.2/s per shard
Writes: 39/s -> 4.9/s per shard primary

Replicas per shard: minimum 1, in production 2 (redundancy)
Total nodes: 8 x (1 primary + 2 replicas) = 24

Adding shard 9 (from 8 to 9), 1,000,000 keys:
  mod-N            remaps   888,920  ( 88.9%)  ~5.33 TB moved
  consistent hash  remaps   130,623  ( 13.1%)  ~0.78 TB moved
  improvement: 6.8x less data moved

Spread among the 9 shards (vnodes=100): [107501, 95552, 117241, 110026, 105966, 111195, 117227, 104669, 130623]

Read the output against the acceptance criteria:

• 8 shards, ~750 GB each. With a budget of 1 TB per shard, the 6 TB call for a minimum of 6; we choose 8 (a power of 2, comfortable to reason about) to leave each shard at ~750 GB, with slack to grow before the next reshard.
• Tiny load per node: after the cache (90%) only 386 reads/s reach the database, which spread among 8 shards are ~48/s per shard; the writes are ~5/s per primary. The cache (M4) and the sharding together leave each node almost idle —that slack is what absorbs the spikes and the mass-miss day—.
• 2 replicas per shard. The arithmetic floor is 1 replica per shard (48/s fits with plenty to spare in the 2,000/s budget), but we deploy 2 to hold up with one crash (N−1 rule, lesson 3): if a replica goes down, the other absorbs its load without sweating. Total: 8 × (1 primary + 2 replicas) = 24 nodes.
• Growing without a storm, measured: adding shard 9 with mod-N remaps 888,920 keys (88.9%) ≈ 5.33 TB; with consistent hashing, 130,623 (13.1%) ≈ 0.78 TB —6.8× less data moved—. That's the difference between an hours-long migration that cools the whole cache and a routine operation that only touches the new node's arc.
• Even spread after growing: with vnodes=100, the 9 shards end up between ~95k and ~130k keys —reasonably even, without the 10× imbalance the ring without vnodes would have (lesson 7)—. The new shard db-8 has 130,623, which is exactly the number of remapped keys: when adding a node, the only keys that move are the ones that node captures from its arc (no one moves between old nodes).

The topology diagram

flowchart TD
    client["Clients"]
    cache["cache (Redis)<br/>hit ratio 90%<br/>absorbs 3,472/s"]
    router["Routing layer<br/>shard = ring(short_code)"]

    client -->|"resolve / shorten"| cache
    cache -->|"miss: 386 reads/s + 39 writes/s"| router

    subgraph shard0["Shard 0 (~750 GB)"]
        p0[("PRIMARY<br/>writes")]
        r0a[("replica")]
        r0b[("replica")]
        p0 -->|replication log| r0a
        p0 -->|replication log| r0b
    end

    subgraph shardN["Shard 7 (~750 GB)"]
        pN[("PRIMARY<br/>writes")]
        rNa[("replica")]
        rNb[("replica")]
        pN -->|replication log| rNa
        pN -->|replication log| rNb
    end

    router -->|"write -> primary"| p0
    router -->|"read -> replica"| r0a
    router -->|"write -> primary"| pN
    router -->|"read -> replica"| rNa

    note["... shards 1..6 the same<br/>(8 shards x 3 nodes = 24)"]

The capacity table

The list of tradeoffs

• Why 8 shards and not 6 (the minimum). 6 shards would fit (1 TB each), but they leave little margin before the next reshard. 8 leaves ~750 GB per shard and is a power of 2. Cost: more nodes to operate. What would change it: if the growth were faster than foreseen, I'd start with more shards; with consistent hashing, adding the 9th is no longer traumatic, so starting tight is less risky than before.
• Why 2 replicas per shard and not 1. The calculation says 1, but 1 doesn't tolerate a crash. Cost: it doubles the read nodes (from 8 to 16 replicas). What would change it: if the cache were even more effective or the SLA laxer, 1 replica + failover to the primary might be enough; for an always-available service, the redundancy is worth its cost.
• Why consistent hashing and not mod-N. mod-N spreads just as well with fixed N, but when growing it remaps 88.9% (5.33 TB, total cold cache). Consistent hashing moves ~13.1% (0.78 TB). Cost: a bit more complexity (the ring, the vnodes, a positions table). What would change it: nothing realistic; the ring is the standard choice. Only a system that never changes N could stick with mod-N, and "never grow" isn't a plan.
• Why asynchronous replication (lag) and not synchronous. Enlace's data (short_code → long_url) tolerates a millisecond lag without drama, and asynchronous is faster. Cost: the "read your own write" problem —a just-created short_code resolved from a behind replica gives a fleeting 404—. Mitigation: read from the primary for a moment after creating (lesson 4). What would change it: if Enlace needed strong consistency on the immediate read, synchronous or permanent read-from-primary, at the cost of latency.

import math
TOTAL_TB, tb_per_shard = 12.0, 1.0
qps_read, hit_ratio, reads_per_replica = 8_000, 0.90, 2_000

shards = math.ceil(TOTAL_TB / tb_per_shard)          # 12 -> we round to 16 (power of 2)
SHARDS = 16
reads_to_db = qps_read * (1 - hit_ratio)             # 800/s
per_shard = reads_to_db / SHARDS                      # 50/s
replicas = max(2, math.ceil(per_shard / reads_per_replica) + 1)
print(SHARDS, f"{per_shard:.1f}/s per shard", f"{replicas} replicas",
      SHARDS * (1 + replicas), "nodes")
# 16 50.0/s per shard 2 replicas 48 nodes

• (a) Shards: 12 TB / 1 TB = 12 minimum; we round to 16 (power of 2, margin), ~750 GB per shard as before.
• (b) Reads per shard: the cache leaves (1 − 0.90) × 8,000 = 800/s to the database, spread among 16 shards it's 50/s per shard —just as tiny as before, because by doubling data and reads we also double the shards—.
• (c) Replicas and nodes: 50/s fits in 1 replica, but for redundancy 2 per shard; total 16 × 3 = 48 nodes. The lesson: when scaling proportionally (data and reads go up together, shards too), the load per node stays constant; the system grows "sideways" without any node heating up. And thanks to consistent hashing, going from 8 to 16 shards is done in increments (9, 10, …16), each moving only ~1/N, not all at once.

• (a) Per shard: 4,000 / 8 = 500 reads/s per shard (spread by hash). The sharding already divides the blow among 8: without sharding, a single node would receive the 4,000.
• (b) Yes, it holds up. Each shard has 2 replicas with a budget of 2,000/s each = 4,000/s of read capacity per shard; it receives 500/s. Even if one of the shard's replicas went down during the incident, the other alone (2,000/s) absorbs the 500/s with plenty of margin. The N−1 design and the sharding together make the mass miss —which without them would bring down a single DB— a non-event: 500/s per shard is a fraction of the budget.
• (c) Plan mitigations: (1) warm up the cache after a restart before exposing it to traffic, so the hit ratio doesn't start at 0; (2) the capacity slack —each node runs almost idle in the steady state (~48/s) precisely to absorb spikes like this—; and (3) the sharding, which spreads the blow among 8 nodes instead of concentrating it. The cache is the first line; replicas + sharding are the net beneath, sized to survive the day the cache fails.

• (a) 24 nodes is a waste → no, it's for the data and the redundancy, not for the throughput. You're right that 40 writes/s and 386 reads/s are trivial loads; the 24 nodes don't come from the throughput but from two things: (1) the 6 TB don't fit on one machine, and splitting them calls for 8 shards; and (2) each shard needs replicas to survive a crash (N−1 redundancy), hence the 2 per shard. The compute capacity is plenty —each node runs almost idle—; what isn't plenty is space (6 TB) or fault tolerance. You could use 6 shards instead of 8 to save, but not much lower without running out of storage or redundancy margin.
• (b) mod-N instead of consistent hashing → it spreads the same with fixed N, but when growing it's a storm. True that mod-N is one line and spreads evenly as long as N doesn't change. The problem appears the day you add shard 9: I measured it, it remaps 88.9% (888,920 keys, 5.33 TB) and cools the whole cache, against the 13.1% (0.78 TB) of consistent hashing. Since sharding exists to be able to grow, choosing the scheme that makes growth a catastrophe makes no sense. The ring's extra complexity is paid once; the mod-N storm is paid on every reshard.
• (c) long_url as shard key → no, the dominant read doesn't know it. The dominant operation is resolve(short_code) → long_url: a short_code arrives and the long_url has to be found. If I sharded by long_url, resolve wouldn't know which shard to go to (it doesn't have the long_url, it's precisely looking for it), so each resolution would ask all the shards —a cross query, slow, at 386/s—. The short_code is the correct key because the dominant read always brings it in hand, it's very high cardinality (62⁷), and hashed it spreads evenly (lesson 5).

The lesson of this exercise: a scaling plan isn't a stack of machines, it's a set of decisions —how many shards and why, how many replicas and why, what spreading scheme and why— that you know how to defend with the arithmetic and the module's experiments. Each objection is answered by acknowledging what's true and anchoring the decision in a measured number.