Publish-Subscribe

The previous lesson ran collision control on a single key: whether the value a client read had been changed by someone else was seen through the watch, and a colliding transaction was retried. There, communication was established indirectly, between two clients touching the same key. A client saying something directly to another client — the store used as a transmission channel — was never covered.

In-memory stores carry a separate path for this: a client publishes to a channel, and clients subscribed to that channel receive the message. Nothing gets written to the key space. What this lesson measures is not that path’s speed, but what it does not guarantee.

The Store’s Publish Path

In the library system, loan events land on branch boards instantly. The mechanism below sets up three branch channels, one board per branch, and an audit subscriber that joins at step three hundred. Board connectivity drops on a deterministic schedule; instead of a real network outage, a rule based on the step number is used, so the numbers come out the same on every run (CU10).

// publish-loss.mjs — a message published to a channel with no subscriber is lost; the loss is counted
class Store {                                       // publish-subscribe: no record, no rewind
  constructor() { this.channels = new Map(); }      // channel -> Set(subscriber)
  subscribe(channel, a) {
    if (!this.channels.has(channel)) this.channels.set(channel, new Set());
    this.channels.get(channel).add(a);
  }
  unsubscribe(channel, a) {
    this.channels.get(channel)?.delete(a);
  }
  publish(channel, message) {                        // return value: number of subscribers reached
    const set = this.channels.get(channel);
    if (!set || set.size === 0) return 0;             // no subscriber: message is written nowhere
    for (const a of set) a.receive(message);
    return set.size;
  }
}

const events = (count) => {                          // deterministic workload, seed visible
  let seed = 20250731;
  const rand = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: count }, (_, i) => ({
    step: i + 1, branch: Math.floor(rand() * 3) + 1, bookId: Math.floor(rand() * 400) + 1,
  }));
};

const store = new Store();
const channel = (s) => `branch:${s}:loans`;
const boards = [1, 2, 3].map((s) => ({ name: `board${s}`, branch: s, received: 0, receive() { this.received += 1; } }));
const audit = { name: "audit", received: 0, receive() { this.received += 1; } };
const counters = [1, 2, 3].map(() => ({ published: 0, delivered: 0, unsubscribed: 0 }));

// Board connectivity: each board disconnects periodically (deterministic schedule, no randomness).
const boardOpen = (s, step) => (step + s * 37) % 100 >= 25;   // connected in 75% of steps
for (const e of events(600)) {
  for (const b of boards) {                          // connection state translates to subscription
    const open = boardOpen(b.branch, e.step);
    if (open) store.subscribe(channel(b.branch), b); else store.unsubscribe(channel(b.branch), b);
  }
  if (e.step === 301) for (const s of [1, 2, 3]) store.subscribe(channel(s), audit);  // late-joining subscriber

  const c = counters[e.branch - 1];
  const reached = store.publish(channel(e.branch), e);
  c.published += 1; c.delivered += reached;
  if (reached === 0) c.unsubscribed += 1;
}

console.log(["channel", "published", "delivered", "unsubscribed publish", "missed by board"]
  .map((h, i) => (i === 0 ? h.padEnd(16) : h.padStart(22))).join(""));
let total = [0, 0, 0, 0];
for (const [i, c] of counters.entries()) {
  const missed = c.published - boards[i].received;
  total = total.map((t, j) => t + [c.published, c.delivered, c.unsubscribed, missed][j]);
  console.log(channel(i + 1).padEnd(16) +
    [c.published, c.delivered, c.unsubscribed, missed].map((n) => String(n).padStart(22)).join(""));
}
console.log("total".padEnd(16) + total.map((n) => String(n).padStart(22)).join(""));
console.log(`received by the late subscriber: ${audit.received}  (published before it joined: 300)`);

channel                      published             delivered  unsubscribed publish       missed by board
branch:1:loans                     199                   266                    21                    46
branch:2:loans                     195                   227                    31                    55
branch:3:loans                     206                   260                    22                    46
total                              600                   753                    74                   147
received by the late subscriber: 300  (published before it joined: 300)

Three numbers need reading. The delivered column is larger than the published count (753 > 600), because a message gets copied once for every subscriber present, and the publisher does not know that count in advance. The unsubscribed publish column says that 74 messages were written nowhere at all: no one was on the channel at that moment, and the message vanished at the step it was produced. The missed by board column is larger still (147), because when one board is disconnected but another subscriber is connected, the message is not lost — it just never reaches that board.

The last line is the sharpest one. The audit subscriber joined at step three hundred and received exactly 300 messages; it has no way to see the 300 that came before it. No trace of those messages remains anywhere in the store’s key space. In the publish-subscribe path, delivery reduces to whether the moment a message is published and the moment a subscriber is connected happen to overlap.

