Replication

The previous lesson tied the store’s type to the access pattern and shortened the read path. There is still a single copy on the table: one node taking 236.11 operations per second, carrying 718.24 GB, and leaving both the tracking query and the pricing job without an answer when it goes down. The first way to spread load across more than one machine is not to split the data but to replicate it: keep the same data in more than one place.

The mechanics of replication — log streaming, replication lag, synchronous acknowledgment’s cost, the majority rule, failover — were established and measured in the Relational Database Administration course and are not repeated here. This lesson’s question is a scaling decision: which number does adding a replica bring down, which does it leave untouched, and what grows in exchange. The consistency models were also defined in the Introduction to System Design course; here they are only used.

What Replication Splits

In a leader–follower layout, one node accepts writes and the others apply its changes and serve reads. What splits is the read; the write does not, because every node must apply every write. The calculation below opens that asymmetry up by replica count, and the second table gives the read side’s counterpart to asynchronous lag.

VD2 — a shipment stays in transit for 3 days. Rationale: the time between acceptance and delivery; V4’s seven state events spread across this span, and tracking queries arrive for shipments inside it. Its sensitivity is linear — doubling the span doubles the number of shipments in transit and halves the stale response ratio. This assumption is not added to K01’s table.

// replication/leader-follower.mjs — reads split across replicas while writes repeat at
// every node, and asynchronous lag converted into stale tracking responses. All arithmetic.
const READ = 41.67;         // K01: read/s behind the cache
const WRITE = 194.44;       // lesson 01: store write ops/s in the split layout
const STORED = 718.24;      // lesson 01: stored data GB
const EVENTS = 97.22;       // K01: peak write requests/s (state event)
const V3 = 400_000;         // K01: daily shipments
const VD2 = 3;               // this lesson's assumption: how long a shipment stays in transit (days)
const INTRANSIT = V3 * VD2, DAY = 86_400;
const b = (x, n = 2) => x.toFixed(n);

console.log(`${"replicas".padStart(9)}${"read/s per replica".padStart(20)}${"write/s per node".padStart(19)}` +
  `${"node load".padStart(12)}${"write/read".padStart(13)}${"replica gain".padStart(14)}${"total GB".padStart(11)}`);
let previous = null;
for (const n of [1, 2, 3, 4, 5]) {
  const read = READ / n, load = WRITE + read;
  console.log(`${String(n).padStart(9)}${b(read).padStart(20)}${b(WRITE).padStart(19)}` +
    `${b(load).padStart(12)}${b(WRITE / read).padStart(13)}` +
    `${(previous === null ? "-" : b(previous - load)).padStart(14)}${b(STORED * (1 + n)).padStart(11)}`);
  previous = load;
}
console.log(`single node (lesson 01): load ${b(WRITE + READ)} ops/s, write/read ${b(WRITE / READ)}, ` +
  `${b(STORED)} GB`);

console.log(`\nshipments in transit = ${INTRANSIT.toLocaleString("en-US")} (VD2 = ${VD2} days)`);
console.log(`${"lag s".padStart(9)}${"changed shipments in window".padStart(29)}` +
  `${"stale read ratio".padStart(19)}${"stale responses/s".padStart(19)}${"stale responses/day".padStart(21)}`);
for (const L of [0.5, 1, 5, 10]) {
  const changed = EVENTS * L, ratio = changed / INTRANSIT, stale = READ * ratio;
  console.log(`${b(L, 1).padStart(9)}${b(changed).padStart(29)}${ratio.toExponential(2).padStart(19)}` +
    `${b(stale, 5).padStart(19)}${b(stale * DAY, 1).padStart(21)}`);
}

 replicas  read/s per replica   write/s per node   node load   write/read  replica gain   total GB
        1               41.67             194.44      236.11         4.67             -    1436.48
        2               20.84             194.44      215.28         9.33         20.84    2154.72
        3               13.89             194.44      208.33        14.00          6.95    2872.96
        4               10.42             194.44      204.86        18.66          3.47    3591.20
        5                8.33             194.44      202.77        23.33          2.08    4309.44
single node (lesson 01): load 236.11 ops/s, write/read 4.67, 718.24 GB

shipments in transit = 1,200,000 (VD2 = 3 days)
    lag s  changed shipments in window   stale read ratio  stale responses/s  stale responses/day
      0.5                        48.61            4.05e-5            0.00169                145.8
      1.0                        97.22            8.10e-5            0.00338                291.7
      5.0                       486.10            4.05e-4            0.01688               1458.4
     10.0                       972.20            8.10e-4            0.03376               2916.8

These numbers belong to the calculation class; their inputs are K01’s arithmetic, lesson 01’s results, and VD2.

Reading the Numbers

Adding a replica does not lower the write load at all — the write-per-node column reads 194.44 ops/s across all five rows. This is a floor, not a ceiling: whatever the replica count, every node applies every state event. The only column that splits is the read, starting from 41.67.

