Streams versus Queues

The previous lesson’s final measurement left a gap: the search index, connecting later, never saw three of the five loan-issued events, because a copy is only dropped into subscriptions that exist at the moment of publication. The same gap had a second face — because a processed copy gets deleted, no consumer could go back and reread it.

Both shortcomings come from a single assumption: a delivered message has been consumed, and a consumed message gets deleted. This lesson builds the second transport model, the one that drops that assumption, and compares both models on the same workload.

Append and Offset

A stream is an ordered record where messages are only appended at the end and stay in place once read. The structure itself is not new: in the Relational Database Administration course, the write-ahead log worked by the same rule — records are appended, no record is modified in place. A stream uses the same structure to carry messages.

It differs from a queue in two respects. First, reading does not delete: a read call does not change the log. Second, how far a consumer has reached is tracked not by the queue but by the consumer. This value is called the offset, and it marks a record’s position in the log.

// log.mjs — an append-only log: reading does not delete a record, the consumer holds the offset
import { appendFileSync, readFileSync, writeFileSync, existsSync } from "node:fs";

export function reset(file) { writeFileSync(file, ""); }

export function append(file, record) {              // writing only ever happens at the end
  appendFileSync(file, JSON.stringify(record) + "\n");
}

export function records(file) {
  if (!existsSync(file)) return [];
  const t = readFileSync(file, "utf8").trimEnd();
  return t === "" ? [] : t.split("\n").map((s) => JSON.parse(s));
}

// returns up to `count` records starting at offset; reading does not change the log
export function read(file, offset, count) {
  return records(file).slice(offset, offset + count);
}

That append‘s only write form is appendFileSync is not a convenience, it is the model’s rule. Appending never touches the file’s earlier bytes; that is why the data underneath a consumer reading at the same time never changes, and reads and writes never lock each other out.

One Copy, Three Offsets

In the previous lesson we measured the fan-out multiplier: as subscriber count rose, the number of copies dropped was multiplied along with it. On a stream, the copy count is one; what gets multiplied is not the copy but the offset value held. The run below stores the same six events in both forms and compares the space they take on disk.

// three-consumers.mjs — a single log, an offset per consumer; offsets are held instead of copies
import { statSync } from "node:fs";
import { reset, append, records, read } from "./log.mjs";

const EVENTS = [
  { type: "loan_issued", loanId: 1, bookId: 1, memberId: 1 },
  { type: "loan_issued", loanId: 2, bookId: 2, memberId: 2 },
  { type: "return_received", loanId: 1, bookId: 1, memberId: 1 },
  { type: "loan_issued", loanId: 3, bookId: 3, memberId: 3 },
  { type: "reservation_opened", bookId: 1, memberId: 2 },
  { type: "return_received", loanId: 2, bookId: 2, memberId: 2 },
];

reset("events.log");
for (const e of EVENTS) append("events.log", e);

const NAMES = ["notification", "stock-summary", "report"];
for (const name of NAMES) {                      // the per-subscription-copy measure
  reset(`copy-${name}.log`);
  for (const e of EVENTS) append(`copy-${name}.log`, e);
}
const copyBytes = NAMES.reduce((t, a) => t + statSync(`copy-${a}.log`).size, 0);
console.log(`single log       : ${records("events.log").length} records  ${statSync("events.log").size} bytes`);
console.log(`subscription copy: ${NAMES.length} files  ${copyBytes} bytes`);

const offset = { notification: 0, "stock-summary": 0, report: 0 };
const rate = { notification: 6, "stock-summary": 4, report: 1 };   // records readable per round
for (const name of NAMES) offset[name] += read("events.log", offset[name], rate[name]).length;
console.log(`offsets after one round: ${NAMES.map((a) => `${a}=${offset[a]}`).join("  ")}`);
console.log(`log length after reading = ${records("events.log").length}`);
console.log(`records report is behind by = ${records("events.log").length - offset.report}`);

node three-consumers.mjs

single log       : 6 records  352 bytes
subscription copy: 3 files  1056 bytes
offsets after one round: notification=6  stock-summary=4  report=1
log length after reading = 6
records report is behind by = 5

