Message Queues and Streams

The previous lesson’s bus had an unmeasured property: the publish call did not return until every attached handler finished. Four handlers ran in sequence on the same call stack, so the module that finalizes the fee took as long as the slowest handler. If archive writing slowed down, fee finalization slowed down too, even though the archive’s write time does not matter for pricing. This lesson puts a store in between: the publisher drops the message and returns, the consumer takes it later.

The message queue and the stream were built and measured in the Messaging topic of the Caching, Queues and Asynchronous Processing course: delivery semantics, the ordering/partitioning trade-off, dead-letter queues, and retry belong there and are not revisited here. What falls to this lesson is the style itself — what the store guarantees, what it does not, and the constraint this brings to the architecture. Four things are measured: consumer work run at the drop, pending messages, whether order is preserved, and the consequence of the same message processed twice.

The quality attribute it connects to is reliability, and the scenario question is: when the consumer falls behind the producer, or a message is delivered a second time, how much work piles up and how many delivery records get opened twice. A second measure falls into the performance efficiency family: how many consumer work units have run by the time the publisher drops the message.

Two Ends and a Store

The consumer is in the delivery operations context: it opens a delivery record for every message that arrives. Each message is assumed to cost three work units. This is a model, not a measurement; what is counted is the work unit count, no duration has been measured. The keyed option skips opening a record the second time it sees the same shipment code; it is measured at the end of the lesson.

mkdir -p async

// async/consumer.mjs — consumer: opens a delivery record for each message and spends 3 work units
export const WORK_UNITS = 3;

export function consumer(keyed = false) {
  const records = [], seen = new Set();
  let units = 0;
  return {
    process(message) {
      units += WORK_UNITS;
      if (keyed && seen.has(message.code)) return;
      seen.add(message.code);
      records.push(message.code);
    },
    units: () => units,
    records: () => records,
  };
}

The store exposes two operations. enqueue appends the message to the end and calls no consumer; dequeue gives the oldest message. The gap between these signatures is the whole of the style: enqueuer and dequeuer do not have to run at the same time.

// async/queue.mjs — message queue: the enqueuer does not wait, the dequeuer takes in order
export function queue() {
  const pending = [];
  let enqueued = 0, dequeued = 0, peak = 0;
  return {
    enqueue(message) { pending.push(message); enqueued += 1; peak = Math.max(peak, pending.length); },
    dequeue() { if (pending.length === 0) return null; dequeued += 1; return pending.shift(); },
    length: () => pending.length,
    counts: () => ({ enqueued, dequeued, peak }),
  };
}

The Publisher Does Not Wait

The first script passes twenty messages through both arrangements. In the synchronous arrangement the publisher calls the consumer directly; in the queue arrangement it drops the message. The consumer work run the moment the drop finishes is counted, the queue is then drained, and both arrangements are verified to produce the same record. In the second section, the producer processes five messages per round and the consumer three, and production stops after the fourth round.

// blocking.mjs — work run when the publisher drops the message, and messages pending in the queue
import { consumer, WORK_UNITS } from "./async/consumer.mjs";
import { queue } from "./async/queue.mjs";

const MESSAGES = Array.from({ length: 20 }, (_, i) => ({ code: `G-${String(i + 1).padStart(2, "0")}` }));

const sync = consumer();
for (const message of MESSAGES) sync.process(message);

const q = queue(), later = consumer();
for (const message of MESSAGES) q.enqueue(message);
const atDrop = { units: later.units(), pending: q.length() };
let dequeued;
while ((dequeued = q.dequeue()) !== null) later.process(dequeued);

const WIDTHS = [15, 25, 16, 18];
const ROWS = [
  ["synchronous", MESSAGES.length, sync.units(), 0, sync.units()],
  ["queue", MESSAGES.length, atDrop.units, atDrop.pending, later.units()],
];
console.log("arrangement    publisher calls  work run at drop  pending messages  work run at end");
for (const [label, ...s] of ROWS) console.log(label.padEnd(13) + s.map((v, i) => String(v).padStart(WIDTHS[i])).join(""));
console.log(`queue counts: ${JSON.stringify(q.counts())}, records equal = ${JSON.stringify(sync.records()) === JSON.stringify(later.records())}`);

const PRODUCTION = [5, 5, 5, 5, 0, 0, 0, 0];
const CONSUMPTION = 3;
const queue2 = queue(), consumer2 = consumer();
let sequence = 0;
console.log("\nround  produced  consumed  pending");
for (const [i, produced] of PRODUCTION.entries()) {
  for (let n = 0; n < produced; n += 1) { sequence += 1; queue2.enqueue({ code: `H-${sequence}` }); }
  let consumed = 0;
  for (let n = 0; n < CONSUMPTION; n += 1) {
    const message = queue2.dequeue();
    if (message === null) break;
    consumer2.process(message);
    consumed += 1;
  }
  console.log(`${String(i + 1).padStart(3)}${String(produced).padStart(10)}${String(consumed).padStart(11)}${String(queue2.length()).padStart(10)}`);
}
console.log(`peak pending = ${queue2.counts().peak}, work units = ${consumer2.units()}, work units per message = ${WORK_UNITS}`);

