Lesson 5: Scaling and Replication

Why this lesson matters

In Lesson 2, we built a tiny key-value store. In Lesson 3, we made it survive restarts. In Lesson 4, we added indexes so reads could be smarter than scanning everything.

Now we hit a new wall: one machine is not enough.

Scaling asks:

• How do we split keys across many machines?
• How does a client know where a key lives?
• What happens when a machine dies?
• How do we keep copies synchronized?

This is the lesson where our datastore stops being just a data structure and starts becoming a distributed system.

Scaling has two separate jobs

When people say scale the database, they often mix two ideas:

1. Partitioning / Sharding

Split the keyspace across machines.

user:1  -> shard A
user:2  -> shard B
user:3  -> shard C

This increases capacity and throughput because each node owns fewer keys.

2. Replication

Store copies of the same data on multiple machines.

shard A primary -> shard A replica 1
                -> shard A replica 2

This improves availability and can improve read throughput.

A scalable datastore usually uses both: first decide which shard owns the key, then decide which replica serves the request.

Routing keys to shards

The simplest shard function is:

shard = hash(key) % number_of_shards

That works, but it has a painful property: if the number of shards changes, many keys move.

A common alternative is rendezvous hashing. For each key, score every node and choose the highest score. When you add a node, only keys that prefer the new node move.

This is useful in real systems because adding capacity should not require reshuffling the entire database.

```python
import hashlib


def score(key, node):
    data = f"{key}:{node}".encode()
    return int(hashlib.sha256(data).hexdigest(), 16)


def choose_node(key, nodes):
    return max(nodes, key=lambda node: score(key, node))

keys = ["account:1", "account:2", "trade:99", "price:AAPL", "price:MSFT"]
nodes_before = ["node-a", "node-b", "node-c"]
nodes_after = ["node-a", "node-b", "node-c", "node-d"]

print("Before adding node-d:")
before = {}
for key in keys:
    before[key] = choose_node(key, nodes_before)
    print(key, "->", before[key])

print("\nAfter adding node-d:")
moved = 0
for key in keys:
    new_node = choose_node(key, nodes_after)
    if new_node != before[key]:
        moved += 1
    print(key, "->", new_node)

print(f"\nMoved keys: {moved} of {len(keys)}")
```

In our key-value store, what is the main purpose of sharding?

A minimal sharded key-value cluster

Our earlier store had one dictionary-like object behind set, get, and delete.

A sharded version keeps the same external API, but adds a routing layer:

client.set("user:42", value)
        |
        v
router chooses shard
        |
        v
node stores key locally

This separation matters. Application code should not manually decide where keys live. The datastore should hide that behind a stable API.

```python
import hashlib

class Node:
    def __init__(self, name):
        self.name = name
        self.data = {}

    def set(self, key, value):
        self.data[key] = value

    def get(self, key):
        return self.data.get(key)

class ShardedKV:
    def __init__(self, node_names):
        self.nodes = [Node(name) for name in node_names]

    def _pick_node(self, key):
        digest = hashlib.sha256(key.encode()).hexdigest()
        index = int(digest, 16) % len(self.nodes)
        return self.nodes[index]

    def set(self, key, value):
        node = self._pick_node(key)
        node.set(key, value)
        print(f"SET {key!r} on {node.name}")

    def get(self, key):
        node = self._pick_node(key)
        value = node.get(key)
        print(f"GET {key!r} from {node.name} -> {value!r}")
        return value

cluster = ShardedKV(["shard-0", "shard-1", "shard-2"])
cluster.set("user:42", {"tier": "pro"})
cluster.set("price:AAPL", 189.25)
cluster.get("user:42")
cluster.get("price:AAPL")
```

Fill in the missing term:

A datastore that decides which node owns a key is using a key-to-_____ function.

Replication: copies with roles

For each shard, we can keep multiple copies:

primary replica: accepts writes
secondary replicas: copy writes from primary

A common simple protocol is leader-based replication:

1. Client sends write to the leader.
2. Leader appends the write to a log.
3. Followers read and apply log entries in order.
4. Reads can go to the leader or, with some staleness risk, followers.

This log-based idea connects directly to Lesson 3: persistence made writes durable on one machine; replication sends those writes to other machines too.

Implementing the core idea

Below is a deliberately small replication model.

The leader records each mutation in a log. Followers apply entries they have not seen yet. This is not a production protocol, but it captures the essential invariant:

Replicas converge when they apply the same ordered log of writes.

Notice that reads from a follower can be stale if replication has not caught up.

```python
class Replica:
    def __init__(self, name):
        self.name = name
        self.data = {}
        self.applied = 0

    def apply(self, entry):
        op, key, value = entry
        if op == "SET":
            self.data[key] = value
        elif op == "DELETE":
            self.data.pop(key, None)
        self.applied += 1

class Leader(Replica):
    def __init__(self, name):
        super().__init__(name)
        self.log = []

    def set(self, key, value):
        entry = ("SET", key, value)
        self.log.append(entry)
        self.apply(entry)
        print(f"leader accepted {entry}")

    def delete(self, key):
        entry = ("DELETE", key, None)
        self.log.append(entry)
        self.apply(entry)
        print(f"leader accepted {entry}")

    def replicate_to(self, follower):
        while follower.applied < len(self.log):
            entry = self.log[follower.applied]
            follower.apply(entry)
            print(f"replicated to {follower.name}: {entry}")

leader = Leader("primary")
follower = Replica("replica-1")

leader.set("order:7", {"status": "open"})
print("Follower before replication:", follower.data)

leader.replicate_to(follower)
print("Follower after replication:", follower.data)

leader.delete("order:7")
leader.replicate_to(follower)
print("Follower after delete:", follower.data)
```

True or false: In leader-based replication, a follower read may return stale data if the follower has not applied the leader's latest log entries.

Availability is a policy choice

Once data is replicated, the system must decide when an operation counts as successful.

Common write policies

• Async replication: leader replies immediately, then followers catch up later.
• Sync replication: leader waits for one or more followers before replying.
• Quorum replication: require acknowledgements from a majority-like subset.

For three replicas, a common quorum rule is:

write succeeds after 2 of 3 replicas acknowledge

This gives stronger durability than pure async replication, but increases latency. In finance-style systems, this tradeoff is serious: a fast acknowledgement that disappears during a crash may be unacceptable.

Failover

If the leader dies, the cluster needs to promote a follower. The safest follower is usually the one with the most complete log.

Next lesson, we'll look at deeper optimizations; for now, focus on the core design triangle: capacity, availability, and consistency.

A shard has 3 replicas. The system requires 2 acknowledgements before a write is considered successful. What is this policy mainly trying to improve compared with replying after only the leader writes locally?