The gain runs out quickly: going from the first replica to the second drops node load 20.84 ops/s; second to third, 6.95; fourth to fifth, 2.08. The fifth replica delivers a tenth of what the first one did, because the only share that can shrink is the read, already 18 percent of the load. Leader–follower replication is, for that reason, a read-heavy tool; K01’s arithmetic, where the store sees it, is write-heavy, and lesson 01’s document update grew that weight further.

The write/read ratio at store grows with every replica: 2.33 in K01, 4.67 after lesson 01, 14.00 at three replicas. The ratio is not degradation but a definition of the layout — as reads spread out, each node’s mix skews toward writes, and that is where the decision to tune a node for writes gets read off.

stored data GB is multiplied outright: 718.24 GB on a single node becomes 2,872.96 GB with three replicas and 4,309.44 GB with five. This is replication’s bill, and the decision that moves K01’s storage row most — the original 712.48 GB calculation can climb to six times its size.

Lag turns into a response count in the second table. At one second of replication lag, changed shipments in the window number 97.22; against the 1,200,000 shipments in transit that is a ratio of 8.10e-5, and the payoff for the 41.67 reads reaching the store is 291.7 stale tracking responses a day — 1,458.4 at five seconds. The number is small because the divisor is large: the stale response ratio is the change rate over shipments in transit, and a short-lived, fast-changing data set would give a different result. Whether reading from a replica is acceptable is, for that reason, not a principle but a result of this division.

A replica’s second payoff is service availability: when the leader goes down, one of the replicas is promoted. Failover and the majority rule were measured in the Relational Database Administration course and are not repeated here.

Multi-Leader and Collision

As long as writes stay on one node, writes do not scale. A multi-leader layout lifts that constraint: more than one node accepts writes and applies the others’ changes — at the cost of something the single leader gave for free, a single order for writes.

The pattern makes this concrete. P2 is the state event write by carrier, and at the moment of transfer, a shipment sits between two zones: the departure zone writes “handed to transfer” and the arrival zone writes “received from transfer” a few seconds apart.

VD3 — a shipment’s rate of near-simultaneous writes from two zones is 0.02. Rationale: one of the seven events falls on the transfer moment, and both zones write at that moment. Its sensitivity is linear; it is given below at 0.04. Not added to K01’s table.

// replication/multi-leader.mjs — two leaders writing the same shipment. This is a MODEL: a
// round is an abstract step, propagation round is a parameter, not a measured duration. What
// is counted is whether the two nodes show the same final state once propagation is done.
const EVENTS = 20_000, SHIPMENTS = 4000, PROPAGATION = 3;   // model parameters
const VD3 = 0.02;            // assumption: a shipment's rate of near-simultaneous writes from two zones
let s = 20260730 % 2147483647;                        // fixed seed: same sequence on every run
const rand = () => (s = (s * 48271) % 2147483647) / 2147483647;

const events = [];
for (let t = 0; t < EVENTS; t += 1) {
  const sh = Math.floor(rand() * SHIPMENTS), zone = rand() < 0.5 ? "A" : "B";
  events.push({ t, sh, zone, label: `o${t}` });
  if (rand() < VD3) {                                  // transfer moment: the opposite zone writes too
    const delay = 1 + Math.floor(rand() * PROPAGATION);
    events.push({ t: t + delay, sh, zone: zone === "A" ? "B" : "A", label: `o${t}b` });
  }
}
const shipmentEvents = new Map();
for (const e of events) (shipmentEvents.get(e.sh) ?? shipmentEvents.set(e.sh, []).get(e.sh)).push(e);
let overlapping = 0;
for (const list of shipmentEvents.values())
  if (list.some((a) => list.some((c) => a.zone !== c.zone && Math.abs(a.t - c.t) <= PROPAGATION))) overlapping += 1;

function run(rule) {
  const node = { A: new Map(), B: new Map() };
  for (const d of ["A", "B"]) {
    const incoming = events.map((e) => ({ arrival: e.zone === d ? e.t : e.t + PROPAGATION, local: e.zone === d ? 1 : 0, e }));
    incoming.sort((x, y) => x.arrival - y.arrival || y.local - x.local);   // arrival order, local first
    for (const { e } of incoming) {
      if (rule === "last-writer") { node[d].set(e.sh, e.label); continue; }
      const k = node[d].get(e.sh) ?? node[d].set(e.sh, []).get(e.sh);     // append: event set
      k.push(e);
    }
  }
  const finalState = (d, sh) => {
    const v = node[d].get(sh);
    if (rule === "last-writer") return v;
    return v.slice().sort((x, y) => y.t - x.t || (x.label < y.label ? 1 : -1))[0].label;   // deterministic rule
  };
  let diverged = 0;
  for (const sh of node.A.keys()) if (finalState("A", sh) !== finalState("B", sh)) diverged += 1;
  const records = rule === "last-writer" ? node.A.size
    : [...node.A.values()].reduce((a, v) => a + v.length, 0);
  return { diverged, records, touched: node.A.size };
}

