From 0c380ddf72e5bf7ba04d75d4cf8a5b0dabb23ec5 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Mon, 10 Aug 2026 16:59:11 +0000 Subject: [PATCH 1/2] feat: add competing-consumer work queue with acks WorkQueue delivers each message to one consumer, with prefetch, ack/nack requeue, and cancel redelivery for unacked work. --- README.md | 56 +++++----- src/index.ts | 12 +-- src/work-queue.ts | 189 ++++++++++++++++++++++++++++++++ test/work-queue.test.ts | 231 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 453 insertions(+), 35 deletions(-) create mode 100644 src/work-queue.ts create mode 100644 test/work-queue.test.ts diff --git a/README.md b/README.md index 59001f5..a2f1c3b 100644 --- a/README.md +++ b/README.md @@ -1,27 +1,26 @@ # event-broker-lab -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. +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. ## 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, 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. +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. ## 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 +- **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 +- **Prefetch** as a per-consumer in-flight limit for fair sharing +- **Redelivery counting** and poison-message safe requeue (tail, not head) ## 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. +- **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). Prefetch defaults to 1 so peers share fairly; cancelling a consumer requeues its unacked deliveries for the remaining workers. ## Usage @@ -56,30 +55,29 @@ 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: +Competing consumers on a work queue (one message, one worker): ```ts -import { Broker, DeadLetterQueue, withDeadLetter } from 'event-broker-lab' - -const broker = new Broker() -const dlq = new DeadLetterQueue({ 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) +import { WorkQueue } from 'event-broker-lab' + +const queue = new WorkQueue<{ jobId: string }>() + +queue.consume((delivery) => { + const { jobId } = delivery.message.payload + try { + // do the work + delivery.ack() + } catch { + delivery.nack() // requeue for another consumer + } }) + +queue.consume((delivery) => { + // second worker competes for the same queue + delivery.ack() +}) + +queue.enqueue({ jobId: 'job-1' }) // only one of the two consumers receives it ``` ## Running the tests diff --git a/src/index.ts b/src/index.ts index 0b4294f..36f9812 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,10 +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 { WorkQueue } from './work-queue.js' export type { - DeadLetterEnvelope, - DeadLetterQueueOptions, - FailResult, - FailStatus, -} from './dead-letter.js' + WorkMessage, + Delivery, + ConsumerHandler, + Unsubscribe as WorkQueueUnsubscribe, +} from './work-queue.js' diff --git a/src/work-queue.ts b/src/work-queue.ts new file mode 100644 index 0000000..8051913 --- /dev/null +++ b/src/work-queue.ts @@ -0,0 +1,189 @@ +/** + * Competing-consumer work queue with explicit acknowledgements. + * + * Unlike topic fan-out, each enqueued message is delivered to exactly one + * consumer. Consumers ack on success or nack to requeue; unacked work held by + * a leaving consumer is requeued so peers can take it (RabbitMQ / SQS model). + */ + +export interface WorkMessage { + readonly id: number + readonly payload: T + /** Times this message has been handed to a consumer (1 on first delivery). */ + readonly deliveryCount: number +} + +export interface Delivery { + readonly message: WorkMessage + readonly deliveryTag: number + ack(): void + /** Reject the delivery. `requeue` defaults to true. */ + nack(options?: { requeue?: boolean }): void +} + +export type ConsumerHandler = (delivery: Delivery) => void +export type Unsubscribe = () => void + +interface PendingMessage { + id: number + payload: T + deliveryCount: number +} + +interface InFlightEntry { + pending: PendingMessage + consumerId: number + settled: boolean +} + +interface ConsumerState { + id: number + handler: ConsumerHandler + prefetch: number + inFlight: number + active: boolean +} + +export class WorkQueue { + private readonly ready: PendingMessage[] = [] + private readonly consumers: ConsumerState[] = [] + private readonly inFlight = new Map>() + private nextMessageId = 1 + private nextDeliveryTag = 1 + private nextConsumerId = 1 + private nextConsumerIndex = 0 + private pumping = false + + /** Enqueue a payload. Dispatches if a consumer has spare capacity. Returns message id. */ + enqueue(payload: T): number { + const id = this.nextMessageId++ + this.ready.push({ id, payload, deliveryCount: 0 }) + this.pump() + return id + } + + /** + * Register a competing consumer. Default prefetch is 1. Unsubscribe requeues + * any still-unacked deliveries held by this consumer. + */ + consume(handler: ConsumerHandler, options?: { prefetch?: number }): Unsubscribe { + const prefetch = options?.prefetch ?? 1 + if (!Number.isInteger(prefetch) || prefetch < 1) { + throw new Error(`prefetch must be a positive integer, got ${prefetch}`) + } + + const consumer: ConsumerState = { + id: this.nextConsumerId++, + handler, + prefetch, + inFlight: 0, + active: true, + } + this.consumers.push(consumer) + this.pump() + + return () => { + if (!consumer.active) return + consumer.active = false + + const toRequeue: PendingMessage[] = [] + for (const [tag, entry] of this.inFlight) { + if (entry.consumerId !== consumer.id || entry.settled) continue + entry.settled = true + this.inFlight.delete(tag) + toRequeue.push(entry.pending) + } + this.ready.unshift(...toRequeue) + consumer.inFlight = 0 + + const i = this.consumers.indexOf(consumer) + if (i !== -1) this.consumers.splice(i, 1) + this.pump() + } + } + + readyCount(): number { + return this.ready.length + } + + inFlightCount(): number { + return this.inFlight.size + } + + consumerCount(): number { + return this.consumers.length + } + + private pump(): void { + if (this.pumping) return + this.pumping = true + try { + // Cursor-based RR so sequential enqueues share fairly, not only batched pumps. + while (this.ready.length > 0 && this.consumers.length > 0) { + const n = this.consumers.length + const start = this.nextConsumerIndex % n + let delivered = false + for (let offset = 0; offset < n; offset++) { + const i = (start + offset) % n + const consumer = this.consumers[i] + if (!consumer || !consumer.active || consumer.inFlight >= consumer.prefetch) { + continue + } + const pending = this.ready.shift() + if (!pending) return + this.deliver(consumer, pending) + this.nextConsumerIndex = (i + 1) % n + delivered = true + break + } + if (!delivered) break + } + } finally { + this.pumping = false + } + } + + private deliver(consumer: ConsumerState, pending: PendingMessage): void { + pending.deliveryCount += 1 + const deliveryTag = this.nextDeliveryTag++ + const entry: InFlightEntry = { + pending, + consumerId: consumer.id, + settled: false, + } + this.inFlight.set(deliveryTag, entry) + consumer.inFlight += 1 + + const delivery: Delivery = { + message: { + id: pending.id, + payload: pending.payload, + deliveryCount: pending.deliveryCount, + }, + deliveryTag, + ack: () => this.settle(deliveryTag, false), + nack: (options) => this.settle(deliveryTag, options?.requeue !== false), + } + + try { + consumer.handler(delivery) + } catch { + // Uncaught handler error must not leave the message stranded in-flight. + 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 consumer = this.consumers.find((c) => c.id === entry.consumerId) + if (consumer?.active) consumer.inFlight -= 1 + + // Tail, not head: avoids a poison message tight-looping on one consumer. + if (requeue) this.ready.push(entry.pending) + this.pump() + } +} diff --git a/test/work-queue.test.ts b/test/work-queue.test.ts new file mode 100644 index 0000000..7900ceb --- /dev/null +++ b/test/work-queue.test.ts @@ -0,0 +1,231 @@ +import { describe, it, expect } from 'vitest' +import { WorkQueue, type Delivery } from '../src/index.js' + +describe('WorkQueue competing consumers', () => { + it('delivers each message to exactly one consumer and round-robins', () => { + const queue = new WorkQueue() + const a: string[] = [] + const b: string[] = [] + queue.consume((d) => { + a.push(d.message.payload) + d.ack() + }) + queue.consume((d) => { + b.push(d.message.payload) + d.ack() + }) + + for (const p of ['one', 'two', 'three', 'four']) queue.enqueue(p) + + expect([...a, ...b].sort()).toEqual(['four', 'one', 'three', 'two']) + expect(a).toEqual(['one', 'three']) + expect(b).toEqual(['two', 'four']) + expect(queue.readyCount()).toBe(0) + expect(queue.inFlightCount()).toBe(0) + }) + + it('buffers when no consumer is registered, then drains on consume', () => { + const queue = new WorkQueue() + expect(queue.enqueue(1)).toBe(1) + expect(queue.enqueue(2)).toBe(2) + expect(queue.readyCount()).toBe(2) + + const seen: number[] = [] + queue.consume((d) => { + seen.push(d.message.payload) + d.ack() + }) + expect(seen).toEqual([1, 2]) + }) + + it('assigns monotonic message ids starting at 1', () => { + const queue = new WorkQueue() + const ids: number[] = [] + queue.consume((d) => { + ids.push(d.message.id) + d.ack() + }) + queue.enqueue('a') + queue.enqueue('b') + expect(ids).toEqual([1, 2]) + }) + + it('respects prefetch: unacked work blocks further delivery to that consumer', () => { + const queue = new WorkQueue() + const held: Delivery[] = [] + queue.consume((d) => held.push(d)) + queue.enqueue('first') + queue.enqueue('second') + expect(held).toHaveLength(1) + expect(held[0]?.message.payload).toBe('first') + expect(queue.readyCount()).toBe(1) + + held[0]?.ack() + expect(held).toHaveLength(2) + expect(held[1]?.message.payload).toBe('second') + + const multi: Delivery[] = [] + const q2 = new WorkQueue() + q2.consume((d) => multi.push(d), { prefetch: 3 }) + q2.enqueue('a') + q2.enqueue('b') + q2.enqueue('c') + q2.enqueue('d') + expect(multi).toHaveLength(3) + expect(q2.readyCount()).toBe(1) + multi[0]?.ack() + expect(multi).toHaveLength(4) + }) + + it('throws on invalid prefetch', () => { + const queue = new WorkQueue() + expect(() => queue.consume(() => {}, { prefetch: 0 })).toThrow(/prefetch/) + expect(() => queue.consume(() => {}, { prefetch: 1.5 })).toThrow(/prefetch/) + }) + + it('shares work when one consumer is blocked at prefetch', () => { + const queue = new WorkQueue() + const aHeld: Delivery[] = [] + const bSeen: number[] = [] + queue.consume((d) => aHeld.push(d)) + queue.consume((d) => { + bSeen.push(d.message.payload) + d.ack() + }) + queue.enqueue(1) + queue.enqueue(2) + queue.enqueue(3) + expect(aHeld[0]?.message.payload).toBe(1) + expect(bSeen).toEqual([2, 3]) + aHeld[0]?.ack() + expect(queue.inFlightCount()).toBe(0) + }) +}) + +describe('ack and nack', () => { + it('ack settles permanently; nack requeues with rising deliveryCount', () => { + const queue = new WorkQueue() + const counts: number[] = [] + queue.consume((d) => { + counts.push(d.message.deliveryCount) + if (d.message.deliveryCount < 3) d.nack() + else d.ack() + }) + queue.enqueue('retry') + expect(counts).toEqual([1, 2, 3]) + expect(queue.readyCount()).toBe(0) + expect(queue.inFlightCount()).toBe(0) + }) + + it('nack({ requeue: false }) drops the message', () => { + const queue = new WorkQueue() + const seen: string[] = [] + queue.consume((d) => { + seen.push(d.message.payload) + d.nack({ requeue: false }) + }) + queue.enqueue('poison') + expect(seen).toEqual(['poison']) + expect(queue.readyCount()).toBe(0) + }) + + it('double settle is a no-op', () => { + const queue = new WorkQueue() + let first: Delivery | undefined + let count = 0 + queue.consume((d) => { + count += 1 + if (count === 1) { + first = d + d.nack() + return + } + d.ack() + }) + queue.enqueue('y') + expect(() => first?.ack()).not.toThrow() + expect(() => first?.nack()).not.toThrow() + expect(count).toBe(2) + }) + + it('requeues on handler throw so work is not stranded', () => { + const queue = new WorkQueue() + let throws = true + const seen: string[] = [] + queue.consume((d) => { + seen.push(d.message.payload) + if (throws) { + throws = false + throw new Error('boom') + } + d.ack() + }) + queue.enqueue('recover') + expect(seen).toEqual(['recover', 'recover']) + expect(queue.inFlightCount()).toBe(0) + }) + + it('keeps delivery tags unique across redeliveries', () => { + const queue = new WorkQueue() + const tags: number[] = [] + queue.consume((d) => { + tags.push(d.deliveryTag) + if (tags.length < 3) d.nack() + else d.ack() + }) + queue.enqueue('z') + expect(tags).toEqual([1, 2, 3]) + }) +}) + +describe('consumer lifecycle', () => { + it('requeues unacked work on unsubscribe; stale ack is harmless', () => { + const queue = new WorkQueue() + let stolen: Delivery | undefined + const off = queue.consume((d) => { + stolen = d + }) + queue.enqueue('held') + expect(queue.inFlightCount()).toBe(1) + + off() + expect(() => off()).not.toThrow() + expect(queue.consumerCount()).toBe(0) + expect(queue.readyCount()).toBe(1) + + const recovered: string[] = [] + queue.consume((d) => { + recovered.push(d.message.payload) + expect(d.message.deliveryCount).toBe(2) + d.ack() + }) + expect(recovered).toEqual(['held']) + expect(() => stolen?.ack()).not.toThrow() + }) + + it('handles enqueue-from-handler without dropping messages', () => { + const queue = new WorkQueue() + const seen: string[] = [] + queue.consume((d) => { + seen.push(d.message.payload) + if (d.message.payload === 'seed') queue.enqueue('child') + d.ack() + }) + queue.enqueue('seed') + expect(seen).toEqual(['seed', 'child']) + }) + + it('round-robins evenly across three consumers', () => { + const queue = new WorkQueue() + const counts = [0, 0, 0] + for (let i = 0; i < 3; i++) { + const slot = i + queue.consume((d) => { + counts[slot] = (counts[slot] ?? 0) + 1 + d.ack() + }) + } + for (let n = 0; n < 9; n++) queue.enqueue(n) + expect(counts).toEqual([3, 3, 3]) + }) +}) From e754456cdfe1cbad2753a5905965bfdc6e5b6644 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Mon, 10 Aug 2026 17:08:34 +0000 Subject: [PATCH 2/2] fix: bound work-queue redelivery so nack/throw cannot hang enqueue Deliver each message id at most once per pump round and drop after maxDeliveryCount. Adds hang and nack-to-tail regression tests. --- README.md | 4 +- src/index.ts | 1 + src/work-queue.ts | 94 ++++++++++++++++++++++++++++++++--------- test/work-queue.test.ts | 68 +++++++++++++++++++++++++++++ 4 files changed, 146 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index a2f1c3b..4583710 100644 --- a/README.md +++ b/README.md @@ -14,13 +14,13 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r - **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 - **Prefetch** as a per-consumer in-flight limit for fair sharing -- **Redelivery counting** and poison-message safe requeue (tail, not head) +- **Redelivery counting**, **bounded redelivery** (`maxDeliveryCount`, default 10; excess drops), and **tail requeue** on nack so other ready work is not starved ## 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). Prefetch defaults to 1 so peers share fairly; cancelling a consumer requeues its unacked deliveries for the remaining workers. +- **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`. ## Usage diff --git a/src/index.ts b/src/index.ts index 36f9812..db282a2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,5 +6,6 @@ export type { WorkMessage, Delivery, ConsumerHandler, + WorkQueueOptions, Unsubscribe as WorkQueueUnsubscribe, } from './work-queue.js' diff --git a/src/work-queue.ts b/src/work-queue.ts index 8051913..64ee2e8 100644 --- a/src/work-queue.ts +++ b/src/work-queue.ts @@ -4,6 +4,11 @@ * Unlike topic fan-out, each enqueued message is delivered to exactly one * consumer. Consumers ack on success or nack to requeue; unacked work held by * a leaving consumer is requeued so peers can take it (RabbitMQ / SQS model). + * + * Handler throw is treated as nack({ requeue: true }) so work is not stranded. + * Redelivery is bounded by maxDeliveryCount (default 10); excess drops the + * message (DLQ is future work). Each message id is delivered at most once per + * pump round so a permanent nack/throw cannot busy-spin inside enqueue. */ export interface WorkMessage { @@ -24,6 +29,15 @@ export interface Delivery { export type ConsumerHandler = (delivery: Delivery) => void export type Unsubscribe = () => void +export interface WorkQueueOptions { + /** + * After this many deliveries, a requeueing nack/throw drops the message + * instead of putting it back on the ready queue. Must be a positive integer. + * Default 10. + */ + maxDeliveryCount?: number +} + interface PendingMessage { id: number payload: T @@ -44,16 +58,27 @@ interface ConsumerState { active: boolean } +const DEFAULT_MAX_DELIVERY_COUNT = 10 + export class WorkQueue { private readonly ready: PendingMessage[] = [] private readonly consumers: ConsumerState[] = [] private readonly inFlight = new Map>() + private readonly maxDeliveryCount: number private nextMessageId = 1 private nextDeliveryTag = 1 private nextConsumerId = 1 private nextConsumerIndex = 0 private pumping = false + constructor(options?: WorkQueueOptions) { + 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 + } + /** Enqueue a payload. Dispatches if a consumer has spare capacity. Returns message id. */ enqueue(payload: T): number { const id = this.nextMessageId++ @@ -64,7 +89,9 @@ export class WorkQueue { /** * Register a competing consumer. Default prefetch is 1. Unsubscribe requeues - * any still-unacked deliveries held by this consumer. + * any still-unacked deliveries held by this consumer at the head of ready + * (prompt recovery). By contrast, nack({ requeue: true }) pushes to the tail + * so a poison message does not starve others already waiting. */ consume(handler: ConsumerHandler, options?: { prefetch?: number }): Unsubscribe { const prefetch = options?.prefetch ?? 1 @@ -114,29 +141,55 @@ export class WorkQueue { return this.consumers.length } + private hasConsumerCapacity(): boolean { + for (const c of this.consumers) { + if (c.active && c.inFlight < c.prefetch) return true + } + return false + } + private pump(): void { if (this.pumping) return this.pumping = true try { // Cursor-based RR so sequential enqueues share fairly, not only batched pumps. - while (this.ready.length > 0 && this.consumers.length > 0) { - const n = this.consumers.length - const start = this.nextConsumerIndex % n - let delivered = false - for (let offset = 0; offset < n; offset++) { - const i = (start + offset) % n - const consumer = this.consumers[i] - if (!consumer || !consumer.active || consumer.inFlight >= consumer.prefetch) { - continue + // Each message id is delivered at most once per round. A sync nack/throw that + // requeues cannot re-enter the same id inside that round (no busy-spin). A new + // round retries requeued work until idle or maxDeliveryCount drops it. + while (this.ready.length > 0 && this.hasConsumerCapacity()) { + const deliveredThisRound = new Set() + let deliveredAny = false + + while (this.ready.length > 0 && this.consumers.length > 0) { + const head = this.ready[0] + if (!head || deliveredThisRound.has(head.id)) break + + const n = this.consumers.length + const start = this.nextConsumerIndex % n + let target: ConsumerState | undefined + let targetIndex = -1 + for (let offset = 0; offset < n; offset++) { + const i = (start + offset) % n + const consumer = this.consumers[i] + if (!consumer || !consumer.active || consumer.inFlight >= consumer.prefetch) { + continue + } + target = consumer + targetIndex = i + break } + if (!target || targetIndex < 0) break + const pending = this.ready.shift() - if (!pending) return - this.deliver(consumer, pending) - this.nextConsumerIndex = (i + 1) % n - delivered = true - break + if (!pending) break + deliveredThisRound.add(pending.id) + this.deliver(target, pending) + const len = this.consumers.length + this.nextConsumerIndex = len > 0 ? (targetIndex + 1) % len : 0 + deliveredAny = true } - if (!delivered) break + + if (!deliveredAny) break } } finally { this.pumping = false @@ -168,7 +221,7 @@ export class WorkQueue { try { consumer.handler(delivery) } catch { - // Uncaught handler error must not leave the message stranded in-flight. + // Throw ≡ nack({ requeue: true }) so uncaught handler errors do not strand work. if (!entry.settled) this.settle(deliveryTag, true) } } @@ -182,8 +235,11 @@ export class WorkQueue { const consumer = this.consumers.find((c) => c.id === entry.consumerId) if (consumer?.active) consumer.inFlight -= 1 - // Tail, not head: avoids a poison message tight-looping on one consumer. - if (requeue) this.ready.push(entry.pending) + // Tail, not head: avoids a poison message starving others already on ready. + // Past maxDeliveryCount: drop (bounded redelivery; DLQ is future work). + if (requeue && entry.pending.deliveryCount < this.maxDeliveryCount) { + this.ready.push(entry.pending) + } this.pump() } } diff --git a/test/work-queue.test.ts b/test/work-queue.test.ts index 7900ceb..37c8aa7 100644 --- a/test/work-queue.test.ts +++ b/test/work-queue.test.ts @@ -2,6 +2,11 @@ import { describe, it, expect } from 'vitest' import { WorkQueue, type Delivery } from '../src/index.js' describe('WorkQueue competing consumers', () => { + it('throws on invalid maxDeliveryCount', () => { + expect(() => new WorkQueue({ maxDeliveryCount: 0 })).toThrow(/maxDeliveryCount/) + expect(() => new WorkQueue({ maxDeliveryCount: 1.5 })).toThrow(/maxDeliveryCount/) + }) + it('delivers each message to exactly one consumer and round-robins', () => { const queue = new WorkQueue() const a: string[] = [] @@ -176,6 +181,69 @@ describe('ack and nack', () => { queue.enqueue('z') expect(tags).toEqual([1, 2, 3]) }) + + it('nack requeues to the tail when other messages are already ready', () => { + const queue = new WorkQueue() + const order: string[] = [] + let holdA: Delivery | undefined + queue.consume((d) => { + if (d.message.payload === 'A' && !holdA) { + holdA = d + return + } + order.push(d.message.payload) + d.ack() + }) + queue.enqueue('A') + expect(holdA).toBeDefined() + expect(queue.readyCount()).toBe(0) + queue.enqueue('B') + expect(queue.readyCount()).toBe(1) + holdA!.nack() + expect(order).toEqual(['B', 'A']) + expect(queue.readyCount()).toBe(0) + expect(queue.inFlightCount()).toBe(0) + }) + + it('always-nack does not hang: enqueue returns and deliveries stay bounded', () => { + const maxDeliveryCount = 5 + const queue = new WorkQueue({ maxDeliveryCount }) + let deliveries = 0 + queue.consume((d) => { + deliveries += 1 + d.nack() + }) + const id = queue.enqueue('spin') + expect(id).toBe(1) + expect(deliveries).toBe(maxDeliveryCount) + expect(queue.readyCount()).toBe(0) + expect(queue.inFlightCount()).toBe(0) + expect(queue.enqueue('still-responsive')).toBe(2) + }) + + it('always-throw / poison parse does not hang: bounded redelivery, enqueue returns', () => { + const maxDeliveryCount = 4 + const queue = new WorkQueue({ maxDeliveryCount }) + let deliveries = 0 + const seen: string[] = [] + queue.consume((d) => { + deliveries += 1 + seen.push(d.message.payload) + // Permanent poison: throw every time on this payload (e.g. bad JSON). + if (d.message.payload === 'not-json') { + JSON.parse(d.message.payload) + } + d.ack() + }) + const id = queue.enqueue('not-json') + expect(id).toBe(1) + expect(deliveries).toBe(maxDeliveryCount) + expect(queue.readyCount()).toBe(0) + expect(queue.inFlightCount()).toBe(0) + expect(queue.enqueue('ok-after')).toBe(2) + expect(seen).toContain('ok-after') + expect(queue.inFlightCount()).toBe(0) + }) }) describe('consumer lifecycle', () => {