feat: Add Exponential Backoff With Jitter on Retry

README.md

@@ -35,6 +35,10 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **Partitioned log** with **key-based partitioning** (FNV-1a) so a key is sticky to one partition
 - **Consumer groups** with **range** and **round-robin partition assignment**, **eager rebalance** on join/leave, and **group-level committed offsets**
 - **Per-key ordering**: records for one key append in order on one partition, and only the assigned member reads them
+- **Exponential backoff** on retry: `min(cap, base * 2^(attempt-1))`
+- **Jitter** (full, equal, decorrelated) so concurrent retries do not align
+- **Delayed retry queue** with an injectable clock, so tests can advance time without sleeping
+
 ## 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).
@@ -65,6 +69,10 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **Consumer groups** with **range** and **round-robin partition assignment**, **eager rebalance** on join/leave, and **group-level committed offsets**
 - **Per-key ordering**: records for one key append in order on one partition, and only the assigned member reads them
 - **Consumer groups with partition assignment.** A `PartitionedTopic` is an append-only log split into N partitions. `produce(key, payload)` hashes the key with FNV-1a so that key always lands on the same partition. A `ConsumerGroup` assigns each partition to at most one member using Kafka's range assignor (consecutive slices, remainder on the first members) or round-robin (interleaved). Independent groups each see the full log. Offsets are stored on the group, so a rebalance hands a partition to a peer at the last committed offset instead of replaying from zero. A thrown handler stalls that partition until the next pump.