node blocking.mjs

arrangement    publisher calls  work run at drop  pending messages  work run at end
synchronous               20                       60               0                60
queue                     20                        0              20                60
queue counts: {"enqueued":20,"dequeued":20,"peak":20}, records equal = true

round  produced  consumed  pending
  1         5          3         2
  2         5          3         4
  3         5          3         6
  4         5          3         8
  5         0          3         5
  6         0          3         2
  7         0          2         0
  8         0          0         0
peak pending = 11, work units = 60, work units per message = 3

Reading the Numbers

The first table’s last column is 60 in both arrangements: no work vanished, the same work got done, and both produced the same record. What changes is when the work happens. In the synchronous arrangement, sixty work units had already spent the publisher’s own time by its twentieth call; in the queue arrangement, zero work units had run at the same moment, and twenty messages waited in the store. The pending-message count is the cost’s name: the publisher escaped waiting, but unprocessed work now sits in the system and has to be held somewhere.

The second table shows how this backlog behaves. While production runs faster than consumption, the pending count grows by two every round; after the fourth round’s production, the queue reaches its peak, eleven. Once production stops, the queue drains and empties three rounds later. This is the style’s core constraint: the queue stores the speed gap, it does not eliminate it. If production stays permanently faster than consumption, the pending count grows without bound; the queue only buys the publisher freedom from waiting while the gap is temporary.

Order and Retry

The second script asks two questions. The first is order: is the enqueue order the same as the processing order? It is measured separately for a single consumer and for two consumers working off the same queue’s end. In the two-consumer arrangement, the workers’ costs differ, and whichever worker frees up first takes the next message; finish order follows from that. This is a model, not a measurement — what is counted is the work unit and the finish order, no duration has been measured. The second question is retry: what happens if an unacknowledged message goes back onto the queue?

// order-and-retry.mjs — order with one and two consumers, delivery records opened when a message is redelivered
import { consumer } from "./async/consumer.mjs";
import { queue } from "./async/queue.mjs";

const MESSAGES = Array.from({ length: 8 }, (_, i) => ({ code: `G-${i + 1}` }));
const ORDER = MESSAGES.map((m) => m.code);

const q = queue(), single = consumer();
for (const message of MESSAGES) q.enqueue(message);
let dequeued;
while ((dequeued = q.dequeue()) !== null) single.process(dequeued);

const inversions = (sequence) => {
  const position = new Map(ORDER.map((code, i) => [code, i]));
  let count = 0;
  for (let a = 0; a < sequence.length; a += 1)
    for (let b = a + 1; b < sequence.length; b += 1) if (position.get(sequence[a]) > position.get(sequence[b])) count += 1;
  return count;
};

const COST = [1, 4];
const freeAt = COST.map(() => 0);
const finish = [];
const queue2 = queue();
for (const message of MESSAGES) queue2.enqueue(message);
while ((dequeued = queue2.dequeue()) !== null) {
  const j = freeAt[0] <= freeAt[1] ? 0 : 1;
  freeAt[j] += COST[j];
  finish.push([dequeued.code, freeAt[j]]);
}
const finishOrder = [...finish].sort((a, b) => a[1] - b[1]).map(([code]) => code);

console.log(`enqueue order   : ${ORDER.join(" ")}`);
console.log(`single consumer : ${single.records().join(" ")}  inversions = ${inversions(single.records())}`);
console.log(`two consumers   : ${finishOrder.join(" ")}  inversions = ${inversions(finishOrder)}`);

for (const [label, keyed] of [["unkeyed", false], ["keyed", true]]) {
  const queue3 = queue(), consumer3 = consumer(keyed);
  for (const message of MESSAGES) queue3.enqueue(message);
  let count = 0;
  while ((dequeued = queue3.dequeue()) !== null) {
    consumer3.process(dequeued);
    count += 1;
    if (count === 3) queue3.enqueue(dequeued);
  }
  const counts = queue3.counts();
  console.log(`${label.padEnd(11)} delivered = ${counts.dequeued}, distinct messages = ${new Set(MESSAGES.map((m) => m.code)).size}, opened delivery records = ${consumer3.records().length}, work units = ${consumer3.units()}`);
}

node order-and-retry.mjs

enqueue order   : G-1 G-2 G-3 G-4 G-5 G-6 G-7 G-8
single consumer : G-1 G-2 G-3 G-4 G-5 G-6 G-7 G-8  inversions = 0
two consumers   : G-1 G-3 G-4 G-2 G-5 G-6 G-8 G-7  inversions = 3
unkeyed     delivered = 9, distinct messages = 8, opened delivery records = 9, work units = 27
keyed       delivered = 9, distinct messages = 8, opened delivery records = 8, work units = 27

For a single consumer, the inversion count is zero: order is preserved, a consequence of the store’s dequeue signature. When a second consumer is added, the same queue hands out messages in the same order, but the finish order broke in three places. Order preservation is tied to having a single consumer; adding a second is not a performance decision, it is a decision to give up the order guarantee. The architectural consequence: work whose order matters and work that can be processed in parallel cannot go through the same queue.