The Slow Subscriber’s Buffer

Loss does not come only from a broken connection. When a subscriber is connected but messages are produced faster than it reads them, the unread messages pile up in an output buffer the store holds. That buffer eats memory, and it is invisible in the publisher’s accounting. The run below drives the subscriber at different read speeds and counts the bytes the buffer holds. Once the buffer exceeds 32,768 bytes, the store drops the subscriber; for a dropped subscriber, every publish after that is a loss.

// slow-subscriber.mjs — a slow subscriber's output buffer: bytes held and the step it hits the limit
const MESSAGE = (step, branch, bookId) =>          // body carried verbatim; its bytes are measured
  JSON.stringify({ event: "loan", step, branch, bookId });

function run(readInterval, byteLimit, count = 2000) {
  let seed = 20250731;
  const rand = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  const buffer = [];                               // messages not yet written to the subscriber
  let bytes = 0, peak = 0, dropStep = 0, afterDrop = 0, read = 0;
  for (let step = 1; step <= count; step += 1) {
    const g = MESSAGE(step, Math.floor(rand() * 3) + 1, Math.floor(rand() * 400) + 1);
    if (dropStep) { afterDrop += 1; continue; }      // dropped subscriber: publish is a loss
    buffer.push(g); bytes += Buffer.byteLength(g);   // the store writes to the buffer first
    peak = Math.max(peak, bytes);
    if (bytes > byteLimit) { dropStep = step; continue; }   // limit exceeded: subscriber is dropped
    if (step % readInterval === 0) {                 // subscriber reads once every `readInterval` steps
      const c = buffer.shift();
      if (c !== undefined) { bytes -= Buffer.byteLength(c); read += 1; }
    }
  }
  return { readInterval, read, peak, pending: buffer.length, dropStep, afterDrop };
}

console.log(`message body: ${MESSAGE(1, 1, 1)}  (${Buffer.byteLength(MESSAGE(1, 1, 1))} bytes)`);
console.log(["read interval", "read", "peak bytes", "pending in buffer", "drop step", "after drop"]
  .map((h, i) => (i === 0 ? h.padEnd(15) : h.padStart(19))).join(""));
for (const interval of [1, 2, 4, 10]) {
  const r = run(interval, 32768);
  console.log(`every ${String(interval).padStart(2)} steps`.padEnd(15) +
    [r.read, r.peak, r.pending, r.dropStep, r.afterDrop]
      .map((n) => String(n).padStart(19)).join(""));
}

message body: {"event":"loan","step":1,"branch":1,"bookId":1}  (47 bytes)
read interval                 read         peak bytes  pending in buffer          drop step         after drop
every  1 steps                2000                 52                  0                  0                  0
every  2 steps                 639              32800                641               1280                720
every  4 steps                 214              32777                646                860               1140
every 10 steps                  71              32801                647                718               1282

A subscriber that keeps up adds 52 bytes to memory: the buffer never holds more than one message. When the read interval grows to two, the same subscriber reaches 32,800 bytes and gets dropped at step 1,280; the remaining 720 publishes do not exist for it. When the interval grows to ten, the limit is hit at step 718, and the loss climbs to 1,282 messages.

The budget relationship here runs in reverse. A subscriber’s slowness is paid out of the store’s memory, and that cost is per subscriber: a hundred slow subscribers with the same limit means a hundred separate buffers, up to 3.2 MB in the worst case. The limit is the store’s way of protecting itself; its cost is that a subscriber that hits the limit gets dropped silently. The publisher sees none of this, because the publish call does not wait for the message to be read (CU11).

Persistent Versus Not

The same event stream can also be built with the store’s stream structure. The difference is a single point: a stream stores every entry with an id, so a subscriber that arrives late or reconnects after dropping can read from where it left off. The price of that is memory. The run below passes 2,000 events through three paths; the subscriber drops at step 800 and returns at step 1,500. A 16-byte pointer/length overhead per entry is assumed (CU12).

// stream-comparison.mjs — same event stream: bytes for a non-persistent publish vs. a trimmed stream
const OVERHEAD = 16;                                // per-entry id/length overhead (my own accounting)
const MESSAGE = (step, branch, bookId) => JSON.stringify({ event: "loan", step, branch, bookId });
const DROP = 800, RETURN = 1500, COUNT = 2000;      // subscriber drops at 800, returns at 1500

function events(count) {
  let seed = 20250731;
  const rand = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: count }, (_, i) =>
    MESSAGE(i + 1, Math.floor(rand() * 3) + 1, Math.floor(rand() * 400) + 1));
}

