Publish–Subscribe

The previous lesson gathered what the queue does not give into three points; the last was broadcast. A message goes to exactly one consumer, because a received message becomes invisible to the others. But the loan-issued event concerns three separate parties: the one that sends the overdue notification to the member, the one that recalculates the branch stock summary, and the one that adds a line to the monthly report. The same event needs to reach all three.

This lesson’s question is where that “all three” gets written to. The answer is not just a distribution mechanism, but a separation that keeps the producer from knowing who is listening.

The Producer’s Three-Destination Problem

The shortest fix is for the producer to write three copies to three queues. It has a measurable cost: the producer now knows the names of three destinations. When a fourth interested party — say, the search index — is added, the loan-issuing code has to change. When a fifth is added, it changes again. The queue separated producer and consumer in time, but the producer still carries the list of consumers.

This is the coupling problem domain events solve in the Business Logic Placement topic: the publishing side does not count listeners, it only announces what happened. Publish–subscribe carries the same separation across the process boundary. The producer writes to a single topic; the topic knows who reads it.

Topic and Subscription

The separation has two names, and they must not be confused. A topic is the named channel that messages are written to. A subscription is an interested party’s own delivery queue on that topic. Every published message is dropped as one copy into every subscription that exists at the moment of publication; subscriptions do not affect each other’s progress.

The model below holds both of these. Time is read not from the real clock but from a budget spent explicitly for each subscription, so the measurements stay independent of the machine.

// topic.mjs — topic and subscriptions: a published message drops one copy into every subscription
export class Topic {
  constructor() {
    this.subscriptions = new Map();  // name -> {queue, duration, filter, processed, filteredOut}
    this.published = 0;              // number of messages published
    this.copies = 0;                 // number of copies dropped into subscriptions
    this.delivered = 0;              // number of copies processed by subscribers
  }
  subscribe(name, { duration = 1, filter = () => true } = {}) {
    this.subscriptions.set(name, { name, queue: [], duration, filter, processed: 0, filteredOut: 0 });
  }
  publish(type, body) {                        // producer side: a single call
    const message = { id: ++this.published, type, body };
    for (const s of this.subscriptions.values()) {
      if (!s.filter(type)) { s.filteredOut += 1; continue; }
      s.queue.push(message);
      this.copies += 1;
    }
    return message.id;
  }
  advance(budget) {                            // each subscription advances at its own rate
    for (const s of this.subscriptions.values()) {
      let remaining = budget;
      while (s.queue.length > 0 && remaining >= s.duration) {
        s.queue.shift(); s.processed += 1; this.delivered += 1; remaining -= s.duration;
      }
    }
  }
  pending(name) { return this.subscriptions.get(name).queue.length; }
  processed(name) { return this.subscriptions.get(name).processed; }
}

The producer’s number of calls and the system’s number of deliveries are two separate numbers. The run below raises the subscriber count from one to four while never changing the producer side at all.

// single-publish.mjs — how many deliveries a single publish call turns into by subscriber count
import { Topic } from "./topic.mjs";

for (const subscribers of [["notification"], ["notification", "stock-summary"],
                           ["notification", "stock-summary", "report"],
                           ["notification", "stock-summary", "report", "search-index"]]) {
  const t = new Topic();
  for (const name of subscribers) t.subscribe(name);
  for (const n of [1, 2, 3, 4]) t.publish("loan_issued", { loanId: n });
  t.advance(1000);
  console.log(`subscribers=${subscribers.length}  publish calls=${t.published}  ` +
              `copies dropped=${t.copies}  delivered=${t.delivered}`);
}

node single-publish.mjs

subscribers=1  publish calls=4  copies dropped=4  delivered=4
subscribers=2  publish calls=4  copies dropped=8  delivered=8
subscribers=3  publish calls=4  copies dropped=12  delivered=12
subscribers=4  publish calls=4  copies dropped=16  delivered=16

