From f8d22965ad416b0aa57b44bdc81c5111f59689f9 Mon Sep 17 00:00:00 2001 From: Brian Bischoff Date: Fri, 30 Oct 2026 22:11:10 +0000 Subject: [PATCH] feat: configure WebSocket worker delivery timeout --- README.md | 21 +++++ src/data-pump/data-pump-cluster.ts | 16 +++- test/tests/worker-delivery-timeout.test.ts | 94 ++++++++++++++++++++++ 3 files changed, 129 insertions(+), 2 deletions(-) create mode 100644 test/tests/worker-delivery-timeout.test.ts diff --git a/README.md b/README.md index 5ee91cf..ab26828 100644 --- a/README.md +++ b/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. diff --git a/src/data-pump/data-pump-cluster.ts b/src/data-pump/data-pump-cluster.ts index 492a130..0ad9fe6 100644 --- a/src/data-pump/data-pump-cluster.ts +++ b/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) { diff --git a/test/tests/worker-delivery-timeout.test.ts b/test/tests/worker-delivery-timeout.test.ts new file mode 100644 index 0000000..117d6a6 --- /dev/null +++ b/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 + distributeEventsWs(events: FlowcoreEvent[]): Promise + } + 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", + ) + }) + } +})