feat: Add Dead-Letter Queue for Poison Messages

README.md

@@ -1,15 +1,27 @@
 # event-broker-lab
 
-A from-scratch in-memory message broker exploring pub/sub, work queues, delivery guarantees, and backpressure: the concepts behind Kafka, RabbitMQ, and SQS, built small enough to read.
+A from-scratch in-memory message broker exploring pub/sub, work queues, delivery guarantees, dead-letter handling, and backpressure: the concepts behind Kafka, RabbitMQ, and SQS, built small enough to read.
 
 ## 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, subscriber lifecycle and snapshot-consistent delivery, and later work queues, delivery guarantees (at-most-once vs at-least-once), acknowledgements, dead-letter handling, 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, subscriber lifecycle and snapshot-consistent delivery, wildcard routing, dead-letter handling for poison messages, and later work queues, acknowledgements, and backpressure. Everything runs in a single process with no dependencies, so the focus stays on the semantics rather than the transport.
+
+## Concepts demonstrated
+
+- **Topic-based publish/subscribe** with fan-out delivery
+- **Snapshot-consistent dispatch** (subscribe/unsubscribe mid-delivery does not change the in-flight recipient set)
+- **AMQP-style topic patterns** (`*` one segment, `#` zero or more) with a small dynamic program for matching
+- **Poison-message isolation** so one throwing subscriber does not abort the rest of a fan-out
+- **Redrive policy** (`maxAttempts`) that retries a failed handler immediately on the same publish, then gives up
+- **Dead-letter queue** that parks the original payload with source topic, attempt count, and last error
+- **Bounded failure queue** (explicit `capacity`) that refuses new entries instead of dropping evidence
+- **Redrive / replay** back onto the original topic, including a publish-then-remove so a failed replay leaves the envelope in place
 
 ## 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.
