feat: Add Bounded Queues With Backpressure Signaling

README.md

@@ -19,6 +19,9 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **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
 
+- **Bounded buffers** with a finite ready-queue `capacity` and **reject-on-full** (`QueueFullError` / `tryEnqueue`)
+- **High/low watermark backpressure** (hysteresis / Schmitt trigger) so producers pause before the wall and resume after the queue drains, without flapping in the band
+- **Producer-facing occupancy**: capacity is ready depth. Prefetch already caps in-flight. Redelivery of accepted work is allowed to sit above capacity so a nack cannot drop a message the queue already took.
 ## 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).
@@ -27,6 +30,10 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **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.
 
+- **Bounded buffers** with a finite ready-queue `capacity` and **reject-on-full** (`QueueFullError` / `tryEnqueue`)
+- **High/low watermark backpressure** (hysteresis / Schmitt trigger) so producers pause before the wall and resume after the queue drains, without flapping in the band
+- **Producer-facing occupancy**: capacity is ready depth. Prefetch already caps in-flight. Redelivery of accepted work is allowed to sit above capacity so a nack cannot drop a message the queue already took.
+- **Bounded queues with backpressure signaling.** Give `WorkQueue` a finite `capacity`. New publishes that would grow the ready backlog past that bound are rejected (`enqueue` throws `QueueFullError`, `tryEnqueue` returns `{ accepted: false }`). A `WatermarkGate` watches ready depth: occupancy at or above `highWatermark` emits `paused`, occupancy at or below `lowWatermark` emits `open`, and values in between hold the last state. Subscribe with `onBackpressure` or poll `backpressure()`. Defaults: high equals capacity, low is half the capacity (or `high - 1` when high is small). Unbounded queues stay the default so existing callers do not change.
 ## Usage
 
 ```ts

src/backpressure.ts

@@ -0,0 +1,92 @@
+export type FlowState = 'open' | 'paused'
+
+export interface BackpressureEvent {
+  readonly state: FlowState
+  readonly occupancy: number
+  readonly capacity: number
+}
+
+export type BackpressureListener = (event: BackpressureEvent) => void
+
+export class QueueFullError extends Error {
+  readonly capacity: number
+
+  constructor(capacity: number) {
+    super(`queue is full (capacity ${capacity})`)
+    this.name = 'QueueFullError'
+    this.capacity = capacity
+  }
+}
+
+export interface QueueBounds {
+  readonly capacity: number
+  readonly highWatermark: number
+  readonly lowWatermark: number
+}
+
+export interface QueueBoundOptions {
+  readonly capacity?: number
+  readonly highWatermark?: number
+  readonly lowWatermark?: number
+}
+
+export function resolveQueueBounds(options: QueueBoundOptions = {}): QueueBounds {
+  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}`)
+  }
+
+  if (capacity === Number.POSITIVE_INFINITY) {
+    if (options.highWatermark !== undefined || options.lowWatermark !== undefined) {
+      throw new Error('highWatermark and lowWatermark require a finite capacity')
+    }
+    return {
+      capacity,
+      highWatermark: Number.POSITIVE_INFINITY,
+      lowWatermark: 0,
+    }
+  }
+
+  const high = options.highWatermark ?? capacity
+  if (!Number.isInteger(high) || high < 1 || high > capacity) {
+    throw new Error(`highWatermark must be an integer in 1..capacity, got ${high}`)
+  }
+
+  const low = options.lowWatermark ?? Math.min(Math.floor(capacity / 2), Math.max(0, high - 1))
+  if (!Number.isInteger(low) || low < 0 || low > high) {
+    throw new Error(`lowWatermark must be an integer in 0..highWatermark, got ${low}`)
+  }
+
+  return { capacity, highWatermark: high, lowWatermark: low }
+}
+
+export class WatermarkGate {
+  readonly high: number
+  readonly low: number
+  private current: FlowState = 'open'
+
+  constructor(high: number, low: number) {
+    if (!Number.isFinite(high) || high < 1) {
+      throw new Error(`high watermark must be >= 1, got ${high}`)
+    }
+    if (!Number.isFinite(low) || low < 0 || low > high) {
+      throw new Error(`low watermark must be in 0..high, got ${low}`)
+    }
+    this.high = high
+    this.low = low
+  }
+
+  get state(): FlowState {
+    return this.current
+  }
+
+  observe(occupancy: number): FlowState | undefined {
+    if (!Number.isFinite(occupancy) || occupancy < 0) {
+      throw new Error(`occupancy must be a finite number >= 0, got ${occupancy}`)
+    }
+    const before = this.current
+    if (occupancy >= this.high) this.current = 'paused'
+    else if (occupancy <= this.low) this.current = 'open'
+    return this.current !== before ? this.current : undefined
+  }
+}

src/index.ts

@@ -1,26 +1,49 @@
 export { Broker } from './broker.js'
-export type { Message, Handler, Unsubscribe } 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 { WorkQueue, QueueFullError } from './work-queue.js'
+
 export type {
   WorkMessage,
   Delivery,
   ConsumerHandler,
   WorkQueueOptions,
   Unsubscribe as WorkQueueUnsubscribe,
+  EnqueueResult,
+  BackpressureEvent,
+  BackpressureListener,
+  FlowState,
 } from './work-queue.js'
+
 export { ReliableBroker } from './reliable-broker.js'
+
 export type {
   ReliableMessage,
   ReliableDelivery,
   ReliableHandler,
   ReliableBrokerOptions,
   ReliableSubscribeOptions,
 } from './reliable-broker.js'
+
+export { WatermarkGate, resolveQueueBounds } from './backpressure.js'
+
+export type {
+  QueueBounds,
+  QueueBoundOptions,
+} from './backpressure.js'

src/work-queue.ts

