Message Queues

The previous lesson wrote the side jobs to the outbox in the same transaction as the loan record and showed that all nine jobs were still there after the process restarted. The questions we left open were these: who reads these rows, what happens if two readers take the same row at the same time, and if a reader takes a row and crashes before processing it, how does that work come back?

All three are answered by a single structure. A message queue is a durable list that one side writes to and another side reads from, where the reader decides when what it read gets deleted. This lesson builds the queue in its smallest form and counts how many times each message gets delivered.

The Queue’s Four Operations

The writing side is called the producer; the reading side is called the consumer. The contract between them consists of four operations. The producer drops a message into the queue with write. The consumer asks for a message with receive. When the work is done, it tells the queue to delete the message with acknowledge. The fourth is not an operation but a duration: a received message is not given to another consumer until it is acknowledged, but this protection does not last forever — once the visibility timeout expires, the message becomes visible again.

The model below holds all four. Time is read not from the real clock but from a virtual clock that is advanced explicitly, so the measurements stay independent of the machine.

// queue.mjs — the smallest queue model, with acknowledgement and a visibility timeout
export class Queue {
  constructor(visibility = 30) {
    this.visibility = visibility;     // how many ms a received message is withheld from others
    this.messages = [];               // {id, body, visibleAt, delivered}
    this.nextId = 1;
    this.now = 0;                     // virtual clock: independent of real time
    this.totalDelivered = 0;
  }
  write(body) {                       // producer side
    this.messages.push({ id: this.nextId++, body, visibleAt: 0, delivered: 0 });
    return this.messages.at(-1).id;
  }
  receive() {                         // consumer side
    const m = this.messages.find((m) => m.visibleAt <= this.now);
    if (m === undefined) return null;
    m.visibleAt = this.now + this.visibility;
    m.delivered += 1;
    this.totalDelivered += 1;
    return { id: m.id, body: m.body, delivered: m.delivered };
  }
  acknowledge(id) {                   // work is done: the message is deleted from the queue
    const before = this.messages.length;
    this.messages = this.messages.filter((m) => m.id !== id);
    return before !== this.messages.length;
  }
  advance(ms) { this.now += ms; }
  get depth() { return this.messages.length; }
  get visible() { return this.messages.filter((m) => m.visibleAt <= this.now).length; }
}

In the expected case, the producer writes five jobs and a single consumer receives and acknowledges all of them.

// happy-path.mjs — five jobs are written, a single consumer receives and acknowledges all of them
import { Queue } from "./queue.mjs";
const q = new Queue();
const JOBS = [
  "overdue_notification loan=1", "overdue_notification loan=2", "stock_summary branch=1",
  "reservation_advance book=3", "report_line loan=1",
];
for (const job of JOBS) q.write(job);
console.log(`produced=${JOBS.length}  queue depth=${q.depth}`);

let processed = 0;
for (let message = q.receive(); message !== null; message = q.receive()) {
  processed += 1;
  q.acknowledge(message.id);
}
console.log(`processed=${processed}  total delivered=${q.totalDelivered}  remaining depth=${q.depth}`);

node happy-path.mjs

produced=5  queue depth=5
processed=5  total delivered=5  remaining depth=0

Five jobs, five deliveries, zero remaining. There is no direct call between producer and consumer: the producer does not know whether the consumer is running, the consumer does not know who produced the job. The queue separates the two in time — a second kind of separation layered on top of the loose coupling in domain events.

Why Acknowledgement Is a Separate Step

receive does not delete the message. If it did, a consumer crashing after taking the job would mean the job is lost. Instead the message is made temporarily invisible, and the decision to delete it is left to acknowledgement. The measurement below runs the case where a consumer terminates without acknowledging its second job.

// unacknowledged.mjs — an unacknowledged message is redelivered once the visibility timeout expires
import { Queue } from "./queue.mjs";
const q = new Queue(30);
for (const job of ["overdue_notification loan=1", "stock_summary branch=1", "report_line loan=1"])
  q.write(job);

// First consumer: it terminates while receiving the second job, without acknowledging it.
const willCrash = new Set(["stock_summary branch=1"]);
for (let message = q.receive(); message !== null; message = q.receive()) {
  const status = willCrash.has(message.body) ? "left unacknowledged" : "acknowledged";
  if (status === "acknowledged") q.acknowledge(message.id);
  console.log(`t=${q.now} consumer-1 message=${message.id} delivered=${message.delivered} ${status}`);
}
console.log(`t=${q.now} depth=${q.depth} visible=${q.visible}`);

q.advance(30);                       // visibility timeout expired
console.log(`t=${q.now} depth=${q.depth} visible=${q.visible}`);

// Second consumer finds the same job.
for (let message = q.receive(); message !== null; message = q.receive()) {
  q.acknowledge(message.id);
  console.log(`t=${q.now} consumer-2 message=${message.id} delivered=${message.delivered} acknowledged`);
}
console.log(`total delivered=${q.totalDelivered}  messages written=3  remaining depth=${q.depth}`);

node unacknowledged.mjs

t=0 consumer-1 message=1 delivered=1 acknowledged
t=0 consumer-1 message=2 delivered=1 left unacknowledged
t=0 consumer-1 message=3 delivered=1 acknowledged
t=0 depth=1 visible=0
t=30 depth=1 visible=1
t=30 consumer-2 message=2 delivered=2 acknowledged
total delivered=4  messages written=3  remaining depth=0