+- **Exponential backoff** on retry: `min(cap, base * 2^(attempt-1))`
+- **Jitter** (full, equal, decorrelated) so concurrent retries do not align
+- **Delayed retry queue** with an injectable clock, so tests can advance time without sleeping
+- **Exponential backoff with jitter on retry.** A nack or handler throw no longer redelivers in the same tick by default. The wait is exponential, then jittered (`full` by default, also `equal` and `decorrelated`). Ready work is not blocked behind a delayed retry. Pass `retryBackoff: false` for immediate requeue.
 ## Usage
 
 ```ts

src/durable-work-queue.ts

@@ -59,7 +59,7 @@ export class DurableWorkQueue<T> {
     if (!Number.isInteger(maxDeliveryCount) || maxDeliveryCount < 1) {
       throw new Error(`maxDeliveryCount must be a positive integer, got ${maxDeliveryCount}`)
     }
-    const inner = new WorkQueue<Inner<T>>({ maxDeliveryCount })
+    const inner = new WorkQueue<Inner<T>>({ maxDeliveryCount, retryBackoff: false })
     const live = new Map<number, T>()
     let nextId = 1
     try {

src/index.ts

@@ -91,3 +91,11 @@ export type {
   RecordHandler,
   GroupMember,
 } from './consumer-group.js'
+
+export { RetryBackoff, ManualClock, systemClock } from './retry-backoff.js'
+
+export type {
+  JitterStrategy,
+  RetryClock,
+  BackoffOptions,
+} from './retry-backoff.js'

src/retry-backoff.ts

@@ -0,0 +1,152 @@
+export type JitterStrategy = 'none' | 'full' | 'equal' | 'decorrelated'
+
+export interface RetryClock {
+  now(): number
+  schedule(fn: () => void, delayMs: number): () => void
+}
+
+export interface BackoffOptions {
+  readonly baseDelayMs?: number
+  readonly maxDelayMs?: number
+  readonly jitter?: JitterStrategy
+  readonly random?: () => number
+}
+
+const DEFAULT_BASE_DELAY_MS = 100
+const DEFAULT_MAX_DELAY_MS = 30_000
+const JITTER: ReadonlySet<JitterStrategy> = new Set([
+  'none',
+  'full',
+  'equal',
+  'decorrelated',
+])
+
+export function systemClock(): RetryClock {
+  return {
+    now: () => Date.now(),
+    schedule(fn, delayMs) {
+      const handle = setTimeout(fn, delayMs)
+      return () => clearTimeout(handle)
+    },
+  }
+}
+
+export class ManualClock implements RetryClock {
+  private current = 0
+  private nextId = 1
+  private readonly timers = new Map<number, { at: number; fn: () => void }>()
+
+  now(): number {
+    return this.current
+  }
+
+  schedule(fn: () => void, delayMs: number): () => void {
+    if (!Number.isFinite(delayMs) || delayMs < 0) {
+      throw new Error(`delayMs must be a non-negative finite number, got ${delayMs}`)
+    }
+    const id = this.nextId++
+    this.timers.set(id, { at: this.current + delayMs, fn })
+    return () => {
+      this.timers.delete(id)
+    }
+  }
+
+  pendingCount(): number {
+    return this.timers.size
+  }
+
+  advance(ms: number): void {
+    if (!Number.isFinite(ms) || ms < 0) {
+      throw new Error(`advance must be a non-negative finite number, got ${ms}`)
+    }
+    const target = this.current + ms
+    for (;;) {
+      let next: { id: number; at: number; fn: () => void } | undefined
+      for (const [id, timer] of this.timers) {
+        if (timer.at > target) continue
+        if (!next || timer.at < next.at || (timer.at === next.at && id < next.id)) {
+          next = { id, at: timer.at, fn: timer.fn }
+        }
+      }
+      if (!next) {
+        this.current = target
+        return
+      }
+      this.current = next.at
+      this.timers.delete(next.id)
+      next.fn()
+    }
+  }
+}
+
+export class RetryBackoff {
+  readonly baseDelayMs: number
+  readonly maxDelayMs: number
+  readonly jitter: JitterStrategy
+  private readonly random: () => number
+
+  constructor(options: BackoffOptions = {}) {
+    const base = options.baseDelayMs ?? DEFAULT_BASE_DELAY_MS
+    const cap = options.maxDelayMs ?? DEFAULT_MAX_DELAY_MS
+    const jitter = options.jitter ?? 'full'
+    if (!Number.isInteger(base) || base < 0) {
+      throw new Error(`baseDelayMs must be a non-negative integer, got ${base}`)
+    }
+    if (!Number.isInteger(cap) || cap < 0) {
+      throw new Error(`maxDelayMs must be a non-negative integer, got ${cap}`)
+    }
+    if (!JITTER.has(jitter)) {
+      throw new Error(`jitter must be none, full, equal, or decorrelated, got ${String(jitter)}`)
+    }
+    this.baseDelayMs = base
+    this.maxDelayMs = cap
+    this.jitter = jitter
+    this.random = options.random ?? Math.random
+  }
+
+  cappedExponential(attempt: number): number {
+    assertAttempt(attempt)
+    const shift = attempt - 1
+    // 2^53 is the last integer power Number can represent exactly.
+    if (shift >= 53) return this.maxDelayMs
+    const raw = this.baseDelayMs * 2 ** shift
+    if (!Number.isFinite(raw)) return this.maxDelayMs
+    return Math.min(this.maxDelayMs, raw)
+  }
+
+  delayMs(attempt: number, lastDelayMs = 0): number {
+    assertAttempt(attempt)
+    if (lastDelayMs < 0 || !Number.isFinite(lastDelayMs)) {
+      throw new Error(`lastDelayMs must be a non-negative finite number, got ${lastDelayMs}`)
+    }
+    const exp = this.cappedExponential(attempt)
+    switch (this.jitter) {
+      case 'none':
+        return exp
+      case 'full':
+        return Math.floor(this.unit() * exp)
+      case 'equal':
+        return Math.floor(exp / 2 + this.unit() * (exp / 2))
+      case 'decorrelated': {
+        const seed = lastDelayMs > 0 ? lastDelayMs : this.baseDelayMs
+        const hi = Math.min(this.maxDelayMs, seed * 3)
+        const lo = Math.min(this.baseDelayMs, hi)
+        return Math.floor(lo + this.unit() * (hi - lo))
+      }
+    }
+  }
+
+  private unit(): number {
+    const u = this.random()
+    if (!Number.isFinite(u) || u < 0 || u >= 1) {
+      throw new Error(`random() must return a number in [0, 1), got ${u}`)
+    }
+    return u
+  }
+}
+
+function assertAttempt(attempt: number): void {
+  if (!Number.isInteger(attempt) || attempt < 1) {
+    throw new Error(`attempt must be a positive integer, got ${attempt}`)
+  }
+}

src/work-queue.ts

@@ -7,6 +7,7 @@ import {
   type FlowState,
   type QueueBoundOptions,
 } from './backpressure.js'
+import { RetryBackoff, systemClock, type BackoffOptions, type RetryClock } from './retry-backoff.js'
 
 export { QueueFullError } from './backpressure.js'
 export type { BackpressureEvent, BackpressureListener, FlowState } from './backpressure.js'
@@ -31,12 +32,20 @@ export type EnqueueResult = { readonly accepted: true; readonly id: number } | {
 
 export interface WorkQueueOptions extends QueueBoundOptions {
   maxDeliveryCount?: number
+  retryBackoff?: BackoffOptions | false
+  clock?: RetryClock
 }
 
 interface PendingMessage<T> {
   id: number
   payload: T
   deliveryCount: number
+  lastRetryDelayMs: number
+}
+
+interface DelayedRetry<T> {
+  pending: PendingMessage<T>
+  availableAt: number
 }
 
 interface InFlightEntry<T> {
@@ -65,6 +74,10 @@ export class WorkQueue<T> {
   private readonly listeners = new Set<BackpressureListener>()
   private readonly gate: WatermarkGate | undefined
   private readonly maxDeliveryCount: number
+  private readonly delayed: DelayedRetry<T>[] = []
+  private readonly backoff: RetryBackoff | null
+  private readonly clock: RetryClock
+  private cancelRetryTimer: (() => void) | undefined
   private nextMessageId = 1
   private nextDeliveryTag = 1
   private nextConsumerId = 1
@@ -85,6 +98,8 @@ export class WorkQueue<T> {
       bounds.capacity === Number.POSITIVE_INFINITY
         ? undefined
         : new WatermarkGate(bounds.highWatermark, bounds.lowWatermark)
+    this.backoff = options?.retryBackoff === false ? null : new RetryBackoff(options?.retryBackoff)
+    this.clock = options?.clock ?? systemClock()
   }
 
   enqueue(payload: T): number {
@@ -96,7 +111,7 @@ export class WorkQueue<T> {
   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.ready.push({ id, payload, deliveryCount: 0, lastRetryDelayMs: 0 })
     this.pump()
     this.applyFlow()
     return { accepted: true, id }
@@ -158,6 +173,19 @@ export class WorkQueue<T> {
     return this.ready.length
   }
 
+  delayedCount(): number {
+    return this.delayed.length
+  }
+
+  nextRetryAt(): number | undefined {
+    if (this.delayed.length === 0) return undefined
+    let soonest = this.delayed[0]!.availableAt
+    for (const item of this.delayed) {
+      if (item.availableAt < soonest) soonest = item.availableAt
+    }
+    return soonest
+  }
+
   inFlightCount(): number {
     return this.inFlight.size
   }
@@ -177,6 +205,7 @@ export class WorkQueue<T> {
     if (this.pumping) return
     this.pumping = true
     try {
+      this.releaseDueRetries()
       // 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>()
@@ -216,6 +245,7 @@ export class WorkQueue<T> {
     } finally {
       this.pumping = false
     }
+    this.armRetryTimer()
   }
 
   private deliver(consumer: ConsumerState<T>, pending: PendingMessage<T>): void {
@@ -259,12 +289,50 @@ export class WorkQueue<T> {
 
     // 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.scheduleRetry(entry.pending)
     }
     this.pump()
     this.applyFlow()
   }
 
+  private scheduleRetry(pending: PendingMessage<T>): void {
+    if (!this.backoff) {
+      this.ready.push(pending)
+      return
+    }
+    const delay = this.backoff.delayMs(pending.deliveryCount, pending.lastRetryDelayMs)
+    pending.lastRetryDelayMs = delay
+    if (delay === 0) {
+      this.ready.push(pending)
+      return
+    }
+    this.delayed.push({ pending, availableAt: this.clock.now() + delay })
+  }
+
+  private releaseDueRetries(): void {
+    if (this.delayed.length === 0) return
+    const now = this.clock.now()
+    const still: DelayedRetry<T>[] = []
+    for (const item of this.delayed) {
+      if (item.availableAt <= now) this.ready.push(item.pending)
+      else still.push(item)
+    }
+    this.delayed.length = 0
+    this.delayed.push(...still)
+  }
+
+  private armRetryTimer(): void {
+    this.cancelRetryTimer?.()
+    this.cancelRetryTimer = undefined
+    const soonest = this.nextRetryAt()
+    if (soonest === undefined) return
+    const wait = Math.max(0, soonest - this.clock.now())
+    this.cancelRetryTimer = this.clock.schedule(() => {
+      this.cancelRetryTimer = undefined
+      this.pump()
+    }, wait)
+  }
+
   private applyFlow(): void {
     if (this.pumping || !this.gate) return
     const next = this.gate.observe(this.ready.length)