+- **Dead-letter queue for poison messages.** A `DeadLetterQueue` plus `withDeadLetter` wrapper. Each failed delivery increments a per-subscription attempt ledger (two wrappers on one queue do not share retries or hide each other's last error). After `maxAttempts` the payload is parked with its source topic, attempt count, and last error. `Broker.publish` finishes the subscriber snapshot even if a handler throws, then rethrows the first error. Operators can inspect, `drop`, `purge`, or `redrive` back onto the original topic. A full DLQ throws `DeadLetterFullError` rather than silently discarding the poison payload.
 
 ## Usage
 
@@ -44,6 +56,32 @@ broker.publish('orders.created.us', { orderId: 'A-2' })  // matches
 broker.publish('shipments.created', { orderId: 'B-1' })  // does not match
 ```
 
+Park a poison payload after it exhausts retries, then redrive it:
+
+```ts
+import { Broker, DeadLetterQueue, withDeadLetter } from 'event-broker-lab'
+
+const broker = new Broker<string>()
+const dlq = new DeadLetterQueue<string>({ maxAttempts: 3 })
+
+const off = broker.subscribe(
+  'orders',
+  withDeadLetter(dlq, (msg) => {
+    if (msg.payload === 'poison') throw new Error('bad payload')
+    console.log(msg.payload)
+  }),
+)
+
+broker.publish('orders', 'poison')
+console.log(dlq.size()) // 1
+console.log(dlq.peek()[0]?.error) // bad payload
+
+off() // unsubscribe first, otherwise redrive hits the same handler and re-parks
+dlq.redriveAll((topic, payload) => {
+  broker.publish(topic, payload)
+})
+```
+
 ## Running the tests
 
 ```sh

src/broker.ts

@@ -67,9 +67,15 @@ export class Broker<T> {
     if (snapshot.length === 0) return 0
 
     const message: Message<T> = { topic, payload, id: this.nextId++ }
+    let firstError: unknown
     for (const handler of snapshot) {
-      handler(message)
+      try {
+        handler(message)
+      } catch (error) {
+        if (firstError === undefined) firstError = error
+      }
     }
+    if (firstError !== undefined) throw firstError
     return snapshot.length
   }
 

src/dead-letter.ts

@@ -0,0 +1,191 @@
+import type { Handler, Message } from './broker.js'
+
+type Delivery<T> = Pick<Message<T>, 'topic' | 'payload' | 'id'>
+
+export class DeadLetterFullError extends Error {
+  readonly capacity: number
+
+  constructor(capacity: number) {
+    super(`dead-letter queue is full (capacity ${capacity})`)
+    this.name = 'DeadLetterFullError'
+    this.capacity = capacity
+  }
+}
+
+export interface DeadLetterEnvelope<T> {
+  readonly id: number
+  readonly sourceTopic: string
+  readonly payload: T
+  readonly originalId: number
+  readonly attempts: number
+  readonly error: string
+  readonly enqueuedAt: number
+}
+
+export type FailStatus = 'retry' | 'dead_lettered'
+
+export interface FailResult<T> {
+  readonly status: FailStatus
+  readonly attempts: number
+  readonly envelope?: DeadLetterEnvelope<T>
+}
+
+export interface DeadLetterQueueOptions {
+  readonly maxAttempts?: number
+  readonly capacity?: number
+  readonly now?: () => number
+}
+
+export class DeadLetterQueue<T> {
+  readonly maxAttempts: number
+  readonly capacity: number
+  private readonly now: () => number
+  private nextId = 1
+  private readonly items = new Map<number, DeadLetterEnvelope<T>>()
+  private readonly byOriginal = new Map<string, number>()
+  private readonly attempts = new Map<string, number>()
+  private readonly ledgerOf = new Map<number, string>()
+
+  constructor(options: DeadLetterQueueOptions = {}) {
+    const maxAttempts = options.maxAttempts ?? 3
+    if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
+      throw new Error(`maxAttempts must be a positive integer, got ${maxAttempts}`)
+    }
+    const capacity = options.capacity ?? Number.POSITIVE_INFINITY
+    if (capacity !== Number.POSITIVE_INFINITY && (!Number.isInteger(capacity) || capacity < 1)) {
+      throw new Error(`capacity must be a positive integer, got ${capacity}`)
+    }
+    this.maxAttempts = maxAttempts
+    this.capacity = capacity
+    this.now = options.now ?? Date.now
+  }
+
+  fail(message: Delivery<T>, error: unknown, subscription?: string): FailResult<T> {
+    const parked = this.parked(message.id, subscription)
+    if (parked) return { status: 'dead_lettered', attempts: parked.attempts, envelope: parked }
+
+    const key = this.ledgerKey(message.id, subscription)
+    let attempts = this.attempts.get(key) ?? 0
+    if (attempts < this.maxAttempts) {
+      attempts += 1
+      this.attempts.set(key, attempts)
+      if (attempts < this.maxAttempts) return { status: 'retry', attempts }
+    }
+
+    const envelope = this.enqueue(message, error, attempts, key)
+    return { status: 'dead_lettered', attempts, envelope }
+  }
+
+  deadLetter(message: Delivery<T>, error: unknown, subscription?: string): DeadLetterEnvelope<T> {
+    const parked = this.parked(message.id, subscription)
+    if (parked) return parked
+    const key = this.ledgerKey(message.id, subscription)
+    const attempts = Math.max(this.attempts.get(key) ?? 0, 1)
+    this.attempts.set(key, attempts)
+    return this.enqueue(message, error, attempts, key)
+  }
+
+  succeed(originalId: number, subscription?: string): void {
+    const key = this.ledgerKey(originalId, subscription)
+    if (!this.byOriginal.has(key)) this.attempts.delete(key)
+  }
+
+  attemptCount(originalId: number, subscription?: string): number {
+    return this.attempts.get(this.ledgerKey(originalId, subscription)) ?? 0
+  }
+
+  peek(): readonly DeadLetterEnvelope<T>[] {
+    return [...this.items.values()]
+  }
+
+  size(): number {
+    return this.items.size
+  }
+
+  drop(id: number): boolean {
+    const envelope = this.items.get(id)
+    if (!envelope) return false
+    this.forget(envelope)
+    return true
+  }
+
+  purge(): number {
+    const n = this.items.size
+    for (const envelope of [...this.items.values()]) this.forget(envelope)
+    return n
+  }
+
+  redrive(id: number, publish: (topic: string, payload: T) => void): DeadLetterEnvelope<T> | undefined {
+    const envelope = this.items.get(id)
+    if (!envelope) return undefined
+    publish(envelope.sourceTopic, envelope.payload)
+    this.forget(envelope)
+    return envelope
+  }
+
+  redriveAll(publish: (topic: string, payload: T) => void): DeadLetterEnvelope<T>[] {
+    const moved: DeadLetterEnvelope<T>[] = []
+    for (const envelope of [...this.items.values()]) {
+      const result = this.redrive(envelope.id, publish)
+      if (result) moved.push(result)
+    }
+    return moved
+  }
+
+  private ledgerKey(originalId: number, subscription?: string): string {
+    return `${subscription ?? ''}:${originalId}`
+  }
+
+  private parked(originalId: number, subscription?: string): DeadLetterEnvelope<T> | undefined {
+    const id = this.byOriginal.get(this.ledgerKey(originalId, subscription))
+    return id === undefined ? undefined : this.items.get(id)
+  }
+
+  private forget(envelope: DeadLetterEnvelope<T>): void {
+    const key = this.ledgerOf.get(envelope.id)
+    this.items.delete(envelope.id)
+    this.ledgerOf.delete(envelope.id)
+    if (key === undefined) return
+    this.byOriginal.delete(key)
+    this.attempts.delete(key)
+  }
+
+  private enqueue(
+    message: Delivery<T>,
+    error: unknown,
+    attempts: number,
+    key: string,
+  ): DeadLetterEnvelope<T> {
+    if (this.items.size >= this.capacity) throw new DeadLetterFullError(this.capacity)
+    const envelope: DeadLetterEnvelope<T> = {
+      id: this.nextId++,
+      sourceTopic: message.topic,
+      payload: message.payload,
+      originalId: message.id,
+      attempts,
+      error: error instanceof Error ? error.message : String(error),
+      enqueuedAt: this.now(),
+    }
+    this.items.set(envelope.id, envelope)
+    this.byOriginal.set(key, envelope.id)
+    this.ledgerOf.set(envelope.id, key)
+    return envelope
+  }
+}
+
+let nextWrapper = 1
+
+export function withDeadLetter<T>(dlq: DeadLetterQueue<T>, handler: Handler<T>): Handler<T> {
+  const subscription = `w${nextWrapper++}`
+  return (message) => {
+    for (;;) {
+      try {
+        handler(message)
+        dlq.succeed(message.id, subscription)
+        return
+      } catch (error) {
+        if (dlq.fail(message, error, subscription).status === 'dead_lettered') return
+      }
+    }
+  }
+}

src/index.ts

@@ -1,3 +1,10 @@
 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'

test/broker.test.ts

@@ -119,6 +119,19 @@ describe('delivery snapshot semantics', () => {
     expect(late).toHaveBeenCalledOnce()
   })
 
+  it('still delivers this message to later subscribers when an earlier handler throws', () => {
+    const broker = new Broker<string>()
+    const later = vi.fn()
+    broker.subscribe('t', () => {
+      throw new Error('unwrapped')
+    })
+    broker.subscribe('t', later)
+
+    expect(() => broker.publish('t', 'payload')).toThrow(/unwrapped/)
+    expect(later).toHaveBeenCalledOnce()
+    expect(later.mock.calls[0]?.[0]).toMatchObject({ topic: 't', payload: 'payload' })
+  })
+
   it('still delivers this message to a handler that unsubscribes a peer mid-dispatch', () => {
     const broker = new Broker<string>()
     const received: string[] = []