feat: Add At-Least-Once Fan-Out With Redelivery On Nack

README.md

@@ -4,7 +4,7 @@ A from-scratch in-memory message broker exploring pub/sub, work queues, delivery
 
 ## What this demonstrates
 
-Message brokers hide a lot of machinery behind `publish` and `subscribe`. This repo pulls that machinery apart one piece at a time and keeps each piece small enough to read in a sitting. The topics it works through include topic-based fan-out, wildcard routing, competing-consumer work queues with acknowledgements, and later delivery guarantees (at-most-once vs at-least-once), dead-letter handling, ordering, durability, and backpressure. Everything runs in a single process with no dependencies, so the focus stays on the semantics rather than the transport.
+Message brokers hide a lot of machinery behind `publish` and `subscribe`. This repo pulls that machinery apart one piece at a time and keeps each piece small enough to read in a sitting. The topics it works through include topic-based fan-out, wildcard routing, competing-consumer work queues with acknowledgements, at-least-once redelivery on nack, dead-letter handling, and later ordering, durability, and backpressure. Everything runs in a single process with no dependencies, so the focus stays on the semantics rather than the transport.
 
 ## Concepts demonstrated
 
@@ -13,14 +13,19 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **AMQP-style topic patterns** (`*` one segment, `#` zero or more) with a small dynamic program for matching
 - **Competing consumers** (work queue): each message is delivered to exactly one consumer
 - **Acknowledgements** (`ack` / `nack`) and **at-least-once** redelivery on nack or consumer cancel
+- **Per-subscriber inbox** on reliable fan-out: each subscriber gets its own copy, settles independently, and a nack redelivers only that copy
+- **Redelivered flag** and **delivery count** so handlers can tell first delivery from a retry
 - **Prefetch** as a per-consumer in-flight limit for fair sharing
 - **Redelivery counting**, **bounded redelivery** (`maxDeliveryCount`, default 10; excess drops), and **tail requeue** on nack so other ready work is not starved
+- **Dead-letter queue** for poison messages after a bounded number of failed attempts
 
 ## What's implemented
 
 - **Topic-based publish/subscribe with fan-out.** A `Broker` where subscribers register handlers against a topic and every publish fans out to all matching subscribers, with monotonic message ids, idempotent unsubscribe, and snapshot-consistent delivery (subscribing or unsubscribing during dispatch never changes who receives the in-flight message).
 - **Wildcard topic subscriptions.** AMQP-style pattern bindings where `*` matches exactly one segment and `#` matches zero or more, so one handler can receive a whole family of topics (`orders.#`, `*.created.us`). Matching uses a `#`-aware dynamic program, and a published message fans out to exact subscribers first, then every matching pattern.
 - **Competing-consumer work queue with acks.** A `WorkQueue` where each enqueued message goes to exactly one consumer. Consumers `ack` to settle or `nack` to requeue (or drop). Handler throw is treated as a requeueing nack. Prefetch defaults to 1 so peers share fairly; cancelling a consumer requeues its unacked deliveries at the head of ready. A nack requeues at the tail. Redelivery is bounded by `maxDeliveryCount` (default 10) so a permanent poison payload cannot busy-spin inside `enqueue`.
+- **Dead-letter queue for poison messages.** A `DeadLetterQueue` that counts attempts per original message (and optional subscription), parks a failed payload once `maxAttempts` is exhausted, and supports peek, drop, purge, and redrive back onto a publisher.
+- **At-least-once delivery with redelivery on nack.** A `ReliableBroker` where publish still fans out, but each subscriber owns a private inbox. `ack` settles that copy. `nack` (or a handler throw) redelivers it to the same subscriber with `deliveryCount` incremented and `redelivered` set. Peers who already acked are not retried. `nack({ requeue: false })` drops only that subscriber's copy. Retries stop at `maxDeliveryCount`. Optional per-subscriber prefetch isolates a slow consumer without stalling everyone else.
 
 ## Usage
 
