From 999ec9a30199e6d9accb6f6bdf0222616762a693 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Thu, 13 Aug 2026 17:31:36 +0000 Subject: [PATCH 1/2] feat: add dead-letter queue for poison messages Park a payload after it exhausts maxAttempts, isolate handler throws so fan-out continues, and redrive only after a successful publish. --- README.md | 41 +++++++- src/dead-letter.ts | 171 ++++++++++++++++++++++++++++++ src/index.ts | 7 ++ test/dead-letter.test.ts | 222 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 439 insertions(+), 2 deletions(-) create mode 100644 src/dead-letter.ts create mode 100644 test/dead-letter.test.ts diff --git a/README.md b/README.md index e50b65e..8deaabd 100644 --- a/README.md +++ b/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`, SQS-style max receive count) that retries a delivery 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-message attempt ledger. After `maxAttempts` the payload is parked with its source topic, attempt count, and last 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,31 @@ 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() +const dlq = new DeadLetterQueue({ maxAttempts: 3 }) + +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 + +dlq.redriveAll((topic, payload) => { + broker.publish(topic, payload) +}) +``` + ## Running the tests ```sh diff --git a/src/dead-letter.ts b/src/dead-letter.ts new file mode 100644 index 0000000..252632c --- /dev/null +++ b/src/dead-letter.ts @@ -0,0 +1,171 @@ +import type { Handler, Message } from './broker.js' + +type Delivery = Pick, '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 { + 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 { + readonly status: FailStatus + readonly attempts: number + readonly envelope?: DeadLetterEnvelope +} + +export interface DeadLetterQueueOptions { + readonly maxAttempts?: number + readonly capacity?: number + readonly now?: () => number +} + +export class DeadLetterQueue { + readonly maxAttempts: number + readonly capacity: number + private readonly now: () => number + private nextId = 1 + private readonly items = new Map>() + private readonly byOriginal = new Map() + private readonly attempts = new Map() + + 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, error: unknown): FailResult { + const parked = this.parked(message.id) + if (parked) return { status: 'dead_lettered', attempts: parked.attempts, envelope: parked } + + let attempts = this.attempts.get(message.id) ?? 0 + if (attempts < this.maxAttempts) { + attempts += 1 + this.attempts.set(message.id, attempts) + if (attempts < this.maxAttempts) return { status: 'retry', attempts } + } + + const envelope = this.enqueue(message, error, attempts) + return { status: 'dead_lettered', attempts, envelope } + } + + deadLetter(message: Delivery, error: unknown): DeadLetterEnvelope { + const parked = this.parked(message.id) + if (parked) return parked + const attempts = Math.max(this.attempts.get(message.id) ?? 0, 1) + this.attempts.set(message.id, attempts) + return this.enqueue(message, error, attempts) + } + + succeed(originalId: number): void { + if (!this.byOriginal.has(originalId)) this.attempts.delete(originalId) + } + + attemptCount(originalId: number): number { + return this.attempts.get(originalId) ?? 0 + } + + peek(): readonly DeadLetterEnvelope[] { + 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 | 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[] { + const moved: DeadLetterEnvelope[] = [] + for (const envelope of [...this.items.values()]) { + const result = this.redrive(envelope.id, publish) + if (result) moved.push(result) + } + return moved + } + + private parked(originalId: number): DeadLetterEnvelope | undefined { + const id = this.byOriginal.get(originalId) + return id === undefined ? undefined : this.items.get(id) + } + + private forget(envelope: DeadLetterEnvelope): void { + this.items.delete(envelope.id) + this.byOriginal.delete(envelope.originalId) + this.attempts.delete(envelope.originalId) + } + + private enqueue(message: Delivery, error: unknown, attempts: number): DeadLetterEnvelope { + if (this.items.size >= this.capacity) throw new DeadLetterFullError(this.capacity) + const envelope: DeadLetterEnvelope = { + 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(message.id, envelope.id) + return envelope + } +} + +export function withDeadLetter(dlq: DeadLetterQueue, handler: Handler): Handler { + return (message) => { + for (;;) { + try { + handler(message) + dlq.succeed(message.id) + return + } catch (error) { + if (dlq.fail(message, error).status === 'dead_lettered') return + } + } + } +} diff --git a/src/index.ts b/src/index.ts index eee4f2e..0b4294f 100644 --- a/src/index.ts +++ b/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' diff --git a/test/dead-letter.test.ts b/test/dead-letter.test.ts new file mode 100644 index 0000000..14bd6e8 --- /dev/null +++ b/test/dead-letter.test.ts @@ -0,0 +1,222 @@ +import { describe, it, expect, vi } from 'vitest' +import { + Broker, + DeadLetterFullError, + DeadLetterQueue, + withDeadLetter, +} from '../src/index.js' + +const msg = (id: number, payload = 'x', topic = 'orders') => ({ topic, payload, id }) + +describe('DeadLetterQueue.fail', () => { + it('retries until maxAttempts then parks the payload', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 3, now: () => 42 }) + expect(dlq.fail(msg(1), new Error('boom'))).toEqual({ status: 'retry', attempts: 1 }) + expect(dlq.fail(msg(1), new Error('boom'))).toEqual({ status: 'retry', attempts: 2 }) + const last = dlq.fail(msg(1), new Error('boom')) + expect(last).toMatchObject({ + status: 'dead_lettered', + attempts: 3, + envelope: { + id: 1, + sourceTopic: 'orders', + payload: 'x', + originalId: 1, + attempts: 3, + error: 'boom', + enqueuedAt: 42, + }, + }) + expect(dlq.peek()).toEqual([last.envelope]) + }) + + it('dead-letters on the first fail when maxAttempts is 1', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + const result = dlq.fail(msg(7, 'poison'), 'bad schema') + expect(result.status).toBe('dead_lettered') + expect(result.envelope).toMatchObject({ attempts: 1, error: 'bad schema' }) + }) + + it('is idempotent once the original id is already parked', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + const first = dlq.fail(msg(1), new Error('a')) + expect(dlq.fail(msg(1), new Error('b')).envelope).toBe(first.envelope) + expect(dlq.peek()[0]?.error).toBe('a') + }) + + it('tracks attempts per original id and stringifies non-Error throws', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 5 }) + dlq.fail(msg(1), 'a') + dlq.fail(msg(2), { code: 9 }) + dlq.fail(msg(1), 'a') + expect(dlq.attemptCount(1)).toBe(2) + expect(dlq.attemptCount(2)).toBe(1) + expect(dlq.attemptCount(99)).toBe(0) + const parked = new DeadLetterQueue({ maxAttempts: 1 }) + expect(parked.fail(msg(1), { code: 9 }).envelope?.error).toBe('[object Object]') + expect(parked.fail(msg(2), 0).envelope?.error).toBe('0') + }) +}) + +describe('succeed and explicit deadLetter', () => { + it('clears the ledger so the next fail starts at attempt 1', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 3 }) + dlq.fail(msg(1), 'x') + dlq.fail(msg(1), 'x') + dlq.succeed(1) + expect(dlq.attemptCount(1)).toBe(0) + expect(dlq.fail(msg(1), 'x').attempts).toBe(1) + expect(dlq.size()).toBe(0) + expect(() => dlq.succeed(99)).not.toThrow() + }) + + it('does not clear a message already on the DLQ', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + dlq.fail(msg(1), 'x') + dlq.succeed(1) + expect(dlq.size()).toBe(1) + expect(dlq.attemptCount(1)).toBe(1) + }) + + it('deadLetter parks immediately without waiting for maxAttempts', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 8 }) + const envelope = dlq.deadLetter(msg(3, 'nope'), new Error('rejected')) + expect(envelope).toMatchObject({ attempts: 1, error: 'rejected' }) + expect(dlq.deadLetter(msg(3, 'nope'), new Error('again'))).toBe(envelope) + }) +}) + +describe('capacity, drop, purge, redrive', () => { + it('refuses a new poison payload when the queue is full', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1, capacity: 1 }) + dlq.fail(msg(1), 'first') + expect(() => dlq.fail(msg(2), 'second')).toThrow(DeadLetterFullError) + expect(dlq.peek()[0]?.originalId).toBe(1) + }) + + it('rejects non-positive options', () => { + expect(() => new DeadLetterQueue({ maxAttempts: 0 })).toThrow(/maxAttempts/) + expect(() => new DeadLetterQueue({ maxAttempts: 1.5 })).toThrow(/maxAttempts/) + expect(() => new DeadLetterQueue({ capacity: 0 })).toThrow(/capacity/) + expect(() => new DeadLetterQueue({ capacity: -1 })).toThrow(/capacity/) + }) + + it('drop and purge remove envelopes and forget their ledgers', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + expect(dlq.purge()).toBe(0) + const parked = dlq.fail(msg(4), 'x').envelope! + expect(dlq.drop(parked.id)).toBe(true) + expect(dlq.drop(parked.id)).toBe(false) + expect(dlq.attemptCount(4)).toBe(0) + dlq.fail(msg(1), 'a') + dlq.fail(msg(2), 'b') + expect(dlq.purge()).toBe(2) + expect(dlq.peek()).toEqual([]) + }) + + it('redrive publishes then removes, and leaves the envelope if publish throws', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + const parked = dlq.fail(msg(9, 'job'), 'x').envelope! + expect(() => + dlq.redrive(parked.id, () => { + throw new Error('broker down') + }), + ).toThrow(/broker down/) + expect(dlq.size()).toBe(1) + const seen: Array<[string, string]> = [] + expect(dlq.redrive(parked.id, (topic, payload) => seen.push([topic, payload]))).toEqual(parked) + expect(seen).toEqual([['orders', 'job']]) + expect(dlq.redrive(parked.id, () => {})).toBeUndefined() + }) + + it('redriveAll walks insertion order and stops if a publish throws', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + dlq.fail(msg(1, 'a'), 'x') + dlq.fail(msg(2, 'b'), 'x') + dlq.fail(msg(3, 'c'), 'x') + const seen: string[] = [] + expect(() => + dlq.redriveAll((_topic, payload) => { + if (payload === 'b') throw new Error('stop') + seen.push(payload) + }), + ).toThrow(/stop/) + expect(seen).toEqual(['a']) + expect(dlq.peek().map((e) => e.payload)).toEqual(['b', 'c']) + }) +}) + +describe('withDeadLetter', () => { + it('retries the same message then parks a permanent failure', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 3 }) + const handler = vi.fn(() => { + throw new Error('poison') + }) + expect(() => withDeadLetter(dlq, handler)({ topic: 'jobs', payload: 'bad', id: 11 })).not.toThrow() + expect(handler).toHaveBeenCalledTimes(3) + expect(dlq.peek()[0]).toMatchObject({ + originalId: 11, + sourceTopic: 'jobs', + payload: 'bad', + attempts: 3, + error: 'poison', + }) + }) + + it('does not park a message that succeeds on a later attempt', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 4 }) + let n = 0 + withDeadLetter(dlq, () => { + n += 1 + if (n < 3) throw new Error('flaky') + })({ topic: 'jobs', payload: 'ok', id: 1 }) + expect(n).toBe(3) + expect(dlq.size()).toBe(0) + expect(dlq.attemptCount(1)).toBe(0) + }) + + it('keeps fan-out going when one subscriber throws', () => { + const broker = new Broker() + const dlq = new DeadLetterQueue({ maxAttempts: 2 }) + const healthy = vi.fn() + broker.subscribe( + 'orders', + withDeadLetter(dlq, () => { + throw new Error('poison') + }), + ) + broker.subscribe('orders', healthy) + expect(broker.publish('orders', 'A-1')).toBe(2) + expect(healthy).toHaveBeenCalledOnce() + expect(dlq.peek()[0]?.payload).toBe('A-1') + }) + + it('redrives a parked payload back through the broker as a new publish', () => { + const broker = new Broker() + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + const seen: string[] = [] + let failNext = true + broker.subscribe( + 'orders', + withDeadLetter(dlq, (m) => { + if (failNext) throw new Error('first pass') + seen.push(m.payload) + }), + ) + broker.publish('orders', 'A-1') + failNext = false + dlq.redrive(dlq.peek()[0]!.id, (topic, payload) => broker.publish(topic, payload)) + expect(seen).toEqual(['A-1']) + expect(dlq.size()).toBe(0) + }) + + it('surfaces DeadLetterFullError instead of parking a new poison payload', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1, capacity: 1 }) + const wrapped = withDeadLetter(dlq, () => { + throw new Error('nope') + }) + wrapped({ topic: 't', payload: 'one', id: 1 }) + expect(() => wrapped({ topic: 't', payload: 'two', id: 2 })).toThrow(DeadLetterFullError) + expect(dlq.peek()[0]?.payload).toBe('one') + }) +}) From 42f1f12f0543d1c4c41435005b5752b057ae5b43 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Thu, 13 Aug 2026 17:40:08 +0000 Subject: [PATCH 2/2] fix: isolate handler throws and per-subscription DLQ retries Publish finishes the subscriber snapshot before rethrowing the first error so later handlers still run. Retry budget and parked envelopes are keyed per wrapper, not only by message id. --- README.md | 7 ++-- src/broker.ts | 8 ++++- src/dead-letter.ts | 72 +++++++++++++++++++++++-------------- test/broker.test.ts | 13 +++++++ test/dead-letter.test.ts | 78 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 148 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 8deaabd..59001f5 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r - **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`, SQS-style max receive count) that retries a delivery then gives up +- **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 @@ -21,7 +21,7 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r - **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-message attempt ledger. After `maxAttempts` the payload is parked with its source topic, attempt count, and last 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. +- **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 @@ -64,7 +64,7 @@ import { Broker, DeadLetterQueue, withDeadLetter } from 'event-broker-lab' const broker = new Broker() const dlq = new DeadLetterQueue({ maxAttempts: 3 }) -broker.subscribe( +const off = broker.subscribe( 'orders', withDeadLetter(dlq, (msg) => { if (msg.payload === 'poison') throw new Error('bad payload') @@ -76,6 +76,7 @@ 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) }) diff --git a/src/broker.ts b/src/broker.ts index 4e9754e..39b8f70 100644 --- a/src/broker.ts +++ b/src/broker.ts @@ -67,9 +67,15 @@ export class Broker { if (snapshot.length === 0) return 0 const message: Message = { 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 } diff --git a/src/dead-letter.ts b/src/dead-letter.ts index 252632c..74da048 100644 --- a/src/dead-letter.ts +++ b/src/dead-letter.ts @@ -42,8 +42,9 @@ export class DeadLetterQueue { private readonly now: () => number private nextId = 1 private readonly items = new Map>() - private readonly byOriginal = new Map() - private readonly attempts = new Map() + private readonly byOriginal = new Map() + private readonly attempts = new Map() + private readonly ledgerOf = new Map() constructor(options: DeadLetterQueueOptions = {}) { const maxAttempts = options.maxAttempts ?? 3 @@ -59,35 +60,38 @@ export class DeadLetterQueue { this.now = options.now ?? Date.now } - fail(message: Delivery, error: unknown): FailResult { - const parked = this.parked(message.id) + fail(message: Delivery, error: unknown, subscription?: string): FailResult { + const parked = this.parked(message.id, subscription) if (parked) return { status: 'dead_lettered', attempts: parked.attempts, envelope: parked } - let attempts = this.attempts.get(message.id) ?? 0 + const key = this.ledgerKey(message.id, subscription) + let attempts = this.attempts.get(key) ?? 0 if (attempts < this.maxAttempts) { attempts += 1 - this.attempts.set(message.id, attempts) + this.attempts.set(key, attempts) if (attempts < this.maxAttempts) return { status: 'retry', attempts } } - const envelope = this.enqueue(message, error, attempts) + const envelope = this.enqueue(message, error, attempts, key) return { status: 'dead_lettered', attempts, envelope } } - deadLetter(message: Delivery, error: unknown): DeadLetterEnvelope { - const parked = this.parked(message.id) + deadLetter(message: Delivery, error: unknown, subscription?: string): DeadLetterEnvelope { + const parked = this.parked(message.id, subscription) if (parked) return parked - const attempts = Math.max(this.attempts.get(message.id) ?? 0, 1) - this.attempts.set(message.id, attempts) - return this.enqueue(message, error, attempts) + 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): void { - if (!this.byOriginal.has(originalId)) this.attempts.delete(originalId) + succeed(originalId: number, subscription?: string): void { + const key = this.ledgerKey(originalId, subscription) + if (!this.byOriginal.has(key)) this.attempts.delete(key) } - attemptCount(originalId: number): number { - return this.attempts.get(originalId) ?? 0 + attemptCount(originalId: number, subscription?: string): number { + return this.attempts.get(this.ledgerKey(originalId, subscription)) ?? 0 } peek(): readonly DeadLetterEnvelope[] { @@ -128,18 +132,30 @@ export class DeadLetterQueue { return moved } - private parked(originalId: number): DeadLetterEnvelope | undefined { - const id = this.byOriginal.get(originalId) + private ledgerKey(originalId: number, subscription?: string): string { + return `${subscription ?? ''}:${originalId}` + } + + private parked(originalId: number, subscription?: string): DeadLetterEnvelope | undefined { + const id = this.byOriginal.get(this.ledgerKey(originalId, subscription)) return id === undefined ? undefined : this.items.get(id) } private forget(envelope: DeadLetterEnvelope): void { + const key = this.ledgerOf.get(envelope.id) this.items.delete(envelope.id) - this.byOriginal.delete(envelope.originalId) - this.attempts.delete(envelope.originalId) - } - - private enqueue(message: Delivery, error: unknown, attempts: number): DeadLetterEnvelope { + this.ledgerOf.delete(envelope.id) + if (key === undefined) return + this.byOriginal.delete(key) + this.attempts.delete(key) + } + + private enqueue( + message: Delivery, + error: unknown, + attempts: number, + key: string, + ): DeadLetterEnvelope { if (this.items.size >= this.capacity) throw new DeadLetterFullError(this.capacity) const envelope: DeadLetterEnvelope = { id: this.nextId++, @@ -151,20 +167,24 @@ export class DeadLetterQueue { enqueuedAt: this.now(), } this.items.set(envelope.id, envelope) - this.byOriginal.set(message.id, envelope.id) + this.byOriginal.set(key, envelope.id) + this.ledgerOf.set(envelope.id, key) return envelope } } +let nextWrapper = 1 + export function withDeadLetter(dlq: DeadLetterQueue, handler: Handler): Handler { + const subscription = `w${nextWrapper++}` return (message) => { for (;;) { try { handler(message) - dlq.succeed(message.id) + dlq.succeed(message.id, subscription) return } catch (error) { - if (dlq.fail(message, error).status === 'dead_lettered') return + if (dlq.fail(message, error, subscription).status === 'dead_lettered') return } } } diff --git a/test/broker.test.ts b/test/broker.test.ts index 99e0649..cf73090 100644 --- a/test/broker.test.ts +++ b/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() + 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() const received: string[] = [] diff --git a/test/dead-letter.test.ts b/test/dead-letter.test.ts index 14bd6e8..f803428 100644 --- a/test/dead-letter.test.ts +++ b/test/dead-letter.test.ts @@ -56,6 +56,29 @@ describe('DeadLetterQueue.fail', () => { expect(parked.fail(msg(1), { code: 9 }).envelope?.error).toBe('[object Object]') expect(parked.fail(msg(2), 0).envelope?.error).toBe('0') }) + + it('does not share retry budget or parked envelopes across subscriptions', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 2, now: () => 7 }) + expect(dlq.fail(msg(1), new Error('A'), 'left')).toEqual({ status: 'retry', attempts: 1 }) + expect(dlq.fail(msg(1), new Error('B'), 'right')).toEqual({ status: 'retry', attempts: 1 }) + const left = dlq.fail(msg(1), new Error('A'), 'left') + const right = dlq.fail(msg(1), new Error('B'), 'right') + expect(left).toMatchObject({ + status: 'dead_lettered', + attempts: 2, + envelope: { originalId: 1, attempts: 2, error: 'A', enqueuedAt: 7 }, + }) + expect(right).toMatchObject({ + status: 'dead_lettered', + attempts: 2, + envelope: { originalId: 1, attempts: 2, error: 'B', enqueuedAt: 7 }, + }) + expect(left.envelope).not.toBe(right.envelope) + expect(dlq.size()).toBe(2) + expect(dlq.attemptCount(1, 'left')).toBe(2) + expect(dlq.attemptCount(1, 'right')).toBe(2) + expect(dlq.attemptCount(1)).toBe(0) + }) }) describe('succeed and explicit deadLetter', () => { @@ -129,6 +152,19 @@ describe('capacity, drop, purge, redrive', () => { expect(dlq.redrive(parked.id, () => {})).toBeUndefined() }) + it('redriveAll replays every envelope when publish succeeds', () => { + const dlq = new DeadLetterQueue({ maxAttempts: 1 }) + dlq.fail(msg(1, 'a'), 'x') + dlq.fail(msg(2, 'b'), 'x') + const seen: string[] = [] + expect(dlq.redriveAll((_topic, payload) => seen.push(payload)).map((e) => e.payload)).toEqual([ + 'a', + 'b', + ]) + expect(seen).toEqual(['a', 'b']) + expect(dlq.size()).toBe(0) + }) + it('redriveAll walks insertion order and stops if a publish throws', () => { const dlq = new DeadLetterQueue({ maxAttempts: 1 }) dlq.fail(msg(1, 'a'), 'x') @@ -210,6 +246,48 @@ describe('withDeadLetter', () => { expect(dlq.size()).toBe(0) }) + it('counts retries and parks an envelope per wrapper on a shared queue', () => { + const broker = new Broker() + const dlq = new DeadLetterQueue({ maxAttempts: 3 }) + const first = vi.fn(() => { + throw new Error('error A') + }) + const second = vi.fn(() => { + throw new Error('error B') + }) + broker.subscribe('orders', withDeadLetter(dlq, first)) + broker.subscribe('orders', withDeadLetter(dlq, second)) + + expect(broker.publish('orders', 'poison')).toBe(2) + expect(first).toHaveBeenCalledTimes(3) + expect(second).toHaveBeenCalledTimes(3) + expect(dlq.size()).toBe(2) + expect(dlq.peek().map((e) => e.error).sort()).toEqual(['error A', 'error B']) + expect(dlq.peek().every((e) => e.originalId === 1 && e.attempts === 3)).toBe(true) + }) + + it('still delivers to a later subscriber when a full DLQ throws', () => { + const broker = new Broker() + const dlq = new DeadLetterQueue({ maxAttempts: 1, capacity: 1 }) + const healthy = vi.fn() + broker.subscribe( + 'orders', + withDeadLetter(dlq, () => { + throw new Error('poison') + }), + ) + broker.subscribe('orders', healthy) + + expect(broker.publish('orders', 'first')).toBe(2) + expect(dlq.peek()[0]?.payload).toBe('first') + expect(healthy).toHaveBeenCalledOnce() + + expect(() => broker.publish('orders', 'second')).toThrow(DeadLetterFullError) + expect(healthy).toHaveBeenCalledTimes(2) + expect(healthy.mock.calls[1]?.[0]).toMatchObject({ payload: 'second' }) + expect(dlq.peek()[0]?.payload).toBe('first') + }) + it('surfaces DeadLetterFullError instead of parking a new poison payload', () => { const dlq = new DeadLetterQueue({ maxAttempts: 1, capacity: 1 }) const wrapped = withDeadLetter(dlq, () => {