feat: configure WebSocket worker delivery timeout

README.md

@@ -951,3 +951,24 @@ app.get("/metrics", async (req, res) => {
 
 Pump event metrics use `tenant`, `data_core`, `flow_type`, and `event_type` labels. The SDK command counter uses only
 `command`. Cluster metrics currently have no tenant or event labels and are process-local.
+
+### WebSocket worker delivery timeout
+
+`FlowcoreDataPumpCluster` accepts optional `workerDeliveryTimeoutMs` (default
+`30000`). It bounds the leader's wait for a WebSocket worker acknowledgement,
+including the handler's rate-limit waits and retries. Supply an integer from 1 to
+2147483647 milliseconds. It does not change the NATS request timeout.
+
+This is separate from `achknowledgeTimeoutMs`, which controls when the source
+reservation can reopen. For slow or throttled handlers, budget both explicitly:
+allow enough worker time for expected processing, and keep the source reservation
+longer than that deadline plus scheduling/acknowledgement margin. For example, a
+single-event handler with at most a 60-second limiter wait and 15 seconds of
+bounded delivery attempts could use an 80-second worker deadline and a source
+reservation longer than 90 seconds. Account separately for batch size and queued
+work; these example values are not safe defaults for arbitrary handlers.
+
+A timeout does not cancel an already-running worker. Its late acknowledgement is
+ignored by the expired tracker; a source retry may duplicate an external side
+effect. Increasing the deadline reduces premature timeouts but does not provide
+exactly-once delivery or eliminate the need for idempotent consumers.

src/data-pump/data-pump-cluster.ts

@@ -16,7 +16,7 @@ const DEFAULT_STALE_THRESHOLD_MS = 30_000
 const DEFAULT_WORKER_CONCURRENCY = 1
 const RECONNECT_BASE_MS = 1_000
 const RECONNECT_MAX_MS = 30_000
-const DELIVERY_TIMEOUT_MS = 30_000
+const DEFAULT_WORKER_DELIVERY_TIMEOUT_MS = 30_000
 
 export interface FlowcoreDataPumpClusterOptions extends FlowcoreDataPumpOptions {
   coordinator: FlowcoreDataPumpCoordinator
@@ -26,6 +26,8 @@ export interface FlowcoreDataPumpClusterOptions extends FlowcoreDataPumpOptions
   leaseRenewIntervalMs?: number
   heartbeatIntervalMs?: number
   workerConcurrency?: number
+  /** WebSocket worker ACK deadline, including handler/rate-limit wait. Default: 30 seconds. */
+  workerDeliveryTimeoutMs?: number
   clusterKey?: string
 }
 
@@ -44,6 +46,7 @@ export class FlowcoreDataPumpCluster {
   private readonly leaseRenewIntervalMs: number
   private readonly heartbeatIntervalMs: number
   private readonly workerConcurrency: number
+  private readonly workerDeliveryTimeoutMs: number
   private readonly logger?: FlowcoreLogger
 
   private running = false
@@ -82,6 +85,15 @@ export class FlowcoreDataPumpCluster {
       throw new Error("advertisedAddress is required when not using NATS distribution (notifier.type !== 'nats')")
     }
 
+    this.workerDeliveryTimeoutMs = options.workerDeliveryTimeoutMs ?? DEFAULT_WORKER_DELIVERY_TIMEOUT_MS
+    if (
+      !Number.isInteger(this.workerDeliveryTimeoutMs) ||
+      this.workerDeliveryTimeoutMs < 1 ||
+      this.workerDeliveryTimeoutMs > 2_147_483_647
+    ) {
+      throw new Error("workerDeliveryTimeoutMs must be an integer between 1 and 2147483647")
+    }
+
     this.instanceId = crypto.randomUUID()
     this.coordinator = options.coordinator
     this.leaseTtlMs = options.leaseTtlMs ?? DEFAULT_LEASE_TTL_MS
@@ -500,7 +512,7 @@ export class FlowcoreDataPumpCluster {
 
     // wait for ack
     try {
-      await worker.deliveryTracker.add(deliveryId, eventIds, DELIVERY_TIMEOUT_MS)
+      await worker.deliveryTracker.add(deliveryId, eventIds, this.workerDeliveryTimeoutMs)
       worker.pendingCount--
       clusterMetrics.workerAcksCounter.inc()
     } catch (error) {

test/tests/worker-delivery-timeout.test.ts

@@ -0,0 +1,94 @@
+import { describe, expect, it, spyOn } from "bun:test"
+import type { FlowcoreEvent } from "@flowcore/sdk"
+import { FlowcoreDataPumpCluster } from "../../src/data-pump/data-pump-cluster.ts"
+import type { FlowcoreDataPumpCoordinator } from "../../src/data-pump/types.ts"
+import { DeliveryTracker } from "../../src/data-pump/ws-protocol.ts"
+
+const options = {
+  auth: { getBearerToken: async () => "synthetic" },
+  dataSource: { tenant: "test", dataCore: "test", flowType: "test", eventTypes: ["test"] },
+  coordinator: {} as FlowcoreDataPumpCoordinator,
+  stateManager: { getState: async () => null, setState: async () => {} },
+  advertisedAddress: "ws://localhost:9999",
+}
+function fixture(timeout?: number) {
+  const cluster = new FlowcoreDataPumpCluster({ ...options, workerDeliveryTimeoutMs: timeout })
+  const tracker = new DeliveryTracker()
+  let deliveryId = ""
+  const worker = {
+    pendingCount: 0,
+    deliveryTracker: tracker,
+    connection: {
+      isClosed: false,
+      send: (message: { deliveryId: string }) => {
+        deliveryId = message.deliveryId
+      },
+    },
+  }
+  const internals = cluster as unknown as {
+    workers: Map<string, typeof worker>
+    distributeEventsWs(events: FlowcoreEvent[]): Promise<void>
+  }
+  internals.workers.set("worker", worker)
+  return {
+    tracker,
+    worker,
+    id: () => deliveryId,
+    send: () => internals.distributeEventsWs([{ eventId: "synthetic" } as FlowcoreEvent]),
+  }
+}
+
+describe("WebSocket worker delivery deadline", () => {
+  it("preserves the default 30 second deadline", async () => {
+    const f = fixture()
+    const add = spyOn(f.tracker, "add")
+    const result = f.send()
+    expect(add.mock.calls[0][2]).toBe(30_000)
+    f.tracker.ack(f.id())
+    await result
+  })
+  it("accepts a worker acknowledgement after a simulated throttle wait within the configured deadline", async () => {
+    const f = fixture(100)
+    const result = f.send()
+    await Bun.sleep(40)
+    expect(f.tracker.size).toBe(1)
+    f.tracker.ack(f.id())
+    await result
+    expect(f.tracker.size).toBe(0)
+    expect(f.worker.pendingCount).toBe(0)
+  })
+  it("expires at the configured deadline and ignores late acknowledgement of the previous delivery", async () => {
+    const f = fixture(100)
+    const expired = f.send().then(
+      () => null,
+      (error: Error) => error,
+    )
+    const oldId = f.id()
+    await Bun.sleep(130)
+    expect((await expired)?.message).toContain("timed out")
+    const retry = f.send()
+    f.tracker.ack(oldId)
+    expect(f.tracker.size).toBe(1)
+    f.tracker.ack(f.id())
+    await retry
+    expect(f.worker.pendingCount).toBe(0)
+  })
+  it("still propagates worker failure immediately", async () => {
+    const f = fixture(100)
+    const result = f.send().then(
+      () => null,
+      (error: Error) => error,
+    )
+    f.tracker.fail(f.id())
+    expect((await result)?.message).toContain("failed by worker")
+    expect(f.tracker.size).toBe(0)
+    expect(f.worker.pendingCount).toBe(0)
+  })
+  for (const value of [0, -1, 1.5, NaN, Infinity, 2_147_483_648]) {
+    it(`rejects invalid timer duration ${value}`, () => {
+      expect(() => new FlowcoreDataPumpCluster({ ...options, workerDeliveryTimeoutMs: value })).toThrow(
+        "workerDeliveryTimeoutMs",
+      )
+    })
+  }
+})