In all four rows, the publish call count is four; the number of copies dropped is multiplied by the subscriber count. That multiplication is both the power and the cost of the publish–subscribe model. The power: the producer still writes a single line even with four subscribers, and adding an interested party never touches the producer. The cost: the work the system carries grows as publish count times subscriber count. The fan-out multiplier determines the cost of carrying the message, not of producing it.

A subscription having its own delivery queue does not invalidate the competing-consumer arrangement from the previous lesson; it combines with it. Multiple workers can run inside a single subscription and share the copies that land in it among themselves. The distinction between the two arrangements fits in one sentence: there is copying between subscriptions, and sharing within a subscription.

Does a Slow Subscriber Hold Up the Others

This is the real test of a separate delivery queue. Let the monthly-report subscriber run slower than the others: let it take three time units to process each message, one for the others. Four events are published every round, and each subscription can spend six time units per round.

// slow-subscriber.mjs — the report subscriber is three times slower; backlog is measured per subscription
import { Topic } from "./topic.mjs";
const t = new Topic();
t.subscribe("notification", { duration: 1 });
t.subscribe("stock-summary", { duration: 1 });
t.subscribe("report", { duration: 3 });

const row = (...v) => v.map((x) => String(x).padStart(14)).join("");
console.log(row("round", "notification", "stock-summary", "report"), "  (processed/pending)");
for (let round = 1; round <= 6; round++) {
  for (let i = 0; i < 4; i++) t.publish("loan_issued", { round, i });
  t.advance(6);                       // each subscription can spend 6 time units per round
  const g = (name) => `${t.processed(name)}/${t.pending(name)}`;
  console.log(row(round, g("notification"), g("stock-summary"), g("report")));
}

node slow-subscriber.mjs

         round  notification stock-summary        report   (processed/pending)
             1           4/0           4/0           2/2
             2           8/0           8/0           4/4
             3          12/0          12/0           6/6
             4          16/0          16/0           8/8
             5          20/0          20/0         10/10
             6          24/0          24/0         12/12

By the end of six rounds, notification and stock-summary have each finished twenty-four messages, with none pending. The report subscriber is stuck at twelve messages and has backlogged twelve. The backlog sits only in the slow subscription; the fast ones’ progress never slowed at all.

What provides this isolation is a separate queue per subscription. If the three subscribers shared a single queue, the slowest one would sit in front of the others and set the pace for the whole topic. Separate queues prevent that, but they do not eliminate the problem: a slow subscriber’s backlog occupies real space in memory or on disk, and its growth is unbounded. The previous lesson’s queue-depth metric is tracked here per subscription; which subscriber is falling behind cannot be seen by looking at the total count.

Not Every Subscriber Wants Every Message

The loan-issued event is not the only thing written to the topic. Returns, member registrations, and branches being added pass through the same channel too. The notification subscriber has no interest in member registrations, and the report subscriber only writes loan-issued lines. Receiving and discarding a message it has no interest in means wasting the write and read cost of every copy dropped into the subscription.

A subscription is given a filter: a condition that decides, by looking at the message’s type, whether the copy gets dropped in at all. It is decisive that the filter runs on the publish side; a filtered-out message is never written to the subscription in the first place.

// filter.mjs — a subscription receives only the types it cares about; filtered messages are not copied
import { Topic } from "./topic.mjs";
const TYPES = ["loan_issued", "return_received", "member_registered", "loan_issued",
               "branch_added", "return_received", "member_registered", "loan_issued"];

function run(filtered) {
  const t = new Topic();
  const f = (set) => filtered ? (type) => set.has(type) : () => true;
  t.subscribe("notification", { filter: f(new Set(["loan_issued", "return_received"])) });
  t.subscribe("stock-summary", { filter: f(new Set(["loan_issued", "return_received", "branch_added"])) });
  t.subscribe("report", { filter: f(new Set(["loan_issued"])) });
  for (const type of TYPES) t.publish(type, {});
  t.advance(1000);
  return t;
}

for (const filtered of [false, true]) {
  const t = run(filtered);
  console.log(`filter=${filtered ? "on" : "off"}  published=${t.published}  copies dropped=${t.copies}  ` +
              `notification=${t.processed("notification")} stock-summary=${t.processed("stock-summary")} ` +
              `report=${t.processed("report")}`);
}

