diff --git a/src/data-pump/data-pump.ts b/src/data-pump/data-pump.ts index 98f6748..ed31ccd 100644 --- a/src/data-pump/data-pump.ts +++ b/src/data-pump/data-pump.ts @@ -95,6 +95,12 @@ export class FlowcoreDataPump { private nextCursor?: string private running = false private restartTo?: FlowcoreDataPumpState + // 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 @@ -300,11 +306,19 @@ export class FlowcoreDataPump { if (stopAt !== undefined) { this.options.stopAt = stopAt ?? undefined } + this.isLive = false this.stop(true) } 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 = [] @@ -313,9 +327,7 @@ export class FlowcoreDataPump { this.pulseEmitter?.stop() this.abortController?.abort() this.waiterBufferThreshold?.() - if (!isRestart) { - this.waiterEvents?.() - } + this.notifyEventWaiters(!isRestart) } private updateState(eventId?: string): Promise | void { @@ -334,6 +346,7 @@ export class FlowcoreDataPump { private async loop(): Promise { do { + this.ensureProcessLoop() const amountToFetch = this.options.bufferSize - this.buffer.length if (amountToFetch <= 0) { @@ -366,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 @@ -416,10 +429,19 @@ 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. The old loop exits through its generation + // guard before a replacement takes ownership. + this.ensureProcessLoop() + this.pulseEmitter?.start() return this.loop() } catch (error) { this.logger?.error("Failed to consume restartTo, dropping it", { error }) this.restartTo = undefined + this.notifyEventWaiters() return } } @@ -429,8 +451,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[] = [] @@ -452,8 +478,8 @@ export class FlowcoreDataPump { } if (!events.length) { - await this.waitForEvents() - return this.reserve(amount) + await this.waitForEvents(generation) + return this.reserveInternal(amount, generation) } this.updateMetricsGauges() @@ -568,7 +594,7 @@ export class FlowcoreDataPump { if (reopenedEvents.length) { this.logger?.info(`Reopened ${reopenedEvents.length} events`) - await this.waiterEvents?.() + this.notifyEventWaiters() } if (!failedEvents.length) { @@ -602,27 +628,79 @@ 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.activeProcessLoopGeneration === this.processLoopGeneration || + this.processLoopBackoffGeneration === this.processLoopGeneration + ) { + 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(() => { - if (!this.running) return - this.startProcessLoop() - }, delay) - }) + const generation = this.processLoopGeneration + if ( + !this.options.processor || + !this.isCurrentProcessLoop(generation) || + this.activeProcessLoopGeneration === generation || + this.processLoopBackoffGeneration === generation + ) { + return + } + this.activeProcessLoopGeneration = generation + this.processLoop(generation) + .then(() => { + 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.isCurrentProcessLoop(generation)) { + this.startProcessLoop() + } + }) + .catch((error) => { + 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})`) + 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) + }) } - 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) { @@ -751,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 1c4356e..9b02561 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" }), @@ -111,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({ @@ -132,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 { @@ -200,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() @@ -448,3 +550,360 @@ 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", () => { + 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() + }) + + 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(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(secondEventId)], 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 === firstEventId) { + 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], 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(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 let a pre-restart handler checkpoint the replayed event", async () => { + const { logger } = createMockLogger() + const firstHandler = deferred() + const replayHandler = deferred() + const checkpoints: FlowcoreDataPumpState[] = [] + let handlerCalls = 0 + let batch = 0 + const fakeDataSource = new FakeDataSource({ + getEventsImpl: () => { + batch++ + if (batch <= 2) return Promise.resolve({ events: [event(firstEventId)], 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: { + getState: () => ({ timeBucket: "20260331120000" }), + setState: (state) => { + checkpoints.push(state) + }, + }, + bufferSize: 1, + processor: { + concurrency: 1, + handler: () => { + handlerCalls++ + return handlerCalls === 1 ? firstHandler.promise : replayHandler.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") + + // 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 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") + + // 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("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() + 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() + 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