From a1c14cae3fc1680199f040172c030db6ec1e8a45 Mon Sep 17 00:00:00 2001 From: jbiskur Date: Fri, 12 Jun 2026 10:33:34 +0100 Subject: [PATCH 1/2] fix: use trusted publishing for npm release --- .github/workflows/build.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 60630da..a1a09a4 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -5,6 +5,7 @@ on: types: [published] permissions: + id-token: write contents: read jobs: @@ -34,7 +35,7 @@ jobs: run: bun run lint - name: Build for NPM run: bun run build - - uses: useblacksmith/setup-node@v5 + - uses: actions/setup-node@v4 with: node-version: "20.x" registry-url: "https://registry.npmjs.org" From 04956e93697c07e1c32835474721d1ae5780960d Mon Sep 17 00:00:00 2001 From: jbiskur Date: Fri, 28 Aug 2026 16:35:51 +0100 Subject: [PATCH 2/2] fix: restart resumes the process loop and the pulse emitter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `restart()` calls `stop(true)`, which clears `running`, empties the buffer and stops the pulse emitter. The fetch loop revives itself from `restartTo`, but the process loop exits on `running` and only `start()` brought it back. A pump restarted while it was delivering kept pulling events and never delivered or checkpointed again — and stopped pulsing, so it looked dead to the control plane while its host reported healthy. This is the defect described on 2026-04-14 in Usable fragment 28875215-003d-4770-9d44-8e3488d25246. It stalled two production data-pathways pathways for 26 days (2026-08-02 → 2026-08-28): 417b8dd9-b6a3-4ce9-b254-6ea12ca97d57 and 5a32bd1c-9d0c-4aef-a813-4fee60280f8e. - `ensureProcessLoop()` starts the delivery loop whenever the pump is running with a processor and no loop is live. The fetch loop calls it every iteration, so a loop that exits for any reason comes back within one iteration instead of leaving a pump that pulls but never delivers. - `processLoopRunning` guards against a second loop when the existing one is only parked in `reserve()`. Two loops would race over the same buffer. - A loop that exits while the pump is running again — the restart landed mid-batch — restarts itself. - The restart path starts the pulse emitter again. `PulseEmitter.start()` is already idempotent. Tests: delivery continues after a restart issued from inside the handler; no second loop when one is parked; the pulse emitter is started again. All three fail on the previous behaviour. Co-Authored-By: Claude Opus 5 (1M context) --- src/data-pump/data-pump.ts | 62 +++++++-- test/tests/data-pump-restart.test.ts | 195 +++++++++++++++++++++++++++ 2 files changed, 247 insertions(+), 10 deletions(-) diff --git a/src/data-pump/data-pump.ts b/src/data-pump/data-pump.ts index 0c6f761..a5f1bf3 100644 --- a/src/data-pump/data-pump.ts +++ b/src/data-pump/data-pump.ts @@ -85,6 +85,7 @@ export class FlowcoreDataPump { private nextCursor?: string private running = false private restartTo?: FlowcoreDataPumpState + private processLoopRunning = false private abortController?: AbortController private buffer: FlowcoreDataPumpBufferItem[] = [] private bufferState: FlowcoreDataPumpState @@ -311,6 +312,7 @@ export class FlowcoreDataPump { private async loop(): Promise { do { + this.ensureProcessLoop() const amountToFetch = this.options.bufferSize - this.buffer.length if (amountToFetch <= 0) { @@ -388,6 +390,14 @@ export class FlowcoreDataPump { this.bufferState = this.restartTo this.restartTo = undefined this.running = true + // `stop(true)` cleared `running` and stopped the pulse emitter, and the + // process loop exits whenever `running` is false. Only `start()` used + // to bring them back, so a pump restarted while it was delivering kept + // pulling events and never delivered or checkpointed again — it looked + // alive from the outside. Both are idempotent, so a loop that merely + // parked in `reserve()` is left alone. + this.ensureProcessLoop() + this.pulseEmitter?.start() return this.loop() } catch (error) { this.logger?.error("Failed to consume restartTo, dropping it", { error }) @@ -548,18 +558,50 @@ export class FlowcoreDataPump { // #region Pusher + /** + * Guarantee a live process loop whenever the pump is running with a + * processor. Called from the fetch loop, so a delivery loop that exited for + * any reason — most importantly a restart, which clears `running` while the + * loop is mid-batch — comes back within one fetch iteration instead of + * leaving a pump that pulls but never delivers. + */ + private ensureProcessLoop(): void { + if (!this.options.processor || !this.running || this.processLoopRunning) { + return + } + this.startProcessLoop() + } + private startProcessLoop(): void { - this.processLoop().catch((error) => { - this.logger?.error("Error in processor", { error }) - if (!this.running) return - this.processLoopRestartAttempts++ - const delay = Math.min(1_000 * Math.pow(2, this.processLoopRestartAttempts - 1), 30_000) - this.logger?.warn(`Restarting process loop in ${delay}ms (attempt ${this.processLoopRestartAttempts})`) - setTimeout(() => { + // Guard against a second loop: `restart()` asks for the loop back, but the + // running one may only have been parked in `reserve()`. Two loops would + // race over the same buffer. + if (this.processLoopRunning) { + return + } + this.processLoopRunning = true + this.processLoop() + .then(() => { + this.processLoopRunning = false + // The loop exits as soon as `running` goes false. If the pump is + // running again by the time we get here, a restart brought it back + // while this loop was finishing its last batch — resume delivery. + if (this.running && this.options.processor) { + this.startProcessLoop() + } + }) + .catch((error) => { + this.processLoopRunning = false + this.logger?.error("Error in processor", { error }) if (!this.running) return - this.startProcessLoop() - }, delay) - }) + this.processLoopRestartAttempts++ + const delay = Math.min(1_000 * Math.pow(2, this.processLoopRestartAttempts - 1), 30_000) + this.logger?.warn(`Restarting process loop in ${delay}ms (attempt ${this.processLoopRestartAttempts})`) + setTimeout(() => { + if (!this.running) return + this.startProcessLoop() + }, delay) + }) } private async processLoop() { diff --git a/test/tests/data-pump-restart.test.ts b/test/tests/data-pump-restart.test.ts index 1c4356e..78f8674 100644 --- a/test/tests/data-pump-restart.test.ts +++ b/test/tests/data-pump-restart.test.ts @@ -448,3 +448,198 @@ describe("backoff formula", () => { assertEquals(delays, [1_000, 2_000, 4_000, 8_000, 16_000, 30_000]) }) }) + +// #region restart resumes delivery + +/** + * `restart()` calls `stop(true)`, which clears `running`. The fetch loop + * revives itself from `restartTo`, but the process loop exits on `running` and + * only `start()` used to bring it back. A pump restarted while it was + * delivering kept pulling and never delivered or checkpointed again, while the + * heartbeat above it stayed healthy. Two production pathways stalled that way + * for 26 days (2026-08-02 → 2026-08-28). + */ +describe("restart resumes the process loop", () => { + beforeEach(() => { + jest.useFakeTimers() + }) + + afterEach(() => { + jest.useRealTimers() + }) + + function event(id: string): FlowcoreEvent { + return { + eventId: id, + eventType: "test.created.0", + aggregator: "test", + validTime: "2026-03-31T12:00:00.000Z", + timeBucket: "20260331120000", + payload: {}, + } as unknown as FlowcoreEvent + } + + it("keeps delivering after a restart issued from inside the handler", async () => { + const { logger } = createMockLogger() + const delivered: string[] = [] + let batch = 0 + const fakeDataSource = new FakeDataSource({ + getEventsImpl: () => { + batch++ + if (batch === 1) return Promise.resolve({ events: [event("evt-1")], nextCursor: undefined }) + // Keep offering evt-2: restart() clears the buffer, so anything pulled + // between the restart request and the resume is re-fetched. + if (batch <= 6) return Promise.resolve({ events: [event("evt-2")], nextCursor: undefined }) + return Promise.resolve({ events: [], nextCursor: undefined }) + }, + }) + + const pump = FlowcoreDataPump.create( + { + auth: { apiKey: FAKE_API_KEY }, + dataSource: { + tenant: "test", + dataCore: "test-dc", + flowType: "test.0", + eventTypes: ["test.created.0"], + }, + stateManager: createMockStateManager(), + processor: { + concurrency: 1, + handler: (events) => { + for (const e of events) { + delivered.push(e.eventId) + // Restart from inside the handler: the process loop is mid-batch, + // so it sees running === false on its next iteration and exits. + if (e.eventId === "evt-1") { + pump.restart({ timeBucket: "20260331130000" }) + } + } + return Promise.resolve() + }, + }, + notifier: { type: "poller", intervalMs: 60_000 }, + logger, + baseUrlOverride: "http://localhost:9999", + noTranslation: true, + }, + fakeDataSource, + ) + + void pump.start(() => {}) + await tickAsync(0) + assertEquals(delivered[0], "evt-1") + + // Let the fetch loop consume restartTo and pull the next batch. + await tickAsync(1_000) + await tickAsync(1_000) + + // Before the fix the pump kept pulling and delivered nothing more. + assert(delivered.includes("evt-2"), `expected evt-2 to be delivered, got ${JSON.stringify(delivered)}`) + assertEquals(pump.isRunning, true) + + pump.stop() + await tickAsync(60_000) + }) + + it("does not start a second process loop when one is only parked", async () => { + const { logger } = createMockLogger() + const delivered: string[] = [] + let batch = 0 + const fakeDataSource = new FakeDataSource({ + getEventsImpl: () => { + batch++ + if (batch === 1) return Promise.resolve({ events: [event("evt-1")], nextCursor: undefined }) + return Promise.resolve({ events: [], nextCursor: undefined }) + }, + }) + + const pump = FlowcoreDataPump.create( + { + auth: { apiKey: FAKE_API_KEY }, + dataSource: { + tenant: "test", + dataCore: "test-dc", + flowType: "test.0", + eventTypes: ["test.created.0"], + }, + stateManager: createMockStateManager(), + processor: { + concurrency: 1, + handler: (events) => { + for (const e of events) delivered.push(e.eventId) + return Promise.resolve() + }, + }, + notifier: { type: "poller", intervalMs: 60_000 }, + logger, + baseUrlOverride: "http://localhost:9999", + noTranslation: true, + }, + fakeDataSource, + ) + + void pump.start(() => {}) + await tickAsync(0) + assertEquals(delivered, ["evt-1"]) + + // The loop is parked in reserve(); a restart must not add a second one. + pump.restart({ timeBucket: "20260331130000" }) + await tickAsync(1_000) + await tickAsync(1_000) + + // evt-1 is delivered once, not twice. + assertEquals(delivered.filter((id) => id === "evt-1").length, 1) + + pump.stop() + await tickAsync(60_000) + }) + + it("restarts the pulse emitter", async () => { + const { logger } = createMockLogger() + const fakeDataSource = new FakeDataSource() + const pump = FlowcoreDataPump.create( + { + auth: { apiKey: FAKE_API_KEY }, + dataSource: { + tenant: "test", + dataCore: "test-dc", + flowType: "test.0", + eventTypes: ["test.created.0"], + }, + stateManager: createMockStateManager(), + notifier: { type: "poller", intervalMs: 60_000 }, + logger, + baseUrlOverride: "http://localhost:9999", + noTranslation: true, + pulse: { url: "http://localhost:9999", pathwayId: "11111111-1111-1111-1111-111111111111" }, + }, + fakeDataSource, + ) + + const emitter = (pump as unknown as { pulseEmitter: { start: () => void; stop: () => void } }).pulseEmitter + let starts = 0 + const originalStart = emitter.start.bind(emitter) + emitter.start = () => { + starts++ + originalStart() + } + + void pump.start(() => {}) + await tickAsync(0) + assertEquals(starts, 1) + + pump.restart({ timeBucket: "20260331130000" }) + await tickAsync(1_000) + await tickAsync(1_000) + + // A pump that stops pulsing looks dead to the control plane even while it + // is working, so the emitter has to come back with the loops. + assertEquals(starts, 2) + + pump.stop() + await tickAsync(60_000) + }) +}) + +// #endregion