Three numbers are decisive. Data held on disk: 352 bytes against 1056 bytes — the copy-holding model takes space directly proportional to the number of interested parties, while the log carries a single copy. The three consumers sit at different offsets — 6, 4, and 1 — meaning they are not waiting on each other; isolation comes from the offset, not the copy. And the most important part is the last two lines: all six records are still in the log after reading, and how far behind the report is — five records — is computed with a single subtraction. This gap is a lag metric that can be measured directly on a stream.

The log file holds one record per line, and its content is readable.

head -3 events.log

{"type":"loan_issued","loanId":1,"bookId":1,"memberId":1}
{"type":"loan_issued","loanId":2,"bookId":2,"memberId":2}
{"type":"return_received","loanId":1,"bookId":1,"memberId":1}

Replaying

This is where the two models part ways. The run below puts six loan-issued events first into a queue that deletes on acknowledgement, then into a log; in both, one consumer processes all of them, and afterward a second consumer tries to ask for the history.

// replay.mjs — the same workload first on a queue, then on a log; the history is replayed
import { reset, append, records, read } from "./log.mjs";

const EVENTS = [1, 2, 3, 4, 5, 6].map((n) => ({ type: "loan_issued", loanId: n }));

// Queue that deletes on acknowledgement: the essence of the behavior built in lesson 02
class Queue {
  constructor() { this.messages = []; }
  write(b) { this.messages.push(b); }
  receive() { return this.messages.length === 0 ? null : this.messages[0]; }
  acknowledge() { this.messages.shift(); }        // acknowledge: the message is deleted from the queue
  get depth() { return this.messages.length; }
}

const q = new Queue();
for (const e of EVENTS) q.write(e);
let first = 0;
for (let m = q.receive(); m !== null; m = q.receive()) { first += 1; q.acknowledge(); }
let second = 0;
for (let m = q.receive(); m !== null; m = q.receive()) { second += 1; q.acknowledge(); }
console.log(`queue: written=${EVENTS.length} first consumer=${first} ` +
            `depth=${q.depth} consumer joining later=${second}`);

reset("events.log");
for (const e of EVENTS) append("events.log", e);
let offset = 0;
for (const record of read("events.log", offset, 100)) offset += 1;
const next = read("events.log", 0, 100).length;      // second consumer, resetting its offset to zero
console.log(`stream: written=${EVENTS.length} first consumer=${offset} ` +
            `log length=${records("events.log").length} consumer joining later=${next}`);

// Can the same consumer fix its own bug and reprocess starting at the fourth record?
const rewound = read("events.log", 3, 100);
console.log(`stream: offset rewound to 3 -> records reread=${rewound.length} ` +
            `first loan=${rewound[0].loanId}`);

node replay.mjs

queue: written=6 first consumer=6 depth=0 consumer joining later=0
stream: written=6 first consumer=6 log length=6 consumer joining later=6
stream: offset rewound to 3 -> records reread=3 first loan=4

On the queue, the second consumer could read zero records. This is not a configuration gap: because acknowledgement deletes the message, there was nothing left to read. On the stream, the second consumer read all six records, and nothing in the log changed to make that happen.

The third line shows replay’s real use. If the report consumer notices it has been computing incorrectly starting at the fourth event, it only needs to rewind its offset to 3 with the fixed code; the log gives it the same records in the same order. The queue has no counterpart to this, because there is no offset to rewind — the only state a queue holds is which message is still there.

The cost sits in the same place. Because the offset lives with the consumer, saving it is also the consumer’s job. On a queue, the queue itself holds the “processed” information; on a stream, unless the consumer writes the offset to its own durable storage, it has no way to know where to resume when it restarts. When the offset gets written — before the work or after — is the subject of the next lesson.

Retention

A queue empties as it is consumed. A log does not empty; it only grows. This is why, on a stream, the decision to delete is tied not to consumption but to a rule: retention. The rule can be based on time, size, or record count; what they share is that none of them look at where the consumers are.

// retention.mjs — records whose retention has expired are trimmed; what a lagging consumer loses
import { writeFileSync } from "node:fs";
import { reset, append, records, read } from "./log.mjs";

