Streams

None of the previous structures kept order and read position together. A set holds the member but forgets the order entries were added in; a counter holds the total but leaves no history; a list keeps order but leaves it to each reader to remember where it left off. The library’s three separate jobs — shelving, overdue notices, and search-index updates — each have to see the same event sequence in the order it happened and each from its own last position.

A stream combines this into a single structure: entries sit in the order they were added, every entry has a sortable id, and each consumer group remembers where it left off by holding only a boundary id. This lesson treats the stream not as a message channel but as a data structure that holds memory: how many bytes an entry takes, what the id is for, and when entries are discarded.

Entry ID

The id is built from two parts: the clock time when the entry was added and the sequence within that time. When the clock does not advance, the sequence advances, so two entries never receive the same id without needing a separate counter structure.

// entry-id.mjs — a stream entry's id and its bytes. The id is generated in <time>-<seq>
// form: when the clock does not advance, the sequence advances, so no two entries collide.
const generator = () => { let lastMs = -1, seq = 0;
  return (ms) => { if (ms === lastMs) seq += 1; else { lastMs = ms; seq = 0; } return `${ms}-${seq}`; }; };
const generate = generator();
const ids = [999, 1000, 1000, 1000, 1001, 1001].map(generate);
console.log("ids generated in one burst: " + ids.join("  "));

const parse = (k) => k.split("-").map(Number);
const isLess = (a, b) => { const [x, y] = parse(a), [p, q] = parse(b);
  return x !== p ? x < p : y < q; };
console.log(`numeric comparison  999-0 < 1000-0: ${isLess("999-0", "1000-0")}`);
console.log(`string comparison  999-0 < 1000-0: ${"999-0" < "1000-0"} (the id does not sort as a string)`);
const boundary = "1000-1";                    // the last id a consumer group has acknowledged
console.log(`entries to read after ${boundary}: ` + ids.filter((k) => isLess(boundary, k)).join(" "));

const METADATA = 24, POINTER = 8;             // id 16 + chain pointer 8; 8 per field
const entry = { type: "loan", reader: "418302", book: "9780000041173", branch: "branch-3" };
let total = METADATA;
console.log("\nfield".padEnd(8) + "name".padStart(4) + "value".padStart(7) + "pointer".padStart(10) +
  "bytes".padStart(6));
for (const [f, v] of Object.entries(entry)) {
  const b = Buffer.byteLength(f) + Buffer.byteLength(v) + POINTER; total += b;
  console.log(f.padEnd(8) + String(Buffer.byteLength(f)).padStart(4) +
    String(Buffer.byteLength(v)).padStart(7) + String(POINTER).padStart(10) + String(b).padStart(6));
}
console.log(`metadata`.padEnd(8) + "".padStart(4) + "".padStart(7) + "".padStart(10) +
  String(METADATA).padStart(6));
console.log(`entry total: ${total} bytes; without field names the same data would take ` +
  `${total - Object.keys(entry).reduce((s, f) => s + Buffer.byteLength(f), 0)} bytes`);

ids generated in one burst: 999-0  1000-0  1000-1  1000-2  1001-0  1001-1
numeric comparison  999-0 < 1000-0: true
string comparison  999-0 < 1000-0: false (the id does not sort as a string)
entries to read after 1000-1: 1000-2 1001-0 1001-1

field  name  value   pointer bytes
type       4      4         8    16
reader     6      6         8    20
book       4     13         8    25
branch     6      8         8    22
metadata                         24
entry total: 107 bytes; without field names the same data would take 87 bytes

The id does three jobs. The first is ordering: entries sort by id, and that order is the insertion order. The second is boundary: the only thing a consumer group holds is the last id it has acknowledged; the phrase “after 1000-1” reduces the group’s read position to sixteen bytes. The third is collision avoidance: three entries falling in the same millisecond become 1000-0, 1000-1, and 1000-2.

The line where the comparison is numeric is not a detail to skip. If the id were compared as a string, 999-0 and 1000-0 would sort in reverse order, because their digit counts differ; the ordering guarantee comes not from the id’s shape but from comparing it parsed.

The byte table makes a design decision visible. Of the entry’s 107 bytes, 20 are field names, and these names are rewritten on every entry. For 600,000 entries a day, that comes to 11.44 MiB just from repeating four words. Short field names or a position-based layout would cut this share; in exchange, the side reading the stream would have to know the field order, meaning the log’s self-description is lost.

A Day’s Stream

The stream’s memory cost is the product of the bytes per entry and how many entries are held, and the policy that trims determines the second factor.

Code	Assumption	Value	Rationale
DS14	daily activity entry	600,000	loan, return, hold, and shelf movement across all branches
DS15	entry metadata	24 bytes	id 16 bytes, chain pointer 8 bytes
DS16	consumer group	3 groups, 1500 / 1000 / 700 entries/min	the three jobs process at different speeds
DS17	hourly density curve	peak multiplier 2.8	the library is open by day, nearly idle at night

