diff --git a/README.md b/README.md
index aec535b..b63e153 100644
--- a/README.md
+++ b/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
diff --git a/src/backpressure.ts b/src/backpressure.ts
new file mode 100644
index 0000000..71801bb
--- /dev/null
+++ b/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
+  }
+}
diff --git a/src/index.ts b/src/index.ts
index eb2a8a9..3d5bf16 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -1,22 +1,38 @@
 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,
@@ -24,3 +40,10 @@ export type {
   ReliableBrokerOptions,
   ReliableSubscribeOptions,
 } from './reliable-broker.js'
+
+export { WatermarkGate, resolveQueueBounds } from './backpressure.js'
+
+export type {
+  QueueBounds,
+  QueueBoundOptions,
+} from './backpressure.js'
diff --git a/src/work-queue.ts b/src/work-queue.ts
index 64ee2e8..6e95667 100644
--- a/src/work-queue.ts
+++ b/src/work-queue.ts
@@ -1,20 +1,19 @@
-/**
- * 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
 }
 
@@ -22,19 +21,15 @@ 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
   }
 }
diff --git a/test/backpressure.test.ts b/test/backpressure.test.ts
new file mode 100644
index 0000000..07c6eb2
--- /dev/null
+++ b/test/backpressure.test.ts
@@ -0,0 +1,100 @@
+import { describe, it, expect } from 'vitest'
+import { WatermarkGate, resolveQueueBounds, QueueFullError } from '../src/index.js'
+
+describe('resolveQueueBounds', () => {
+  it('defaults to an unbounded ready queue', () => {
+    expect(resolveQueueBounds()).toEqual({
+      capacity: Number.POSITIVE_INFINITY,
+      highWatermark: Number.POSITIVE_INFINITY,
+      lowWatermark: 0,
+    })
+  })
+
+  it('defaults high to capacity and low to half, with hysteresis when possible', () => {
+    expect(resolveQueueBounds({ capacity: 8 })).toEqual({
+      capacity: 8,
+      highWatermark: 8,
+      lowWatermark: 4,
+    })
+    expect(resolveQueueBounds({ capacity: 1 })).toEqual({
+      capacity: 1,
+      highWatermark: 1,
+      lowWatermark: 0,
+    })
+    expect(resolveQueueBounds({ capacity: 10, highWatermark: 3 })).toEqual({
+      capacity: 10,
+      highWatermark: 3,
+      lowWatermark: 2,
+    })
+  })
+
+  it('rejects invalid capacity and watermark combinations', () => {
+    expect(() => resolveQueueBounds({ capacity: 0 })).toThrow(/capacity/)
+    expect(() => resolveQueueBounds({ capacity: 1.5 })).toThrow(/capacity/)
+    expect(() => resolveQueueBounds({ highWatermark: 1 })).toThrow(/require a finite capacity/)
+    expect(() => resolveQueueBounds({ capacity: 4, highWatermark: 5 })).toThrow(/highWatermark/)
+    expect(() => resolveQueueBounds({ capacity: 4, highWatermark: 2, lowWatermark: 3 })).toThrow(
+      /lowWatermark/,
+    )
+    expect(() => resolveQueueBounds({ capacity: 4, lowWatermark: -1 })).toThrow(/lowWatermark/)
+  })
+})
+
+describe('WatermarkGate hysteresis', () => {
+  it('starts open and only emits on transitions', () => {
+    const gate = new WatermarkGate(3, 1)
+    expect(gate.state).toBe('open')
+    expect(gate.observe(0)).toBeUndefined()
+    expect(gate.observe(2)).toBeUndefined()
+    expect(gate.observe(3)).toBe('paused')
+    expect(gate.observe(3)).toBeUndefined()
+    expect(gate.observe(2)).toBeUndefined()
+    expect(gate.observe(1)).toBe('open')
+    expect(gate.observe(1)).toBeUndefined()
+  })
+
+  it('holds the current state inside the band between low and high', () => {
+    const gate = new WatermarkGate(4, 1)
+    expect(gate.observe(4)).toBe('paused')
+    expect(gate.observe(3)).toBeUndefined()
+    expect(gate.observe(2)).toBeUndefined()
+    expect(gate.state).toBe('paused')
+    expect(gate.observe(1)).toBe('open')
+    expect(gate.observe(2)).toBeUndefined()
+    expect(gate.observe(3)).toBeUndefined()
+    expect(gate.state).toBe('open')
+  })
+
+  it('does not flap when high equals low', () => {
+    const gate = new WatermarkGate(2, 2)
+    expect(gate.observe(2)).toBe('paused')
+    expect(gate.observe(2)).toBeUndefined()
+    expect(gate.observe(1)).toBe('open')
+    expect(gate.observe(1)).toBeUndefined()
+    expect(gate.observe(2)).toBe('paused')
+  })
+
+  it('handles occupancy jumps that skip the band', () => {
+    const gate = new WatermarkGate(5, 2)
+    expect(gate.observe(9)).toBe('paused')
+    expect(gate.observe(0)).toBe('open')
+  })
+
+  it('rejects inverted watermarks and negative occupancy', () => {
+    expect(() => new WatermarkGate(0, 0)).toThrow(/high watermark/)
+    expect(() => new WatermarkGate(2, 3)).toThrow(/low watermark/)
+    const gate = new WatermarkGate(2, 1)
+    expect(() => gate.observe(-1)).toThrow(/occupancy/)
+    expect(() => gate.observe(Number.NaN)).toThrow(/occupancy/)
+  })
+})
+
+describe('QueueFullError', () => {
+  it('names the capacity that rejected the producer', () => {
+    const error = new QueueFullError(4)
+    expect(error).toBeInstanceOf(Error)
+    expect(error.name).toBe('QueueFullError')
+    expect(error.capacity).toBe(4)
+    expect(error.message).toMatch(/capacity 4/)
+  })
+})
diff --git a/test/work-queue-backpressure.test.ts b/test/work-queue-backpressure.test.ts
new file mode 100644
index 0000000..7e4afcf
--- /dev/null
+++ b/test/work-queue-backpressure.test.ts
@@ -0,0 +1,204 @@
+import { describe, it, expect } from 'vitest'
+import { QueueFullError, WorkQueue, type BackpressureEvent, type Delivery } from '../src/index.js'
+
+describe('WorkQueue bounded ready backlog', () => {
+  it('throws QueueFullError once ready depth hits capacity', () => {
+    const queue = new WorkQueue<string>({ capacity: 2 })
+    expect(queue.enqueue('a')).toBe(1)
+    expect(queue.enqueue('b')).toBe(2)
+    expect(queue.readyCount()).toBe(2)
+    expect(() => queue.enqueue('c')).toThrow(QueueFullError)
+    expect(() => queue.enqueue('c')).toThrow(/capacity 2/)
+    expect(queue.tryEnqueue('c')).toEqual({ accepted: false })
+    expect(queue.readyCount()).toBe(2)
+  })
+
+  it('does not spend a message id on a rejected enqueue', () => {
+    const queue = new WorkQueue<string>({ capacity: 1 })
+    expect(queue.enqueue('held')).toBe(1)
+    expect(queue.tryEnqueue('nope')).toEqual({ accepted: false })
+    const seen: number[] = []
+    queue.consume((d) => {
+      seen.push(d.message.id)
+      d.ack()
+    })
+    expect(seen).toEqual([1])
+    expect(queue.enqueue('after')).toBe(2)
+  })
+
+  it('accepts again after consumers drain below capacity', () => {
+    const queue = new WorkQueue<number>({ capacity: 1 })
+    queue.enqueue(1)
+    expect(queue.tryEnqueue(2).accepted).toBe(false)
+    const seen: number[] = []
+    queue.consume((d) => {
+      seen.push(d.message.payload)
+      d.ack()
+    })
+    expect(seen).toEqual([1])
+    expect(queue.enqueue(2)).toBe(2)
+    expect(seen).toEqual([1, 2])
+  })
+
+  it('counts only ready depth, not in-flight deliveries', () => {
+    const queue = new WorkQueue<string>({ capacity: 1 })
+    const held: Delivery<string>[] = []
+    queue.consume((d) => held.push(d))
+    expect(queue.enqueue('in-flight')).toBe(1)
+    expect(queue.readyCount()).toBe(0)
+    expect(queue.inFlightCount()).toBe(1)
+    expect(queue.enqueue('ready')).toBe(2)
+    expect(queue.readyCount()).toBe(1)
+    expect(queue.tryEnqueue('over')).toEqual({ accepted: false })
+    held[0]?.ack()
+    expect(queue.readyCount()).toBe(0)
+    expect(queue.enqueue('after-ack')).toBe(3)
+  })
+
+  it('lets redelivery exceed capacity so accepted work is not dropped', () => {
+    const queue = new WorkQueue<string>({ capacity: 1 })
+    const held: Delivery<string>[] = []
+    const off = queue.consume((d) => held.push(d), { prefetch: 2 })
+    queue.enqueue('a')
+    queue.enqueue('b')
+    expect(queue.inFlightCount()).toBe(2)
+    expect(queue.readyCount()).toBe(0)
+    off()
+    expect(queue.readyCount()).toBe(2)
+    expect(queue.readyCount()).toBeGreaterThan(queue.capacity)
+    expect(queue.tryEnqueue('c')).toEqual({ accepted: false })
+    expect(queue.backpressure()).toBe('paused')
+  })
+})
+
+describe('WorkQueue backpressure signaling', () => {
+  it('stays open when a consumer drains as fast as we enqueue', () => {
+    const queue = new WorkQueue<string>({ capacity: 2, highWatermark: 2, lowWatermark: 1 })
+    const events: BackpressureEvent[] = []
+    queue.onBackpressure((event) => events.push(event))
+    queue.consume((d) => d.ack())
+    queue.enqueue('a')
+    queue.enqueue('b')
+    queue.enqueue('c')
+    expect(queue.backpressure()).toBe('open')
+    expect(events).toEqual([])
+  })
+
+  it('pauses at high watermark and resumes at or below low, without flapping in the band', () => {
+    const queue = new WorkQueue<string>({ capacity: 4, highWatermark: 3, lowWatermark: 1 })
+    const events: Pick<BackpressureEvent, 'state' | 'occupancy'>[] = []
+    queue.onBackpressure((event) => events.push({ state: event.state, occupancy: event.occupancy }))
+
+    queue.enqueue('1')
+    queue.enqueue('2')
+    expect(queue.backpressure()).toBe('open')
+    expect(events).toEqual([])
+
+    queue.enqueue('3')
+    expect(queue.backpressure()).toBe('paused')
+    expect(events).toEqual([{ state: 'paused', occupancy: 3 }])
+
+    queue.enqueue('4')
+    expect(queue.tryEnqueue('5')).toEqual({ accepted: false })
+    expect(events).toEqual([{ state: 'paused', occupancy: 3 }])
+
+    const held: Delivery<string>[] = []
+    queue.consume((d) => held.push(d))
+    expect(queue.readyCount()).toBe(3)
+    expect(queue.backpressure()).toBe('paused')
+
+    held[0]?.ack()
+    expect(queue.readyCount()).toBe(2)
+    expect(queue.backpressure()).toBe('paused')
+    held[1]?.ack()
+    expect(queue.readyCount()).toBe(1)
+    expect(queue.backpressure()).toBe('open')
+    expect(events).toEqual([
+      { state: 'paused', occupancy: 3 },
+      { state: 'open', occupancy: 1 },
+    ])
+  })
+
+  it('unsubscribing a listener stops further events', () => {
+    const queue = new WorkQueue<string>({ capacity: 2 })
+    const seen: string[] = []
+    const off = queue.onBackpressure((event) => seen.push(event.state))
+    queue.enqueue('a')
+    queue.enqueue('b')
+    expect(seen).toEqual(['paused'])
+    off()
+    off()
+    const next: string[] = []
+    queue.onBackpressure((event) => next.push(event.state))
+    queue.consume((d) => d.ack())
+    expect(seen).toEqual(['paused'])
+    expect(next).toEqual(['open'])
+  })
+
+  it('delivers the first listener error after the rest of the snapshot runs', () => {
+    const queue = new WorkQueue<string>({ capacity: 1 })
+    const order: string[] = []
+    queue.onBackpressure(() => {
+      order.push('a')
+      throw new Error('listener-a')
+    })
+    queue.onBackpressure(() => {
+      order.push('b')
+    })
+    expect(() => queue.enqueue('full')).toThrow(/listener-a/)
+    expect(order).toEqual(['a', 'b'])
+    expect(queue.readyCount()).toBe(1)
+    expect(queue.backpressure()).toBe('paused')
+  })
+
+  it('surfaces a listener error when consume drains a paused queue with sync ack', () => {
+    const queue = new WorkQueue<string>({ capacity: 4, highWatermark: 3, lowWatermark: 1 })
+    queue.enqueue('1')
+    queue.enqueue('2')
+    queue.enqueue('3')
+    queue.enqueue('4')
+    expect(queue.backpressure()).toBe('paused')
+    expect(queue.readyCount()).toBe(4)
+
+    const delivered: string[] = []
+    queue.onBackpressure(() => {
+      throw new Error('listener-boom')
+    })
+
+    expect(() => {
+      queue.consume((d) => {
+        delivered.push(d.message.payload)
+        d.ack()
+      })
+    }).toThrow(/listener-boom/)
+
+    expect(delivered).toEqual(['1', '2', '3', '4'])
+    expect(queue.readyCount()).toBe(0)
+    expect(queue.backpressure()).toBe('open')
+  })
+
+  it('does not reject a nack requeue when ready is already at capacity', () => {
+    const queue = new WorkQueue<string>({ capacity: 1 })
+    const events: string[] = []
+    queue.onBackpressure((event) => events.push(event.state))
+    let first: Delivery<string> | undefined
+    const seen: string[] = []
+    queue.consume((d) => {
+      if (!first) {
+        first = d
+        return
+      }
+      seen.push(d.message.payload)
+      d.ack()
+    })
+    queue.enqueue('held')
+    queue.enqueue('waiting')
+    expect(queue.readyCount()).toBe(1)
+    expect(queue.backpressure()).toBe('paused')
+    expect(() => first!.nack()).not.toThrow()
+    expect(seen).toEqual(['waiting', 'held'])
+    expect(queue.readyCount()).toBe(0)
+    expect(queue.backpressure()).toBe('open')
+    expect(events).toEqual(['paused', 'open'])
+  })
+})