Three messages were written, four deliveries were made. The second message was delivered twice because its first delivery was left unacknowledged. The measurement’s most important lines are the two in the middle: the unacknowledged message was still sitting in the queue (depth=1) but was not visible (visible=0); once the timeout expired, it became visible again. The visibility timeout is the expiration date on the assumption that “this job is currently being done by someone else.”

Choosing the duration is a trade-off. If it is set shorter than the job’s longest possible duration, the message gets handed to a second consumer while the job is still running, and the same job gets done twice. If it is set too long, work left behind by a consumer that really did crash takes a long time to reach anyone. The right fix for long-running jobs is not to lengthen the timeout, but to have the consumer extend it while it is still working.

From this comes the queue’s basic guarantee: unacknowledged work is not lost, but not losing it comes at the price of it possibly being repeated. The delivery count is not the number of times a message was processed; it is only the number of times the queue handed it out.

Competing Consumers

The queue’s second basic property is that multiple consumers reading from the same queue share the work. This arrangement is called competing consumers: every message goes to exactly one consumer, because a received message becomes invisible to the others.

// competing-consumers.mjs — three consumers read from the same queue; each message goes to one consumer
import { Queue } from "./queue.mjs";
const q = new Queue();
for (let i = 1; i <= 12; i++) q.write(`overdue_notification loan=${i}`);

const takenBy = new Map();                    // message -> the consumer that handled it
const count = { "consumer-1": 0, "consumer-2": 0, "consumer-3": 0 };
const names = Object.keys(count);
for (let turn = 0; q.depth > 0; turn++) {
  const name = names[turn % names.length];
  const message = q.receive();
  if (message === null) break;
  takenBy.set(message.id, name);
  count[name] += 1;
  q.acknowledge(message.id);
}
console.log(`distribution: ${Object.entries(count).map(([a, n]) => `${a}=${n}`).join("  ")}`);
console.log(`distinct messages=${takenBy.size}  total delivered=${q.totalDelivered}  remaining=${q.depth}`);

node competing-consumers.mjs

distribution: consumer-1=4  consumer-2=4  consumer-3=4
distinct messages=12  total delivered=12  remaining=0

Twelve jobs were split four each across three consumers. The distinct message count matching the total delivery count says that no message went to more than one consumer. Because the overdue-notification job can be split three ways, adding a consumer is directly adding capacity; the loan request itself has no such divisibility.

Queue Depth

The number of messages waiting in the queue is called queue depth, and it measures exactly one thing: the difference between production rate and consumption rate. The run below produces five jobs per round; each consumer finishes three jobs per round.

// depth.mjs — queue depth grows once the production rate exceeds the consumption rate
import { Queue } from "./queue.mjs";

function run(consumerCount, rounds = 6, production = 5, consumptionRate = 3) {
  const q = new Queue();
  const depths = [];
  for (let t = 1; t <= rounds; t++) {
    for (let i = 0; i < production; i++) q.write(`job t=${t} n=${i}`);
    for (let i = 0; i < consumerCount * consumptionRate; i++) {
      const message = q.receive();
      if (message === null) break;
      q.acknowledge(message.id);
    }
    q.advance(1000);
    depths.push(q.depth);
  }
  return depths;
}

for (const n of [1, 2, 3]) {
  const d = run(n);
  console.log(`consumers=${n}  end-of-round depths: ${d.join(" ")}`);
}

node depth.mjs

consumers=1  end-of-round depths: 2 4 6 8 10 12
consumers=2  end-of-round depths: 0 0 0 0 0 0
consumers=3  end-of-round depths: 0 0 0 0 0 0

With a single consumer, depth grows by two every round; the queue is not swallowing the work, only postponing it. With two consumers, capacity rises to six and depth stays at zero; a third consumer adds nothing. A queue is not a rate regulator; it smooths out a burst, it does not close a sustained gap. Depth rising over time is a direct signal of a consumer shortage, and it is the first metric worth watching.

What the Queue Gives and What It Does Not

The queue gives three things: durability of the message, divisibility of the work, and separation of the parties in time. What it does not give is equally decisive.

It does not give single delivery. As we saw in the measurement, an unacknowledged message is redelivered; making processing idempotent under repetition is the consumer’s responsibility. The processed_message guard built in the Idempotent Transactions lesson exists exactly for this condition.

It does not give ordering. The moment competing consumers share the work, two messages’ finish order can differ from their write order. If the same book’s two events land on two different consumers, the second one can finish first.

It does not give broadcast. A message goes to exactly one consumer. If the same event needs to be seen by the notification service, the report, and the search index all at once, the queue by itself is not enough.

Summary

• The queue is defined by four operations: the producer writing, the consumer receiving, acknowledging once the work is done, and a received message not being given to anyone else during the visibility timeout.
• An unacknowledged message stays in the queue but becomes invisible; once the timeout expired it became visible again, and a total of four deliveries were measured for the three messages written.
• If the visibility timeout is set shorter than the job’s longest duration, the same job gets done twice; if set too long, a crashed consumer’s work waits a long time.
• With competing consumers, every message goes to a single side: twelve jobs split four each across three consumers, and the distinct message count matched the total delivery count.
• Queue depth measures the difference between production and consumption rate; with a single consumer it grew by two every round, it stayed at zero once a second consumer was added, and a third one changed nothing.

Next Step

The last of the things the queue does not give is the subject of the next lesson. The loan-issued event concerns not just the notification service, but also the branch stock summary and the monthly report. Writing three copies to three separate queues to reach all three parties ties the producer back to the list of consumers — exactly what we wanted to avoid. The next lesson builds the publish–subscribe model that breaks that tie, counts how many deliveries a single message turns into depending on the number of subscribers, and measures whether a slow subscriber affects the others.