let earliest = 0;                             // offset of the oldest record still in the log
function trim(file, keep) {                   // keeps only the newest `keep` records
  const all = records(file);
  const dropped = Math.max(0, all.length - keep);
  writeFileSync(file, all.slice(dropped).map((r) => JSON.stringify(r) + "\n").join(""));
  earliest += dropped;
  return dropped;
}
function readFromOffset(file, offset, count) {  // absolute offset, into the trimmed log
  if (offset < earliest) return null;           // the requested record no longer exists
  return read(file, offset - earliest, count);
}

reset("events.log");
for (let n = 1; n <= 10; n++) append("events.log", { type: "loan_issued", loanId: n });

const offset = { notification: 10, report: 2 };  // report is eight records behind
console.log(`before trim: log=${records("events.log").length} records  earliest offset=${earliest}`);
console.log(`  report offset=${offset.report} -> readable=${readFromOffset("events.log", offset.report, 100).length}`);

const dropped = trim("events.log", 4);
console.log(`after trim: dropped=${dropped}  log=${records("events.log").length} records  earliest offset=${earliest}`);
const reportRead = readFromOffset("events.log", offset.report, 100);
console.log(`  notification offset=${offset.notification} -> readable=${readFromOffset("events.log", offset.notification, 100).length}`);
console.log(`  report offset=${offset.report} -> ${reportRead === null ? "offset trimmed, record cannot be recovered" : reportRead.length}`);
console.log(`records report lost = ${earliest - offset.report}`);

node retention.mjs

before trim: log=10 records  earliest offset=0
  report offset=2 -> readable=8
after trim: dropped=6  log=4 records  earliest offset=6
  notification offset=10 -> readable=0
  report offset=2 -> offset trimmed, record cannot be recovered
records report lost = 4

Before trimming, report could read eight records. Once only the newest four records were kept, the log’s earliest offset became 6, and report’s offset was left behind that value; there is nothing left for it to read. The number of records it lost is the difference between the two offsets: four.

The notification consumer was not affected by the same trim at all, because its offset was at the end of the log and it had no new record to read anyway. The metric worth watching for the rule follows from this: the gap between the most-behind consumer’s offset and the log’s earliest offset. As that gap approaches zero, either retention gets extended or the lagging consumer gets sped up.

Which Model Solves What

The two models are not interchangeable; they answer different questions.

A queue is built for dividing work. Every message goes to one consumer once, work is shared among consumers, and processed work is deleted to free up space. Sending an overdue notification works this way: who sends it does not matter, and once sent, keeping the message has no value.

A stream is built for recording an event. Every consumer reads all the records starting from its own offset, more than one party can see the same record, and because the past is not deleted, it can be reread. The branch stock summary derived from loan events works this way: if the calculation changes, the summary is rebuilt from the log.

A single question is enough to draw the line: is reading the same record a second time meaningful? If the answer is no, use a queue; if yes, use a stream. This distinction is not free — a stream needs storage for retention, offset management, and a trim rule; a queue needs none of these.

Summary

• A stream is an ordered log where messages are only appended and reading does not delete a record; how far a consumer has reached lives not in the queue but in the consumer’s own offset value.
• Six events took 352 bytes in a single log, versus 1056 bytes across three files when copied per subscription; the three consumers advanced at offsets 6, 4, and 1 without waiting on each other.
• On the same workload, a consumer joining later could read zero records on the queue and six on the stream; a consumer that rewound its offset to 3 reread three records.
• Because the offset lives with the consumer, saving it durably is also the consumer’s job; on a queue, the queue itself holds that information.
• A log does not empty with consumption, it gets trimmed by a retention rule: once only the newest four records were kept, the earliest offset became 6, and the consumer at offset 2 was left unable to recover four records.

Next Step

In both models, one question was left unanswered: doing the work and writing the “done” information are two separate steps. On a queue, is the acknowledgement sent before the work or after? On a stream, is the offset recorded before the work or after? Ordering has two options, and both lose something — one counts work as processed before it is done, the other makes the same work happen twice. The next lesson runs the same workload in three modes: at-most-once, at-least-once, and deduplication with an idempotency key. It measures the loss count and the repeated-delivery count, and shows why “exactly-once” end to end is not a mode but a result.