node filter.mjs

filter=off  published=8  copies dropped=24  notification=8 stock-summary=8 report=8
filter=on  published=8  copies dropped=14  notification=5 stock-summary=6 report=3

Eight publishes produce twenty-four copies without a filter; fourteen with one. The report subscriber processes three messages instead of eight, and never even sees the five it would have discarded. The filter has a hidden benefit too: what a subscription wants lives in the subscription’s definition, not inside the code. Which events the notification subscriber cares about can be seen without reading the consumer code.

The filter’s limit depends on what is in the message’s header at the time it decides. A filter that looks at the type is cheap. A filter that looks at a field inside the body — “only loans from the Central branch” — requires the message to be parsed, and its cost is multiplied by the number of subscriptions. This is why the split criterion is usually placed not on the message body, but on the type or topic given at publish time.

The Limit of Publish–Subscribe: There Is No History

The model’s most decisive property shows up in a measurement. A copy is dropped only into the subscriptions that exist at the moment of publication. For a subscription that does not exist at that moment, the message was never written at all.

// late-subscriber.mjs — if a subscription is created after publication, earlier messages never reach it
import { Topic } from "./topic.mjs";
const t = new Topic();
t.subscribe("notification");
t.subscribe("stock-summary");

for (const n of [1, 2, 3]) t.publish("loan_issued", { loanId: n });
t.advance(1000);
console.log(`after three publishes : notification=${t.processed("notification")} stock-summary=${t.processed("stock-summary")}`);

t.subscribe("search-index");          // new interested party joins later
for (const n of [4, 5]) t.publish("loan_issued", { loanId: n });
t.advance(1000);
console.log(`after five publishes: notification=${t.processed("notification")} stock-summary=${t.processed("stock-summary")} ` +
            `search-index=${t.processed("search-index")}`);
console.log(`total published=${t.published}  missed by search-index=${t.published - t.processed("search-index")}`);

node late-subscriber.mjs

after three publishes : notification=3 stock-summary=3
after five publishes: notification=5 stock-summary=5 search-index=2
total published=5  missed by search-index=3

The search index saw two of the five loan-issued events. The three it missed are not waiting somewhere; those copies never existed at all. The same gap shows up in a second form: a processed copy is removed from the subscription’s queue, so if the search index loses its own data to a failure, it cannot reread even the messages it already saw.

This is not a flaw in the model, it is its definition. Publish–subscribe distributes the message, it does not store it. A subscriber that wants to keep it has to accumulate what it receives on its own side. If a new interested party needs to build itself up by looking at the past, or a subscriber needs to go back and reread, what is needed is not distribution — it is a durable record.

Summary

• A topic is the named channel that messages are written to; a subscription is an interested party’s own delivery queue on that topic. There is copying between subscriptions, sharing within a subscription.
• Four publish calls turned into sixteen copies instead of four once subscriber count rose from one to four; the producer’s code never changed. The fan-out multiplier determines the carrying cost.
• Separate delivery queues isolate a slow subscriber: by the end of six rounds the report subscriber had backlogged twelve messages while notification and stock-summary had each finished twenty-four and were left with zero pending.
• The filter runs on the publish side: eight publishes produced twenty-four copies without a filter and fourteen with one, and the report subscriber never even saw the five messages it had no interest in.
• A copy is dropped only into subscriptions that exist at the moment of publication; the search index, joining later, never saw three of five events. The model distributes the message, it does not store it.

Next Step

Both the queue and the topic share the same assumption: a delivered message has been consumed, and a consumed message gets deleted. The cost of that assumption showed up in the final measurement — a party that connects later cannot see the past, and after a failure no consumer can go back. The next lesson builds the second transport model, the one that separates consuming from deleting: a durable, ordered log where messages are only appended and stay in place once read. The same workload is run on both the queue and the log; a number shows why replaying is impossible on one and as cheap as rolling back an offset value on the other.