Pub/Sub, From First Principles: Part 2 — Asynchronous Workers

Part 2 of 4 in the Pub/Sub, From First Principles series.

The fix for both edge cases from Part 1 is the same: stop calling subscribers directly from inside publish(). Enqueue the message instead, return immediately, and let each subscriber process on its own schedule. Each subscriber gets its own queue and its own dedicated worker that drains it.

// pubsub.ts — non-blocking publish, per-subscriber failure isolation
type Subscriber = (message: unknown) => void;

interface SubscriberWorker {
  subscriber: Subscriber;
  queue: unknown[];
  processing: boolean;
}

class PubSub {
  private subscribers: SubscriberWorker[] = [];

  subscribe(subscriber: Subscriber): void {
    this.subscribers.push({ subscriber, queue: [], processing: false });
  }

  publish(message: unknown): void {
    for (const worker of this.subscribers) {
      worker.queue.push(message);
      if (!worker.processing) this.processQueue(worker);
    }
  }

  private async processQueue(worker: SubscriberWorker): Promise<void> {
    worker.processing = true;
    while (worker.queue.length > 0) {
      const message = worker.queue.shift();
      try {
        await Promise.resolve(worker.subscriber(message));
      } catch (err) {
        console.error("Subscriber failed:", err);
      }
    }
    worker.processing = false;
  }
}

publish() now just pushes and returns — it never waits on a subscriber, so a slow one only ever blocks itself. And each subscriber's try/catch means a thrown error is isolated to that subscriber; it no longer takes any other subscriber down with it.

                  ┌──> [queue: ▓▓▓░░] ──> [worker A] ──> handler A
[publish()] ──────┼──> [queue: ▓░░░░] ──> [worker B] ──> handler B
 returns          └──> [queue: ▓▓▓▓▓] ──> [worker C] ──> handler C (slow —
 immediately                                              only its own queue grows)

This queue-per-consumer shape is exactly what managed systems give you at infrastructure scale: an SQS queue per consuming service, a BullMQ queue per job type, a RabbitMQ queue bound per consumer — same isolation idea, with the queue living outside your process instead of in an array.

Ordering: giving each subscriber its own worker isn't just about isolation — it's also what keeps messages in order. The while loop only dequeues the next message after await-ing the current one, so message n can't start before message n-1 finishes, no matter how fast either one actually is. Publish three messages with wildly different processing times — the slowest first, the fastest second — and they still come out in the order they were published, because the queue never starts message two before message one is done.

Edge case: memory limits. Nothing bounds worker.queue here. A subscriber that's chronically slower than the rate messages arrive just accumulates an ever-growing array, with no signal back to the publisher that anything's wrong.

Edge case: a worker crashing mid-job. If the process dies while a message is being processed — after the real work happened, but before anything recorded that it happened — that message's fate becomes ambiguous. Was it done? Partially done? On restart, is it retried and processed twice, or is it just gone?

Next up: every subscriber still sees every message, whether it's relevant to it or not. How do you route messages to the subscribers that actually care, instead of broadcasting everything to everyone? Continue to Part 3: Message Routing with Topics.