DS16 is this lesson’s deciding assumption: if the three groups’ speed were equal, the trimming policy alone would determine the memory; since it is not equal, the slowest group determines the memory.

// daily-stream.mjs — a day's activity stream: trimming policies and the memory effect of
// the unread share. The day is split into 1440 one-minute steps, arrivals are generated
// from hourly multipliers, the seed is visible.
const EVENTS = 600_000, SEED = 20260731, MIN = 1440, METADATA = 24, POINTER = 8;
const MULT = [0.05, 0.03, 0.02, 0.02, 0.03, 0.08, 0.2, 0.6, 1.6, 2.2, 2.6, 2.4,
              1.8, 2.0, 2.6, 2.8, 2.5, 2.0, 1.4, 0.9, 0.5, 0.3, 0.15, 0.08];
const GROUPS = [["shelving", 1500], ["overdue notice", 1000], ["search index", 700]];
let d = SEED;
const rand = () => { d = (d + 0x6D2B79F5) | 0; let t = Math.imul(d ^ (d >>> 15), 1 | d);
  t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 2 ** 32; };
const TYPES = ["loan", "return", "hold", "reshelving"];
const entry = (i) => ({ type: TYPES[Math.floor(rand() * 4)], reader: String(100000 + (i % 900000)),
  book: String(9780000000000 + (i % 120000)), branch: `branch-${1 + Math.floor(rand() * 6)}` });
const entryBytes = (e) => METADATA + Object.entries(e)
  .reduce((s, [f, v]) => s + Buffer.byteLength(f) + Buffer.byteLength(v) + POINTER, 0);

const scale = MULT.reduce((s, c) => s + c, 0) * 60;
const eb = new Uint16Array(EVENTS * 2), et = new Uint16Array(EVENTS * 2);
let n = 0;
for (let t = 0; t < MIN; t += 1) {
  const k = Math.round((EVENTS * MULT[Math.floor(t / 60)]) / scale);
  for (let i = 0; i < k; i += 1) { eb[n] = entryBytes(entry(n)); et[n] = t; n += 1; }
}
const cum = new Float64Array(n + 1);                          // cumulative bytes
for (let i = 0; i < n; i += 1) cum[i + 1] = cum[i] + eb[i];
const fmt = (x) => x.toLocaleString("en-US", { maximumFractionDigits: 2 });
const hourMin = (t) => `${Math.floor(t / 60)}:${String(t % 60).padStart(2, "0")}`;
let min = eb[0], max = eb[0];
for (let j = 1; j < n; j += 1) { if (eb[j] < min) min = eb[j]; if (eb[j] > max) max = eb[j]; }
console.log(`model: ${fmt(n)} entries/day, ${MIN} one-minute steps, seed ${SEED}`);
console.log(`entry bytes: average ${fmt(cum[n] / n)}, smallest ${min}, largest ${max}; ` +
  `the whole day ${fmt(Math.round(cum[n] / 1024 ** 2))} MiB\n`);

function run(label, trim, pause = null) {
  const cursor = GROUPS.map(() => 0);
  let head = 0, tail = 0, i = 0, peak = 0, peakBytes = 0, peakMin = 0, lost = 0;
  for (let t = 0; t < MIN; t += 1) {
    while (i < n && et[i] === t) { i += 1; tail += 1; }
    for (let g = 0; g < GROUPS.length; g += 1) {
      const stalled = pause && g === pause[0] && t >= pause[1] && t < pause[2];
      if (!stalled) cursor[g] = Math.min(tail, cursor[g] + GROUPS[g][1]);
    }
    head = Math.max(head, trim(tail, t, cursor));
    for (let g = 0; g < GROUPS.length; g += 1)
      if (cursor[g] < head) { lost += head - cursor[g]; cursor[g] = head; }
    if (tail - head > peak) { peak = tail - head; peakBytes = cum[tail] - cum[head]; peakMin = t; }
  }
  console.log(label.padEnd(26) + fmt(tail - head).padStart(11) + fmt(peak).padStart(14) +
    fmt(Math.round(peakBytes / 1024)).padStart(11) + hourMin(peakMin).padStart(9) +
    fmt(lost).padStart(11));
}
const firstIndex = (t, T) => { let l = 0, r = n; while (l < r) {
  const m = (l + r) >> 1; if (et[m] < t - T) l = m + 1; else r = m; } return l; };
console.log("policy".padEnd(26) + "day end".padStart(11) + "peak entries".padStart(14) +
  "peak KiB".padStart(11) + "peak min".padStart(9) + "missed".padStart(11));
run("no trimming", () => 0);
run("length 50,000", (tail) => tail - 50_000);
run("duration 60 min", (tail, t) => firstIndex(t, 60));
run("safe (slowest group)", (tail, t, im) => Math.min(...im));
run("safe + 30 min pause", (tail, t, im) => Math.min(...im), [2, 1020, 1050]);

console.log("\ngroup".padEnd(26) + "rate/min".padStart(9) + "peak unread".padStart(16) +
  "peak KiB".padStart(11) + "peak min".padStart(9) + "day end".padStart(10));
