From e6af1281f5f714658f8b86173a4c8fb1e796c4b5 Mon Sep 17 00:00:00 2001 From: jbiskur Date: Tue, 9 Jun 2026 15:18:37 +0100 Subject: [PATCH] fix(notifier): arm timeout/abort before connect() to prevent hung-reconnect wedge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit waitWebSocket awaited notificationClient.connect() before arming the re-poll timeout and abort listener. A reconnect whose connect() never resolves (half-open / black-holed socket) therefore blocked the fetch loop forever: the timer was never armed, abort was never wired, and the finally never ran. The pulse emitter (separate timer) kept reporting "alive", so the pump appeared healthy while silently processing nothing — recoverable only by a full process restart. The existing main-loop self-heal (#74) only catches thrown errors, not a hang. Arm the timeout + abort listener BEFORE connect(), and race connect() against the resolution promise. A hung connect now resolves the wait at timeoutMs; the loop re-polls (catching up backlog) and attempts a fresh connect each cycle, auto-recovering to push/WS mode the moment a connect succeeds. A connect rejection still surfaces, preserving main-loop self-heal. No sticky poll-mode state. Adds notifier test cases 9 (hung connect must not wedge) and 10 (retries while degraded, returns to WS mode on reconnect). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/data-pump/notifier.ts | 22 +++++++-- test/tests/notifier.test.ts | 98 +++++++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 4 deletions(-) diff --git a/src/data-pump/notifier.ts b/src/data-pump/notifier.ts index 9749d1a..c1274eb 100644 --- a/src/data-pump/notifier.ts +++ b/src/data-pump/notifier.ts @@ -120,6 +120,16 @@ export class FlowcoreNotifier { }, }) + // Arm the timeout + abort listener BEFORE connect(). A reconnect whose + // connect() never resolves (half-open / black-holed socket) must not wedge + // the pump: the timer then resolves the wait at timeoutMs and the fetch loop + // re-polls (catching up any backlog) and attempts a fresh connect next cycle. + // The moment a connect() succeeds, that cycle waits on real WS events again — + // there is no sticky poll-mode state, so recovery to push mode is automatic. + clearTimeout(this.timer) + this.timer = setTimeout(() => this.eventResolver?.(), this.options.timeoutMs ?? DEFAULT_TIMEOUT_MS) + signal?.addEventListener("abort", () => this.eventResolver?.()) + this.notificationClient = _internals.createNotificationClient( this.subject, this.webSocketAuth(), @@ -134,13 +144,17 @@ export class FlowcoreNotifier { }, ) + const connectPromise = this.notificationClient.connect() try { - await this.notificationClient.connect() - clearTimeout(this.timer) - this.timer = setTimeout(() => this.eventResolver?.(), this.options.timeoutMs ?? DEFAULT_TIMEOUT_MS) - signal?.addEventListener("abort", () => this.eventResolver?.()) + // Race connect() against the resolution promise so a hung connect can't + // block. A connect REJECTION still surfaces (preserving the main-loop + // self-heal that restarts on a thrown error). + await Promise.race([connectPromise, promise]) await promise } finally { + // Swallow a late connect() rejection that may arrive after we already + // moved on via the timeout, so it never surfaces as an unhandledRejection. + connectPromise.catch(() => {}) clearTimeout(this.timer) this.eventResolver = undefined this.notificationClient.disconnect() diff --git a/test/tests/notifier.test.ts b/test/tests/notifier.test.ts index 744b1fe..8137bf0 100644 --- a/test/tests/notifier.test.ts +++ b/test/tests/notifier.test.ts @@ -24,6 +24,10 @@ interface FakeClientHandle { connectError?: Error // If true, subject.error fires BEFORE connect() resolves (synchronous race). errorBeforeConnect?: Error + // If true, connect() returns a promise that never resolves (simulates a + // hung websocket reconnect — half-open / black-holed socket). This is the + // production wedge: pre-fix, `await connect()` blocks the loop forever. + hangConnect?: boolean } type FakeFactoryConfig = { @@ -53,6 +57,10 @@ function createFakeFactory(config: FakeFactoryConfig) { if (handle.errorBeforeConnect) { handle.subject.error(handle.errorBeforeConnect) } + if (handle.hangConnect) { + // Never resolves — simulates a hung reconnect. + await new Promise(() => {}) + } if (handle.connectError) { throw handle.connectError } @@ -451,6 +459,96 @@ describe("FlowcoreNotifier.waitWebSocket — bug fix (v0.20.1)", () => { assertEquals(resolvedCleanly, false) }) + it("case 9 — hung connect() must not wedge: wait() resolves via the pre-armed timeout", async () => { + // Production wedge: a reconnect whose connect() never resolves. Pre-fix, the + // 20s re-poll timer is armed only AFTER connect() resolves, so the loop hangs + // forever. Post-fix, timer + abort are armed BEFORE connect(), so wait() + // resolves at timeoutMs and the loop re-polls. + const handles: FakeClientHandle[] = [] + const { factory } = createFakeFactory({ + onCreate: (h) => { + h.hangConnect = true + handles.push(h) + }, + }) + factoryStub = stub(_internals, "createNotificationClient", factory) + + const notifier = createNotifier(30) + // Safety: if the bug is present, wait() hangs. Force-resolve at 250ms so the + // runner doesn't stall — the elapsed assertion then fails (proving the wedge). + const safety = setTimeout(() => { + const r = (notifier as unknown as { eventResolver?: () => void }).eventResolver + r?.() + }, 250) + + const start = Date.now() + await notifier.wait() + const elapsed = Date.now() - start + clearTimeout(safety) + + assert(elapsed >= 25, `expected resolve via ~30ms timeout, got ${elapsed} ms`) + assert(elapsed < 200, `hung connect must not wedge — expected < 200 ms, got ${elapsed} ms`) + assertEquals(handles.length, 1) + assertEquals(handles[0].disconnectCalls, 1, "disconnect must run via try/finally even on hung connect") + }) + + it("case 10 — retries while degraded, then returns to WS mode on reconnect", async () => { + // Cycle 1: connect() hangs (degraded → resolves via timeout re-poll). + // Cycle 2: connect() succeeds and delivers an event (back in push/WS mode). + // Proves there is no sticky poll-mode state and each cycle builds a fresh client. + const handles: FakeClientHandle[] = [] + let cycle = 0 + const { factory } = createFakeFactory({ + onCreate: (h) => { + cycle++ + if (cycle === 1) { + h.hangConnect = true + } else { + h.pendingPostConnect.push((handle) => { + handle.subject.next({ + pattern: "stored.event.notify.0", + data: { + tenant: "test-tenant", + eventId: "event-1", + dataCoreId: "test-dc", + flowType: "test.0", + eventType: "test.created.0", + validTime: new Date().toISOString(), + }, + }) + }) + } + handles.push(h) + }, + }) + factoryStub = stub(_internals, "createNotificationClient", factory) + + const notifier = createNotifier(40) + const safety = setTimeout(() => { + const r = (notifier as unknown as { eventResolver?: () => void }).eventResolver + r?.() + }, 400) + + // Cycle 1 — degraded: hung connect resolves via the ~40ms timeout. + const start1 = Date.now() + await notifier.wait() + const elapsed1 = Date.now() - start1 + + // Cycle 2 — recovered: connect succeeds, event delivered, resolves fast. + const start2 = Date.now() + await notifier.wait() + const elapsed2 = Date.now() - start2 + clearTimeout(safety) + + assert(elapsed1 >= 25 && elapsed1 < 200, `cycle 1 should re-poll at ~40ms, got ${elapsed1} ms`) + assert(elapsed2 < 100, `cycle 2 should resolve fast via WS event (back in WS mode), got ${elapsed2} ms`) + assertEquals(cycle, 2, "a fresh client is built each cycle") + assertEquals(handles.length, 2) + assertNotStrictEquals(handles[0].client, handles[1].client) + assertEquals(handles[0].disconnectCalls, 1, "degraded cycle still disconnects the dead client") + assertEquals(handles[1].disconnectCalls, 1) + }) + it("demotes normal SDK notification lifecycle info logs to debug", async () => { const logger = createRecordingLogger() const { factory } = createFakeFactory({