@@ -1,40 +1,35 @@
-/**
- * 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).
- *
- * 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.
- */
+import {
+  QueueFullError,
+  WatermarkGate,
+  resolveQueueBounds,
+  type BackpressureEvent,
+  type BackpressureListener,
+  type FlowState,
+  type QueueBoundOptions,
+} from './backpressure.js'
+
+export { QueueFullError } from './backpressure.js'
+export type { BackpressureEvent, BackpressureListener, FlowState } from './backpressure.js'
 
 export interface WorkMessage<T> {
   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<T> {
   readonly message: WorkMessage<T>
   readonly deliveryTag: number
   ack(): void
-  /** Reject the delivery. `requeue` defaults to true. */
   nack(options?: { requeue?: boolean }): void
 }
 
 export type ConsumerHandler<T> = (delivery: Delivery<T>) => 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.
-   */
+export type EnqueueResult = { readonly accepted: true; readonly id: number } | { readonly accepted: false }
+
+export interface WorkQueueOptions extends QueueBoundOptions {
   maxDeliveryCount?: number
 }
 
@@ -61,9 +56,14 @@ interface ConsumerState<T> {
 const DEFAULT_MAX_DELIVERY_COUNT = 10
 
 export class WorkQueue<T> {
+  readonly capacity: number
+  readonly highWatermark: number
+  readonly lowWatermark: number
   private readonly ready: PendingMessage<T>[] = []
   private readonly consumers: ConsumerState<T>[] = []
   private readonly inFlight = new Map<number, InFlightEntry<T>>()
+  private readonly listeners = new Set<BackpressureListener>()
+  private readonly gate: WatermarkGate | undefined
   private readonly maxDeliveryCount: number
   private nextMessageId = 1
   private nextDeliveryTag = 1
@@ -77,22 +77,45 @@ export class WorkQueue<T> {
       throw new Error(`maxDeliveryCount must be a positive integer, got ${max}`)
     }
     this.maxDeliveryCount = max
+    const bounds = resolveQueueBounds(options)
+    this.capacity = bounds.capacity
+    this.highWatermark = bounds.highWatermark
+    this.lowWatermark = bounds.lowWatermark
+    this.gate =
+      bounds.capacity === Number.POSITIVE_INFINITY
+        ? undefined
+        : new WatermarkGate(bounds.highWatermark, bounds.lowWatermark)
   }
 
-  /** Enqueue a payload. Dispatches if a consumer has spare capacity. Returns message id. */
   enqueue(payload: T): number {
+    const result = this.tryEnqueue(payload)
+    if (!result.accepted) throw new QueueFullError(this.capacity)
+    return result.id
+  }
+
+  tryEnqueue(payload: T): EnqueueResult {
+    if (this.ready.length >= this.capacity) return { accepted: false }
     const id = this.nextMessageId++
     this.ready.push({ id, payload, deliveryCount: 0 })
     this.pump()
-    return id
+    this.applyFlow()
+    return { accepted: true, id }
+  }
+
+  backpressure(): FlowState {
+    return this.gate?.state ?? 'open'
+  }
+
+  onBackpressure(listener: BackpressureListener): Unsubscribe {
+    this.listeners.add(listener)
+    let active = true
+    return () => {
+      if (!active) return
+      active = false
+      this.listeners.delete(listener)
+    }
   }
 
-  /**
-   * Register a competing consumer. Default prefetch is 1. Unsubscribe requeues
-   * 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<T>, options?: { prefetch?: number }): Unsubscribe {
     const prefetch = options?.prefetch ?? 1
     if (!Number.isInteger(prefetch) || prefetch < 1) {
@@ -108,6 +131,7 @@ export class WorkQueue<T> {
     }
     this.consumers.push(consumer)
     this.pump()
+    this.applyFlow()
 
     return () => {
       if (!consumer.active) return
@@ -126,6 +150,7 @@ export class WorkQueue<T> {
       const i = this.consumers.indexOf(consumer)
       if (i !== -1) this.consumers.splice(i, 1)
       this.pump()
+      this.applyFlow()
     }
   }
 
@@ -152,10 +177,7 @@ export class WorkQueue<T> {
     if (this.pumping) return
     this.pumping = true
     try {
-      // Cursor-based RR so sequential enqueues share fairly, not only batched pumps.
-      // 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.
+      // Cursor RR. Same id at most once per round so a sync nack cannot busy-spin.
       while (this.ready.length > 0 && this.hasConsumerCapacity()) {
         const deliveredThisRound = new Set<number>()
         let deliveredAny = false
@@ -235,11 +257,31 @@ export class WorkQueue<T> {
     const consumer = this.consumers.find((c) => c.id === entry.consumerId)
     if (consumer?.active) consumer.inFlight -= 1
 
-    // Tail, not head: avoids a poison message starving others already on ready.
-    // Past maxDeliveryCount: drop (bounded redelivery; DLQ is future work).
+    // Tail, not head: a poison nack must not starve work already on ready.
     if (requeue && entry.pending.deliveryCount < this.maxDeliveryCount) {
       this.ready.push(entry.pending)
     }
     this.pump()
+    this.applyFlow()
+  }
+
+  private applyFlow(): void {
+    if (this.pumping || !this.gate) return
+    const next = this.gate.observe(this.ready.length)
+    if (next === undefined) return
+    this.emit({ state: next, occupancy: this.ready.length, capacity: this.capacity })
+  }
+
+  private emit(event: BackpressureEvent): void {
+    const snapshot = [...this.listeners]
+    let firstError: unknown
+    for (const listener of snapshot) {
+      try {
+        listener(event)
+      } catch (error) {
+        if (firstError === undefined) firstError = error
+      }
+    }
+    if (firstError !== undefined) throw firstError
   }
 }