The last two lines are retry’s bill. Eight distinct messages were delivered nine times. In the unkeyed consumer, opened delivery records numbered nine: two for the same shipment. In the keyed consumer it stayed at eight — the second delivery had no effect — but work units are 27 in both, since the message was taken and processed again regardless. This gives the style’s constraint: the store puts the message in order, it does not guarantee uniqueness; that has to come from the consumer’s own key. How a two-consumer design deviates against the same message arriving twice is measured separately in this course’s Event-Driven Architecture lesson.

Consuming versus Deleting

The queue’s dequeue operation removes the message from the store: consuming is deleting. The stream separates these operations — the record stays in place, and each consumer keeps its own offset. Offset, retention, and rereading were measured in the Streams versus Queues lesson of the Messaging topic. The question here is architectural: can a unit added afterward build its own history alone?

// async/log.mjs — stream: consuming does not delete, each consumer keeps its own offset
export function log() {
  const entries = [];
  return { append: (message) => entries.push(message), read: (offset) => entries.slice(offset), length: () => entries.length };
}

// history.mjs — queue empties once consumed, stream is read by offset: history for a unit added afterward
import { queue } from "./async/queue.mjs";
import { log } from "./async/log.mjs";
import { consumer } from "./async/consumer.mjs";

const MESSAGES = Array.from({ length: 20 }, (_, i) => ({ code: `G-${i + 1}` }));

const q = queue(), queueConsumer = consumer();
for (const message of MESSAGES) q.enqueue(message);
let dequeued;
while ((dequeued = q.dequeue()) !== null) queueConsumer.process(dequeued);

const stream = log(), streamConsumer = consumer();
for (const message of MESSAGES) stream.append(message);
let offset = 0;
for (const message of stream.read(offset)) { streamConsumer.process(message); offset += 1; }
const initialOffset = offset;

const newcomer = consumer();
for (const message of stream.read(0)) newcomer.process(message);

console.log(`queue: enqueued ${q.counts().enqueued}, pending after being consumed ${q.length()}`);
console.log(`  history a newcomer can read from the queue on its own = ${q.length()} records`);
console.log(`  the history is only obtained by exposing the consumer's records: ${queueConsumer.records().length} records, bindings created to another consumer = 1`);
console.log(`stream: log length ${stream.length()}, first consumer's offset ${initialOffset}`);
console.log(`  history a newcomer reading from offset 0 can read = ${newcomer.records().length} records, bindings created to another consumer = 0`);
console.log(`  first consumer's offset ${initialOffset} -> ${offset}, log length ${stream.length()}`);

node history.mjs

queue: enqueued 20, pending after being consumed 0
  history a newcomer can read from the queue on its own = 0 records
  the history is only obtained by exposing the consumer's records: 20 records, bindings created to another consumer = 1
stream: log length 20, first consumer's offset 20
  history a newcomer reading from offset 0 can read = 20 records, bindings created to another consumer = 0
  first consumer's offset 20 -> 20, log length 20

Once the queue has emptied, the history a newly added unit can read on its own is zero records. The only way to get the same information is for the existing consumer to expose its record — building a new binding between two consumers: the new unit now depends not on the publisher but on another consumer. In the stream the same number is twenty, and the binding created is zero; the new unit knows only the log. The second read left the first consumer’s offset unchanged, and the log’s length did not grow — consumer count does not grow the knowledge obligation.

This is the choice criterion: if the work happens once and is done, a queue is enough; if units added afterward need to build their own state from history, a stream is needed — its cost is keeping the record around.

Summary

• The store separates the publisher from the consumer: a publisher dropping twenty messages spent 60 work units of its own time synchronously, spent 0 through the queue, and left 20 messages waiting in the store.
• Total work is 60 in both arrangements and both produced the same record; the queue changes when work happens, not how much.
• The queue stores the speed gap: with 5 produced and 3 consumed per round, the pending peak was 11, and the queue emptied three rounds after production stopped.
• Order was preserved with a single consumer (0 inversions); adding a second consumer broke the finish order in three places — a decision to give up the order guarantee.
• When eight messages were delivered nine times, the unkeyed consumer opened 9 delivery records and the keyed consumer 8; work units were 27 in both — the store guarantees order, not uniqueness.
• In a queue, consuming is deleting: a unit added afterward can read 0 history records, versus 20 in a stream, which creates no binding to another consumer.

Next Step

This topic covered how units talk to each other. The same three things were measured across six styles: a unit’s obligation to know the other side, the direction of connection, and the data crossing the boundary. Client and server, peers, model and view, pipe and filter, publisher and subscriber, the two ends of a queue — all compared on the same three numbers. But these measurements passed by one question: how many separate parts are these units deployed as? This lesson’s queue, and the previous lesson’s bus, both lived inside a single running program; the publish–subscribe arrangement needed no second process, one object was enough. Interaction style and part count are two independent decisions. The next topic asks the second one: how many separate deployment units does the software get published as, and what does that number change.