@@ -80,6 +85,32 @@ queue.consume((delivery) => {
 queue.enqueue({ jobId: 'job-1' }) // only one of the two consumers receives it
 ```
 
+At-least-once fan-out: every subscriber gets a copy, and a nack retries only that copy.
+
+```ts
+import { ReliableBroker } from 'event-broker-lab'
+
+const reliable = new ReliableBroker<{ orderId: string }>()
+
+reliable.subscribe('orders.created', (delivery) => {
+  if (delivery.message.redelivered) {
+    // already tried once
+  }
+  try {
+    // handle the order
+    delivery.ack()
+  } catch {
+    delivery.nack()
+  }
+})
+
+reliable.subscribe('orders.created', (delivery) => {
+  delivery.ack() // this copy settles even if the other subscriber nacks
+})
+
+reliable.publish('orders.created', { orderId: 'A-1' })
+```
+
 ## Running the tests
 
 ```sh

src/index.ts

@@ -1,6 +1,13 @@
 export { Broker } from './broker.js'
 export type { Message, Handler, Unsubscribe } from './broker.js'
 export { matchTopic, isValidTopic, isValidPattern } from './topic-match.js'
+export { DeadLetterQueue, DeadLetterFullError, withDeadLetter } from './dead-letter.js'
+export type {
+  DeadLetterEnvelope,
+  DeadLetterQueueOptions,
+  FailResult,
+  FailStatus,
+} from './dead-letter.js'
 export { WorkQueue } from './work-queue.js'
 export type {
   WorkMessage,
@@ -9,3 +16,11 @@ export type {
   WorkQueueOptions,
   Unsubscribe as WorkQueueUnsubscribe,
 } from './work-queue.js'
+export { ReliableBroker } from './reliable-broker.js'
+export type {
+  ReliableMessage,
+  ReliableDelivery,
+  ReliableHandler,
+  ReliableBrokerOptions,
+  ReliableSubscribeOptions,
+} from './reliable-broker.js'

src/reliable-broker.ts

@@ -0,0 +1,200 @@
+import type { Unsubscribe } from './broker.js'
+
+export interface ReliableMessage<T> {
+  readonly id: number
+  readonly topic: string
+  readonly payload: T
+  readonly deliveryCount: number
+  readonly redelivered: boolean
+}
+
+export interface ReliableDelivery<T> {
+  readonly message: ReliableMessage<T>
+  readonly deliveryTag: number
+  ack(): void
+  nack(options?: { requeue?: boolean }): void
+}
+
+export type ReliableHandler<T> = (delivery: ReliableDelivery<T>) => void
+
+export interface ReliableBrokerOptions {
+  maxDeliveryCount?: number
+}
+
+export interface ReliableSubscribeOptions {
+  prefetch?: number
+}
+
+interface PendingCopy<T> {
+  id: number
+  topic: string
+  payload: T
+  deliveryCount: number
+}
+
+interface InFlightCopy<T> {
+  copy: PendingCopy<T>
+  subscriberId: number
+  settled: boolean
+}
+
+interface Subscriber<T> {
+  id: number
+  topic: string
+  handler: ReliableHandler<T>
+  prefetch: number | undefined
+  ready: PendingCopy<T>[]
+  inFlight: number
+  active: boolean
+}
+
+const DEFAULT_MAX_DELIVERY_COUNT = 10
+
+export class ReliableBroker<T> {
+  private readonly subscribers: Subscriber<T>[] = []
+  private readonly inFlight = new Map<number, InFlightCopy<T>>()
+  private readonly maxDeliveryCount: number
+  private nextMessageId = 1
+  private nextDeliveryTag = 1
+  private nextSubscriberId = 1
+  private pumping = false
+
+  constructor(options?: ReliableBrokerOptions) {
+    const max = options?.maxDeliveryCount ?? DEFAULT_MAX_DELIVERY_COUNT
+    if (!Number.isInteger(max) || max < 1) {
+      throw new Error(`maxDeliveryCount must be a positive integer, got ${max}`)
+    }
+    this.maxDeliveryCount = max
+  }
+
+  subscribe(
+    topic: string,
+    handler: ReliableHandler<T>,
+    options?: ReliableSubscribeOptions,
+  ): Unsubscribe {
+    const prefetch = options?.prefetch
+    if (prefetch !== undefined && (!Number.isInteger(prefetch) || prefetch < 1)) {
+      throw new Error(`prefetch must be a positive integer, got ${prefetch}`)
+    }
+
+    const subscriber: Subscriber<T> = {
+      id: this.nextSubscriberId++,
+      topic,
+      handler,
+      prefetch,
+      ready: [],
+      inFlight: 0,
+      active: true,
+    }
+    this.subscribers.push(subscriber)
+
+    return () => {
+      if (!subscriber.active) return
+      subscriber.active = false
+      for (const [tag, entry] of this.inFlight) {
+        if (entry.subscriberId !== subscriber.id || entry.settled) continue
+        entry.settled = true
+        this.inFlight.delete(tag)
+      }
+      subscriber.ready.length = 0
+      subscriber.inFlight = 0
+      const i = this.subscribers.indexOf(subscriber)
+      if (i !== -1) this.subscribers.splice(i, 1)
+    }
+  }
+
+  publish(topic: string, payload: T): number {
+    const targets = this.subscribers.filter((s) => s.active && s.topic === topic)
+    if (targets.length === 0) return 0
+    const id = this.nextMessageId++
+    for (const subscriber of targets) {
+      subscriber.ready.push({ id, topic, payload, deliveryCount: 0 })
+    }
+    this.pump()
+    return targets.length
+  }
+
+  readyCount(): number {
+    return this.subscribers.reduce((n, s) => n + s.ready.length, 0)
+  }
+
+  inFlightCount(): number {
+    return this.inFlight.size
+  }
+
+  subscriberCount(): number {
+    return this.subscribers.length
+  }
+
+  private hasCapacity(subscriber: Subscriber<T>): boolean {
+    return subscriber.prefetch === undefined || subscriber.inFlight < subscriber.prefetch
+  }
+
+  private pump(): void {
+    if (this.pumping) return
+    this.pumping = true
+    try {
+      let progressed = true
+      while (progressed) {
+        progressed = false
+        for (const subscriber of this.subscribers) {
+          if (!subscriber.active) continue
+          const seenThisRound = new Set<number>()
+          while (subscriber.ready.length > 0 && this.hasCapacity(subscriber)) {
+            const head = subscriber.ready[0]
+            // sync nack requeues the same id; skip it until the next pump round
+            if (!head || seenThisRound.has(head.id)) break
+            subscriber.ready.shift()
+            seenThisRound.add(head.id)
+            this.deliver(subscriber, head)
+            progressed = true
+          }
+        }
+      }
+    } finally {
+      this.pumping = false
+    }
+  }
+
+  private deliver(subscriber: Subscriber<T>, copy: PendingCopy<T>): void {
+    copy.deliveryCount += 1
+    const deliveryTag = this.nextDeliveryTag++
+    const entry: InFlightCopy<T> = { copy, subscriberId: subscriber.id, settled: false }
+    this.inFlight.set(deliveryTag, entry)
+    subscriber.inFlight += 1
+
+    const delivery: ReliableDelivery<T> = {
+      message: {
+        id: copy.id,
+        topic: copy.topic,
+        payload: copy.payload,
+        deliveryCount: copy.deliveryCount,
+        redelivered: copy.deliveryCount > 1,
+      },
+      deliveryTag,
+      ack: () => this.settle(deliveryTag, false),
+      nack: (options) => this.settle(deliveryTag, options?.requeue !== false),
+    }
+
+    try {
+      subscriber.handler(delivery)
+    } catch {
+      if (!entry.settled) this.settle(deliveryTag, true)
+    }
+  }
+
+  private settle(deliveryTag: number, requeue: boolean): void {
+    const entry = this.inFlight.get(deliveryTag)
+    if (!entry || entry.settled) return
+    entry.settled = true
+    this.inFlight.delete(deliveryTag)
+
+    const subscriber = this.subscribers.find((s) => s.id === entry.subscriberId)
+    if (subscriber?.active) subscriber.inFlight -= 1
+
+    if (requeue && subscriber?.active && entry.copy.deliveryCount < this.maxDeliveryCount) {
+      subscriber.ready.push(entry.copy)
+    }
+    this.pump()
+  }
+}