From 347c37e5a1dd6a435e8527c0f0cedaebe0dd67db Mon Sep 17 00:00:00 2001 From: jbiskur Date: Fri, 28 Aug 2026 16:35:51 +0100 Subject: [PATCH 1/3] 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 98f6748..cba5575 100644 --- a/src/data-pump/data-pump.ts +++ b/src/data-pump/data-pump.ts @@ -95,6 +95,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 @@ -334,6 +335,7 @@ export class FlowcoreDataPump { private async loop(): Promise { do { + this.ensureProcessLoop() const amountToFetch = this.options.bufferSize - this.buffer.length if (amountToFetch <= 0) { @@ -416,6 +418,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 }) @@ -602,18 +612,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 From 76bf591651ca104c4ae96e93f1e31316c4653076 Mon Sep 17 00:00:00 2001 From: jbiskur Date: Tue, 1 Sep 2026 12:49:12 +0100 Subject: [PATCH 2/3] fix: guard restart delivery ownership --- src/data-pump/data-pump.ts | 44 ++++++++---- test/tests/data-pump-restart.test.ts | 104 +++++++++++++++++++++------ 2 files changed, 113 insertions(+), 35 deletions(-) diff --git a/src/data-pump/data-pump.ts b/src/data-pump/data-pump.ts index cba5575..d89f032 100644 --- a/src/data-pump/data-pump.ts +++ b/src/data-pump/data-pump.ts @@ -96,6 +96,9 @@ export class FlowcoreDataPump { private running = false private restartTo?: FlowcoreDataPumpState private processLoopRunning = false + // Invalidates delivery work that crossed a stop/restart boundary. Event IDs + // can reappear during replay, so `running` alone cannot identify the owner. + private processLoopGeneration = 0 private abortController?: AbortController private buffer: FlowcoreDataPumpBufferItem[] = [] private bufferState: FlowcoreDataPumpState @@ -301,11 +304,13 @@ export class FlowcoreDataPump { if (stopAt !== undefined) { this.options.stopAt = stopAt ?? undefined } + this.isLive = false this.stop(true) } - public stop(isRestart = false): void { + public stop(_isRestart = false): void { this.running = false + this.processLoopGeneration++ this.processLoopRestartAttempts = 0 this.mainLoopRestartAttempts = 0 this.buffer = [] @@ -314,9 +319,7 @@ export class FlowcoreDataPump { this.pulseEmitter?.stop() this.abortController?.abort() this.waiterBufferThreshold?.() - if (!isRestart) { - this.waiterEvents?.() - } + this.waiterEvents?.() } private updateState(eventId?: string): Promise | void { @@ -422,8 +425,8 @@ export class FlowcoreDataPump { // 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. + // alive from the outside. The old loop exits through its generation + // guard before a replacement takes ownership. this.ensureProcessLoop() this.pulseEmitter?.start() return this.loop() @@ -439,8 +442,12 @@ export class FlowcoreDataPump { // #region Puller - public async reserve(amount: number): Promise { - if (!this.running) { + public reserve(amount: number): Promise { + return this.reserveInternal(amount) + } + + private async reserveInternal(amount: number, generation?: number): Promise { + if (!this.running || (generation !== undefined && generation !== this.processLoopGeneration)) { return [] } const events: FlowcoreEvent[] = [] @@ -463,7 +470,7 @@ export class FlowcoreDataPump { if (!events.length) { await this.waitForEvents() - return this.reserve(amount) + return this.reserveInternal(amount, generation) } this.updateMetricsGauges() @@ -633,8 +640,9 @@ export class FlowcoreDataPump { if (this.processLoopRunning) { return } + const generation = this.processLoopGeneration this.processLoopRunning = true - this.processLoop() + this.processLoop(generation) .then(() => { this.processLoopRunning = false // The loop exits as soon as `running` goes false. If the pump is @@ -647,24 +655,30 @@ export class FlowcoreDataPump { .catch((error) => { this.processLoopRunning = false this.logger?.error("Error in processor", { error }) - if (!this.running) return + if (!this.isCurrentProcessLoop(generation)) 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(() => { - if (!this.running) return + if (!this.isCurrentProcessLoop(generation)) return this.startProcessLoop() }, delay) }) } - private async processLoop() { - while (this.running) { + private isCurrentProcessLoop(generation: number): boolean { + return this.running && generation === this.processLoopGeneration + } + + private async processLoop(generation: number) { + while (this.isCurrentProcessLoop(generation)) { try { - const events = await this.reserve(this.options.processor?.concurrency ?? 1) + const events = await this.reserveInternal(this.options.processor?.concurrency ?? 1, generation) + if (!this.isCurrentProcessLoop(generation)) return await this.replayObserver.observeHandler(events, async () => { await this.options.processor?.handler(events) }) + if (!this.isCurrentProcessLoop(generation)) return await this.acknowledge(events.map((event) => event.eventId)) this.processLoopRestartAttempts = 0 } catch (error) { diff --git a/test/tests/data-pump-restart.test.ts b/test/tests/data-pump-restart.test.ts index 78f8674..0243b6c 100644 --- a/test/tests/data-pump-restart.test.ts +++ b/test/tests/data-pump-restart.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, jest } from "bun:test" import type { EventListOutput, FlowcoreEvent } from "@flowcore/sdk" +import { TimeUuid } from "@flowcore/time-uuid" import { FlowcoreDataPump } from "../../src/data-pump/data-pump.ts" import { FlowcoreDataSource } from "../../src/data-pump/data-source.ts" import type { FlowcoreDataPumpState, FlowcoreDataPumpStateManager } from "../../src/data-pump/types.ts" @@ -32,6 +33,22 @@ async function flushMicrotasks() { } } +async function waitUntil(condition: () => boolean, message: string) { + for (let attempt = 0; attempt < 50; attempt++) { + if (condition()) return + await tickAsync(0) + } + throw new Error(message) +} + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((resolver) => { + resolve = resolver + }) + return { promise, resolve } +} + function createMockStateManager(): FlowcoreDataPumpStateManager { return { getState: () => Promise.resolve({ timeBucket: "20260331120000" }), @@ -460,6 +477,9 @@ describe("backoff formula", () => { * for 26 days (2026-08-02 → 2026-08-28). */ describe("restart resumes the process loop", () => { + const firstEventId = TimeUuid.fromDate(new Date("2026-03-31T12:00:00.000Z")).toString() + const secondEventId = TimeUuid.fromDate(new Date("2026-03-31T12:00:00.001Z")).toString() + beforeEach(() => { jest.useFakeTimers() }) @@ -486,10 +506,10 @@ describe("restart resumes the process loop", () => { 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 + if (batch === 1) return Promise.resolve({ events: [event(firstEventId)], nextCursor: undefined }) + // Keep offering the second event: 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 }) + if (batch <= 6) return Promise.resolve({ events: [event(secondEventId)], nextCursor: undefined }) return Promise.resolve({ events: [], nextCursor: undefined }) }, }) @@ -511,7 +531,7 @@ describe("restart resumes the process loop", () => { 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") { + if (e.eventId === firstEventId) { pump.restart({ timeBucket: "20260331130000" }) } } @@ -528,28 +548,34 @@ describe("restart resumes the process loop", () => { void pump.start(() => {}) await tickAsync(0) - assertEquals(delivered[0], "evt-1") + assertEquals(delivered[0], firstEventId) // 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)}`) + assert( + delivered.includes(secondEventId), + `expected the second event 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 () => { + it("does not let a pre-restart handler checkpoint the replayed event", async () => { const { logger } = createMockLogger() - const delivered: string[] = [] + const firstHandler = deferred() + const replayHandler = deferred() + const checkpoints: FlowcoreDataPumpState[] = [] + let handlerCalls = 0 let batch = 0 const fakeDataSource = new FakeDataSource({ getEventsImpl: () => { batch++ - if (batch === 1) return Promise.resolve({ events: [event("evt-1")], nextCursor: undefined }) + if (batch <= 2) return Promise.resolve({ events: [event(firstEventId)], nextCursor: undefined }) return Promise.resolve({ events: [], nextCursor: undefined }) }, }) @@ -563,12 +589,18 @@ describe("restart resumes the process loop", () => { flowType: "test.0", eventTypes: ["test.created.0"], }, - stateManager: createMockStateManager(), + stateManager: { + getState: () => ({ timeBucket: "20260331120000" }), + setState: (state) => { + checkpoints.push(state) + }, + }, + bufferSize: 1, processor: { concurrency: 1, - handler: (events) => { - for (const e of events) delivered.push(e.eventId) - return Promise.resolve() + handler: () => { + handlerCalls++ + return handlerCalls === 1 ? firstHandler.promise : replayHandler.promise }, }, notifier: { type: "poller", intervalMs: 60_000 }, @@ -580,21 +612,53 @@ describe("restart resumes the process loop", () => { ) void pump.start(() => {}) - await tickAsync(0) - assertEquals(delivered, ["evt-1"]) + await waitUntil(() => handlerCalls === 1, "initial handler did not start") - // The loop is parked in reserve(); a restart must not add a second one. + // Restart while the old handler owns the event. The source replays the + // same event ID into the new buffer before that handler completes. pump.restart({ timeBucket: "20260331130000" }) - await tickAsync(1_000) - await tickAsync(1_000) + await waitUntil(() => batch === 2 && pump.isRunning, "restart did not refill the buffer") + + firstHandler.resolve() + await waitUntil(() => handlerCalls === 2, "replayed event was not delivered by the new process loop") - // evt-1 is delivered once, not twice. - assertEquals(delivered.filter((id) => id === "evt-1").length, 1) + // The old handler must not acknowledge the replayed copy. Only the handler + // started by the new generation may advance durable state. + assertEquals(checkpoints, []) + replayHandler.resolve() + await waitUntil(() => checkpoints.length === 1, "replayed event was not checkpointed") + assertEquals(checkpoints, [{ timeBucket: "20260331120000", eventId: firstEventId }]) pump.stop() await tickAsync(60_000) }) + it("reports a historical restart as replaying until it catches up", async () => { + const { logger } = createMockLogger() + const fakeDataSource = new FakeDataSource() + const pump = createPumpWithFakeDataSource(fakeDataSource, logger) + + void pump.start(() => {}) + await waitUntil(() => pump.getSnapshot()?.isLive === true, "pump did not reach live state") + + const replayFetch = deferred() + const callsBeforeRestart = fakeDataSource.getEventsCalls + fakeDataSource.setGetEventsImpl(() => replayFetch.promise) + + pump.restart({ timeBucket: "20260331120000" }) + await waitUntil( + () => pump.isRunning && fakeDataSource.getEventsCalls > callsBeforeRestart, + "historical replay did not start", + ) + + assertEquals(pump.getSnapshot()?.isLive, false) + + replayFetch.resolve({ events: [], nextCursor: undefined }) + await tickAsync(0) + pump.stop() + await tickAsync(60_000) + }) + it("restarts the pulse emitter", async () => { const { logger } = createMockLogger() const fakeDataSource = new FakeDataSource() From c5f5fff2ea1bbafbcfebc8417d1efae2aa7a1f3c Mon Sep 17 00:00:00 2001 From: jbiskur Date: Tue, 1 Sep 2026 13:43:38 +0100 Subject: [PATCH 3/3] fix: preserve pump consumers across lifecycle changes --- src/data-pump/data-pump.ts | 79 ++++++++--- test/tests/data-pump-restart.test.ts | 202 ++++++++++++++++++++++++++- 2 files changed, 260 insertions(+), 21 deletions(-) diff --git a/src/data-pump/data-pump.ts b/src/data-pump/data-pump.ts index d89f032..ed31ccd 100644 --- a/src/data-pump/data-pump.ts +++ b/src/data-pump/data-pump.ts @@ -95,10 +95,12 @@ export class FlowcoreDataPump { private nextCursor?: string private running = false private restartTo?: FlowcoreDataPumpState - private processLoopRunning = false // Invalidates delivery work that crossed a stop/restart boundary. Event IDs // can reappear during replay, so `running` alone cannot identify the owner. private processLoopGeneration = 0 + private activeProcessLoopGeneration?: number + private processLoopBackoffGeneration?: number + private processLoopRestartTimer?: ReturnType private abortController?: AbortController private buffer: FlowcoreDataPumpBufferItem[] = [] private bufferState: FlowcoreDataPumpState @@ -308,9 +310,15 @@ export class FlowcoreDataPump { this.stop(true) } - public stop(_isRestart = false): void { + public stop(isRestart = false): void { this.running = false this.processLoopGeneration++ + this.activeProcessLoopGeneration = undefined + this.processLoopBackoffGeneration = undefined + if (this.processLoopRestartTimer) { + clearTimeout(this.processLoopRestartTimer) + this.processLoopRestartTimer = undefined + } this.processLoopRestartAttempts = 0 this.mainLoopRestartAttempts = 0 this.buffer = [] @@ -319,7 +327,7 @@ export class FlowcoreDataPump { this.pulseEmitter?.stop() this.abortController?.abort() this.waiterBufferThreshold?.() - this.waiterEvents?.() + this.notifyEventWaiters(!isRestart) } private updateState(eventId?: string): Promise | void { @@ -371,7 +379,7 @@ export class FlowcoreDataPump { this.nextCursor = nextCursor this.updateMetricsGauges() - events.length && this.waiterEvents?.() + events.length && this.notifyEventWaiters() this.bufferState.eventId = events[events.length - 1]?.eventId ?? this.bufferState.eventId @@ -433,6 +441,7 @@ export class FlowcoreDataPump { } catch (error) { this.logger?.error("Failed to consume restartTo, dropping it", { error }) this.restartTo = undefined + this.notifyEventWaiters() return } } @@ -469,7 +478,7 @@ export class FlowcoreDataPump { } if (!events.length) { - await this.waitForEvents() + await this.waitForEvents(generation) return this.reserveInternal(amount, generation) } @@ -585,7 +594,7 @@ export class FlowcoreDataPump { if (reopenedEvents.length) { this.logger?.info(`Reopened ${reopenedEvents.length} events`) - await this.waiterEvents?.() + this.notifyEventWaiters() } if (!failedEvents.length) { @@ -627,39 +636,52 @@ export class FlowcoreDataPump { * leaving a pump that pulls but never delivers. */ private ensureProcessLoop(): void { - if (!this.options.processor || !this.running || this.processLoopRunning) { + if ( + !this.options.processor || + !this.running || + this.activeProcessLoopGeneration === this.processLoopGeneration || + this.processLoopBackoffGeneration === this.processLoopGeneration + ) { return } this.startProcessLoop() } private startProcessLoop(): void { - // 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) { + const generation = this.processLoopGeneration + if ( + !this.options.processor || + !this.isCurrentProcessLoop(generation) || + this.activeProcessLoopGeneration === generation || + this.processLoopBackoffGeneration === generation + ) { return } - const generation = this.processLoopGeneration - this.processLoopRunning = true + this.activeProcessLoopGeneration = generation this.processLoop(generation) .then(() => { - this.processLoopRunning = false + if (this.activeProcessLoopGeneration !== generation) return + this.activeProcessLoopGeneration = undefined // 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) { + if (this.isCurrentProcessLoop(generation)) { this.startProcessLoop() } }) .catch((error) => { - this.processLoopRunning = false + if (this.activeProcessLoopGeneration !== generation) return + this.activeProcessLoopGeneration = undefined this.logger?.error("Error in processor", { error }) if (!this.isCurrentProcessLoop(generation)) 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(() => { + this.processLoopBackoffGeneration = generation + this.processLoopRestartTimer = setTimeout(() => { + if (this.processLoopBackoffGeneration !== generation) return + this.processLoopBackoffGeneration = undefined + this.processLoopRestartTimer = undefined if (!this.isCurrentProcessLoop(generation)) return this.startProcessLoop() }, delay) @@ -807,10 +829,27 @@ export class FlowcoreDataPump { // #region Waiters - private waiterEvents?: () => void - private async waitForEvents() { + private publicEventWaiter?: () => void + private readonly processEventWaiters = new Map void>() + + private notifyEventWaiters(includePublic = true): void { + if (includePublic) { + const publicWaiter = this.publicEventWaiter + this.publicEventWaiter = undefined + publicWaiter?.() + } + const processWaiters = [...this.processEventWaiters.values()] + this.processEventWaiters.clear() + for (const waiter of processWaiters) waiter() + } + + private async waitForEvents(generation?: number) { const promise = new Promise((resolve) => { - this.waiterEvents = resolve + if (generation === undefined) { + this.publicEventWaiter = resolve + } else { + this.processEventWaiters.set(generation, resolve) + } }) await this.replayObserver.observeIdle("waiting_for_events", () => promise) } diff --git a/test/tests/data-pump-restart.test.ts b/test/tests/data-pump-restart.test.ts index 0243b6c..9b02561 100644 --- a/test/tests/data-pump-restart.test.ts +++ b/test/tests/data-pump-restart.test.ts @@ -128,12 +128,14 @@ describe("processLoop restart", () => { interface FakeDataSourceOptions { timeBuckets?: string[] getEventsImpl?: () => Promise + getTimeBucketsImpl?: () => Promise } class FakeDataSource extends FlowcoreDataSource { public getEventsCalls = 0 private readonly timeBucketsValue: string[] private getEventsImpl: () => Promise + private readonly getTimeBucketsImpl?: () => Promise constructor(opts: FakeDataSourceOptions = {}) { super({ @@ -149,10 +151,11 @@ class FakeDataSource extends FlowcoreDataSource { }) this.timeBucketsValue = opts.timeBuckets ?? ["20260331120000", "20260331130000"] this.getEventsImpl = opts.getEventsImpl ?? (() => Promise.resolve({ events: [], nextCursor: undefined })) + this.getTimeBucketsImpl = opts.getTimeBucketsImpl } public override getTimeBuckets(_force = false): Promise { - return Promise.resolve(this.timeBucketsValue) + return this.getTimeBucketsImpl?.() ?? Promise.resolve(this.timeBucketsValue) } public override getClosestTimeBucket(timeBucket: string, getBefore = false): Promise { @@ -217,6 +220,88 @@ function createPumpWithFakeDataSource( // #endregion +describe("process-loop ownership", () => { + beforeEach(() => { + jest.useFakeTimers() + }) + + afterEach(() => { + jest.useRealTimers() + }) + + function createOwnedLoopPump(fakeDataSource: FakeDataSource): FlowcoreDataPump { + return FlowcoreDataPump.create( + { + auth: { apiKey: FAKE_API_KEY }, + dataSource: { + tenant: "test", + dataCore: "test-dc", + flowType: "test.0", + eventTypes: ["test.created.0"], + }, + stateManager: createMockStateManager(), + processor: { handler: () => Promise.resolve() }, + notifier: { type: "poller", intervalMs: 60_000 }, + baseUrlOverride: "http://localhost:9999", + noTranslation: true, + }, + fakeDataSource, + ) + } + + it("starts at most one process loop for a generation", async () => { + const loop = deferred() + const pump = createOwnedLoopPump(new FakeDataSource()) + let processLoopCalls = 0 + const internals = pump as unknown as { + processLoop: (generation: number) => Promise + ensureProcessLoop: () => void + } + internals.processLoop = () => { + processLoopCalls++ + return loop.promise + } + + void pump.start(() => {}) + await waitUntil(() => processLoopCalls === 1, "process loop did not start") + internals.ensureProcessLoop() + internals.ensureProcessLoop() + internals.ensureProcessLoop() + assertEquals(processLoopCalls, 1) + + pump.stop() + loop.resolve() + await tickAsync(60_000) + assertEquals(processLoopCalls, 1) + }) + + it("does not bypass process-loop retry backoff", async () => { + const pump = createOwnedLoopPump(new FakeDataSource()) + let processLoopCalls = 0 + const internals = pump as unknown as { + processLoop: (generation: number) => Promise + ensureProcessLoop: () => void + } + internals.processLoop = () => { + processLoopCalls++ + return Promise.reject(new Error("outer process-loop failure")) + } + + void pump.start(() => {}) + await waitUntil(() => processLoopCalls === 1, "process loop did not start") + await tickAsync(0) + internals.ensureProcessLoop() + internals.ensureProcessLoop() + assertEquals(processLoopCalls, 1) + + await tickAsync(1_000) + assertEquals(processLoopCalls, 2) + + pump.stop() + await tickAsync(60_000) + }) +}) + describe("startMainLoop self-heal", () => { beforeEach(() => { jest.useFakeTimers() @@ -633,6 +718,121 @@ describe("restart resumes the process loop", () => { await tickAsync(60_000) }) + it("keeps the documented pull consumer parked across restart", async () => { + const { logger } = createMockLogger() + const restartCatalog = deferred() + let serveReplay = false + const fakeDataSource = new FakeDataSource({ + getTimeBucketsImpl: () => restartCatalog.promise, + getEventsImpl: () => + Promise.resolve({ + events: serveReplay ? [event(secondEventId)] : [], + nextCursor: undefined, + }), + }) + const pump = createPumpWithFakeDataSource(fakeDataSource, logger) + const delivered: string[] = [] + let consumerStarted = false + let consumerExited = false + + void pump.start(() => {}) + await waitUntil(() => pump.isRunning, "pump did not start") + void (async () => { + consumerStarted = true + while (pump.isRunning) { + const events = await pump.reserve(1) + delivered.push(...events.map((item) => item.eventId)) + await pump.acknowledge(events.map((item) => item.eventId)) + } + consumerExited = true + })() + await waitUntil(() => consumerStarted, "pull consumer did not start") + await tickAsync(0) + + pump.restart({ timeBucket: "20260331130000" }) + await tickAsync(0) + + assertEquals(consumerExited, false, "restart must not end a public reserve loop") + assertEquals(delivered, []) + + serveReplay = true + restartCatalog.resolve(["20260331120000", "20260331130000"]) + await waitUntil(() => delivered.includes(secondEventId), "pull consumer did not receive replayed event") + assertEquals(consumerExited, false) + + pump.stop() + await waitUntil(() => consumerExited, "pull consumer did not exit after stop") + await tickAsync(60_000) + }) + + it("starts a new process generation while an old handler is stuck", async () => { + const { logger } = createMockLogger() + const firstHandler = deferred() + const secondHandler = deferred() + const checkpoints: FlowcoreDataPumpState[] = [] + let handlerCalls = 0 + let fetchCalls = 0 + const fakeDataSource = new FakeDataSource({ + getEventsImpl: () => { + fetchCalls++ + return Promise.resolve({ + events: fetchCalls <= 2 ? [event(fetchCalls === 1 ? firstEventId : secondEventId)] : [], + 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: { + getState: () => ({ timeBucket: "20260331120000" }), + setState: (state) => { + checkpoints.push(state) + }, + }, + bufferSize: 1, + processor: { + concurrency: 1, + handler: () => { + handlerCalls++ + return handlerCalls === 1 ? firstHandler.promise : secondHandler.promise + }, + }, + notifier: { type: "poller", intervalMs: 60_000 }, + logger, + baseUrlOverride: "http://localhost:9999", + noTranslation: true, + }, + fakeDataSource, + ) + + void pump.start(() => {}) + await waitUntil(() => handlerCalls === 1, "initial handler did not start") + + pump.stop() + void pump.start(() => {}) + await waitUntil(() => fetchCalls >= 2 && pump.isRunning, "pump did not refill after start") + await waitUntil(() => handlerCalls === 2, "new process generation stayed blocked by old handler") + assertEquals(checkpoints, []) + + firstHandler.resolve() + await tickAsync(0) + assertEquals(checkpoints, [], "stale handler must not checkpoint the new buffer") + + secondHandler.resolve() + await waitUntil(() => checkpoints.length === 1, "current handler did not checkpoint") + assertEquals(checkpoints, [{ timeBucket: "20260331120000", eventId: secondEventId }]) + + pump.stop() + await tickAsync(60_000) + }) + it("reports a historical restart as replaying until it catches up", async () => { const { logger } = createMockLogger() const fakeDataSource = new FakeDataSource()