function pubsubPath() {                             // non-persistent: nothing but the channel record
  let bytes = 0;
  for (const name of ["branch:activity", "board"]) bytes += Buffer.byteLength(name) + OVERHEAD;
  return { name: "publish-subscribe", returnBytes: bytes, endBytes: bytes, entries: 0, recovered: 0 };
}

function streamPath(bodies, trim) {                  // persistent: entry id and body sit in memory
  const stream = [];
  let bytes = 0, returnBytes = 0, recovered = 0;
  const size = (g) => Buffer.byteLength(g.id) + Buffer.byteLength(g.body) + OVERHEAD;
  bodies.forEach((body, i) => {
    const entry = { id: `${i + 1}-0`, body };
    stream.push(entry); bytes += size(entry);
    while (trim && stream.length > trim) bytes -= size(stream.shift());   // oldest entry drops
    if (i + 1 === RETURN) {                          // subscriber returns at exactly this step
      returnBytes = bytes;
      recovered = stream.filter((g) => {              // only entries still in the window are recoverable
        const step = Number(g.id.split("-")[0]);
        return step >= DROP && step < RETURN;
      }).length;
    }
  });
  return { name: trim ? `stream (trim=${trim})` : "stream (no trim)",
           returnBytes, endBytes: bytes, entries: stream.length, recovered };
}

const bodies = events(COUNT);
const gap = RETURN - DROP;                           // events the subscriber was disconnected for
console.log(`events=${COUNT}  gap=${gap} events (steps ${DROP}-${RETURN - 1})`);
console.log(["path", "bytes@1500", "bytes@2000", "entries", "recovered", "lost"]
  .map((h, i) => (i === 0 ? h.padEnd(20) : h.padStart(13))).join(""));
for (const r of [pubsubPath(), streamPath(bodies, 500), streamPath(bodies, 0)]) {
  console.log(r.name.padEnd(20) +
    [r.returnBytes, r.endBytes, r.entries, r.recovered, gap - r.recovered]
      .map((n) => String(n).padStart(13)).join(""));
}

events=2000  gap=700 events (steps 800-1499)
path                   bytes@1500   bytes@2000      entries    recovered         lost
publish-subscribe              52           52            0            0          700
stream (trim=500)           36865        36862          500          499          201
stream (no trim)           108381       145243         2000          700            0

The three rows show this course’s rule exactly as it is. The publish-subscribe path holds 52 bytes; that is nothing more than the channel name and the subscription record, and it never grows with event count. In exchange, all 700 of the dropped subscriber’s events are lost.

The trimmed stream holds 36,865 bytes and returns 499 of the gap; the remaining 201 events are lost because they were trimmed away before the subscriber returned. The trim length is directly a recovery window here: 500 entries means 500 events of lookback.

The untrimmed stream loses nothing, but holds 108,381 bytes at step 1,500 and 145,243 at step 2,000. The growth between those two steps is the real warning: an untrimmed stream’s memory grows without bound as events accumulate, and once a memory limit kicks in, that growth forces the eviction of other data.

Where Publishing Fits

These measurements do not make the publish-subscribe path worthless; they set its place. Work where loss is acceptable is work where missing a message is correctable: a board’s live counter, a cache-invalidation notice, an announcement of a configuration change. In every one of these, a subscriber makes up what it missed on the next full read.

For work where loss is not acceptable, the measured cost gets paid: a loan record posting to accounting, a late fee being processed, the record of a book returning to the shelf. This work either gets written to a persistent stream or handed off outside the store, to a structure that gives a delivery guarantee. The distinction is not made by asking “which is faster” but by asking “can a missed message be recovered.”

Summary

• The publish-subscribe path does not write the message to the key space: 74 of 600 publishes vanished without reaching anywhere because no subscriber was on the channel at that moment, and boards missed 147 messages besides.
• The delivered count is independent of the published count; 600 publishes produced 753 delivered copies, and the publisher does not know that number in advance.
• A late-joining subscriber cannot see any of the 300 messages published before it joined; the history sits nowhere in the store.
• The cost of a slow subscriber is the store’s memory: when the read interval grew to two, the output buffer reached 32,800 bytes, the subscriber was dropped at step 1,280, and the next 720 publishes were lost for it.
• When the same stream is built with a persistent structure, it holds 36,865 bytes (trim 500) or 145,243 bytes (no trim) instead of 52; what that buys is 499 and 700 recoverable events respectively.

Next Step

The lessons so far measured the store’s abilities one at a time: replication, failover, partitioning, collision control, and now publishing. Each was defined by its own cost, but none of them answered end to end the question “which structure does this job get built with in this store, and how many bytes does it hold.” The next lesson takes on four typical jobs from the library system — the catalog cache, session records, a rate limit, and a popular-book leaderboard — one at a time; it fits each one to a structure and gathers each one’s memory accounting into the same table.