const cursor = GROUPS.map(() => 0), top = GROUPS.map(() => [0, 0, 0]);
let tail = 0, i = 0;
for (let t = 0; t < MIN; t += 1) {
  while (i < n && et[i] === t) { i += 1; tail += 1; }
  for (let g = 0; g < GROUPS.length; g += 1) {
    cursor[g] = Math.min(tail, cursor[g] + GROUPS[g][1]);
    if (tail - cursor[g] > top[g][0]) top[g] = [tail - cursor[g], cum[tail] - cum[cursor[g]], t];
  }
}
for (let g = 0; g < GROUPS.length; g += 1)
  console.log(GROUPS[g][0].padEnd(26) + fmt(GROUPS[g][1]).padStart(9) + fmt(top[g][0]).padStart(16) +
    fmt(Math.round(top[g][1] / 1024)).padStart(11) + hourMin(top[g][2]).padStart(9) +
    fmt(tail - cursor[g]).padStart(10));

model: 600,000 entries/day, 1440 one-minute steps, seed 20260731
entry bytes: average 109, smallest 107, largest 113; the whole day 62 MiB

policy                        day end  peak entries   peak KiB peak min     missed
no trimming                   600,000       600,000     63,865    23:59          0
length 50,000                  50,000        50,000      5,321     8:45     38,920
duration 60 min                 1,856        63,488      6,757    15:59     46,220
safe (slowest group)                0        88,920      9,465    17:59          0
safe + 30 min pause                 0       109,920     11,701    17:59          0

group                     rate/min     peak unread   peak KiB peak min   day end
shelving                      1,500               0          0     0:00         0
overdue notice                1,000           2,520        268    15:59         0
search index                    700          88,920      9,465    17:59         0

These numbers are in the measurement class; the distribution of entries depends on the hourly curve and the seed, the difference between the policies does not.

The Unread Share

The upper table compares four policies over the same day. No trimming brings the stream to 63,865 KiB and misses no entry: history stays queryable, at the cost of the whole day sitting in memory. Length 50,000 fixes memory at 5,321 KiB — twelve times less — but the search index loses 38,920 entries without ever seeing them. Duration 60 min looks more measured at first glance, holding only 1,856 entries at day’s end; yet it peaks at 63,488 entries and misses 46,220. The difference is what each policy fixes: a length limit fixes memory, a duration limit fixes age and leaves memory to the arrival rate. If density tripled, the duration-limited stream would hold three times the space.

Safe trimming is the fourth option: an entry is not deleted until every group has read it. Missed entries drop to zero, bytes held peak at 9,465 KiB and reset to zero at day’s end. But the ceiling is no longer in the operator’s hands. The lower table shows this: shelving never falls behind, overdue notice accumulates a peak of 2,520 entries, search index 88,920. The entire 9,465 KiB the stream holds is the share the slowest group has not read — not because the stream is a queue, but because it cannot discard anything before the furthest-behind boundary id.

The last row measures exactly how sharp this is. If the search index pauses for thirty minutes at peak hour, the stream climbs to 109,920 entries: exactly 21,000 entries and 2,236 KiB more — the pause duration times the group’s rate. The memory cost here belongs not to the stream but to the decision of the side that is not reading it.

The reverse is equally clear: the number of readers does not increase memory. Three groups hold three boundary ids, sixteen bytes each for forty-eight bytes total; adding a fourth, fifth, or tenth group to the stream adds sixteen bytes to a 62 MiB body. This is what separates a stream from a list: a list is consumed as it is read and a second reader needs a second copy, while a stream is held once and the position stays on the reader’s side.

Summary

• The entry id is built from time and sequence and does three jobs: it gives the insertion order, separates entries within the same millisecond, and reduces a consumer group’s read position to sixteen bytes; the comparison must be numeric — a string comparison sorts 999-0 and 1000-0 in reverse.
• A share of the entry byte count is field names repeated on every entry: 20 of 107 bytes, 11.44 MiB a day. Shortening them costs the stream its self-description.
• A length limit fixes memory (5,321 KiB), a duration limit fixes age and leaves memory to the arrival rate (peaking at 63,488 entries); both miss 38,920 and 46,220 entries respectively because neither knows the slowest group.
• Under safe trimming, missed entries are zero but the ceiling is not the operator’s: the entire 9,465 KiB held is the slowest group’s unread share, and if that group pauses for thirty minutes the stream grows by 21,000 entries (2,236 KiB).
• The number of readers does not increase memory, the number of non-readers does: the three groups’ total ledger cost is forty-eight bytes.

Next Step

Because the stream’s id is sortable, the question “everything after now” was a position lookup, not a search. The library has one more question, and it has no sortable key: a reader wants the collection point nearest to where they stand. Every way of sorting a two-dimensional position along a single axis pushes some neighbors far apart, so neither a sorted set nor a stream answers this question directly. The next lesson builds the same proximity query with an in-memory structure and compares it against how many candidates a flat scan has to examine.