console.log(`model: ${events.length} events, ${shipmentEvents.size} shipments touched, ` +
  `propagation ${PROPAGATION} rounds, VD3 = ${VD3}`);
console.log(`shipments with near-simultaneous writes from two zones = ${overlapping} ` +
  `(${((100 * overlapping) / shipmentEvents.size).toFixed(2)}% of touched)`);
console.log(`\n${"merge rule".padEnd(22)}${"diverged shipments".padStart(20)}` +
  `${"share of overlapping".padStart(22)}${"records kept at node".padStart(22)}`);
for (const rule of ["last-writer", "append"]) {
  const r = run(rule);
  console.log(`${rule.padEnd(22)}${String(r.diverged).padStart(20)}` +
    `${(r.diverged / overlapping).toFixed(4).padStart(22)}${String(r.records).padStart(22)}`);
}

const DAILY_EVENTS = 2_800_000, V4 = 7;                // K01: daily state events, events/shipment
const ratio = run("last-writer").diverged / shipmentEvents.size;
console.log(`\nscaled to K01: ${DAILY_EVENTS.toLocaleString("en-US")} events and ` +
  `${(DAILY_EVENTS / V4).toLocaleString("en-US")} shipments a day (V4 = ${V4})`);
console.log(`under the last-writer rule, ${Math.round((DAILY_EVENTS / V4) * ratio).toLocaleString("en-US")} ` +
  `shipments a day show a different final state in the two zones; VD3 = 0.04 would be about twice that`);

model: 20365 events, 3971 shipments touched, propagation 3 rounds, VD3 = 0.02
shipments with near-simultaneous writes from two zones = 363 (9.14% of touched)

merge rule              diverged shipments  share of overlapping  records kept at node
last-writer                             66                0.1818                  3971
append                                   0                0.0000                 20365

scaled to K01: 2,800,000 events and 400,000 shipments a day (V4 = 7)
under the last-writer rule, 6,648 shipments a day show a different final state in the two zones; VD3 = 0.04 would be about twice that

The model says three things. First, collision is not rare: at an average of five events per touched shipment, a 2-percent-per-event transfer rate turns 9.14 percent of shipments into ones written from both zones.

Second, not every colliding write produces divergence — only 66 of 363 shipments, one in five. Divergence arises only when the two events reach the two nodes in a different order: node A applies its local event first, node B applies the other one first, and once propagation finishes the two nodes hold two different final states that nothing corrects on its own. Scaled to K01, that is 6,648 shipments a day with a different final state in the two zones; doubling VD3 roughly doubles the number.

Third, and the important point for design: divergence is a consequence of the storage shape, not of the data. Under the append rule the same event sequence produces zero divergence, because the merge is a set union and a union does not depend on order — the final state derives from the set by a deterministic rule, and the two nodes reach the same result because they hold the same set. The cost sits in the last column: records held at a node rise from 3,971 to 20,365, a 5.13× factor. In this course’s flow that cost is on paper only, since K01 already stores every event (V6, V12); what actually changes is that the final state stops being an overwritten field and becomes a derived value.

P2’s shape is, for that reason, decisive: a state event is an insert, so it fits a multi-leader layout, while an overwritten field would not. The multi-leader decision questions not the data but the shape of the write. Replicas converging once propagation finishes is what eventual consistency promises, defined in the Introduction to System Design course; this measurement shows the promise holds only when the merge rule is order-independent.

Summary

• In a leader–follower layout, the read splits and the write does not: write per node stays at 194.44 ops/s across all five replicas, while read per replica falls from 41.67 to 8.33.
• The gain from adding a replica runs out quickly: node load drops 20.84 ops/s on the first addition and only 2.08 on the fourth; the write/read ratio at store climbs from 4.67 to 14.00 at three replicas.
• stored data GB is multiplied outright: 718.24 GB becomes 2,872.96 GB with three replicas and 4,309.44 GB with five.
• Lag converts into a response count: with VD2 = 3 days and 1,200,000 shipments in transit, one second of lag produces 291.7 stale tracking responses a day, and five seconds produces 1,458.4.
• Under multi-leader with VD3 = 0.02, 9.14 percent of shipments are written from two zones; under the last-writer rule, one in five of those produces permanent divergence — 6,648 shipments a day at K01’s scale.
• Divergence arises from the storage shape: the append rule produces zero divergence, at the cost of records held at a node climbing from 3,971 to 20,365 (a 5.13× factor).

Next Step

Replication split the read but left the same 194.44 write on every node and multiplied stored data by the replica count. Both limits come from the same place: all the data sits everywhere. Yet the data the store carries is not a single whole — pricing’s questions and delivery operations’ questions touch different records, as lesson 01’s scan measurement already showed. The next lesson splits the data not by key but by function: once the two contexts split into their own stores, what happens to the requests and data each carries, how many stores the write path stretches across, how many stores a cross-context query has to reach, and what duplicating the shared fields adds to stored data.