diff --git a/bun-tests/pathways-kv-expiry.test.ts b/bun-tests/pathways-kv-expiry.test.ts index 63afa3c..4a08f29 100644 --- a/bun-tests/pathways-kv-expiry.test.ts +++ b/bun-tests/pathways-kv-expiry.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, spyOn, test } from "bun:test" import { resolve } from "node:path" +import process from "node:process" import { z } from "../npm/node_modules/zod/index.js" // Exercise both published runtime paths, not a reimplementation of the adapter. diff --git a/src/pathways/builder.ts b/src/pathways/builder.ts index 961b334..b7d6d59 100644 --- a/src/pathways/builder.ts +++ b/src/pathways/builder.ts @@ -515,6 +515,10 @@ export class PathwaysBuilder< private currentPumpProvisionsPathway = false private currentPumpUsesExplicitPulse = false private currentPumpUsesAutoPulse = false + // Bumped on every leadership change so a stale bootstrap retry loop stops. + private leadershipGeneration = 0 + // Backoff between leader bootstrap attempts; the last delay repeats. + private leaderBootstrapRetryDelaysMs = [1_000, 2_000, 5_000, 10_000, 30_000] /** * Creates a new PathwaysBuilder instance @@ -2230,13 +2234,15 @@ export class PathwaysBuilder< } private async handleLeadershipChange(isLeader: boolean): Promise { + const generation = ++this.leadershipGeneration + if (isLeader) { if (!this.pathwayPump) { return } this.logger.info("Became leader, bootstrapping pump") - await this.bootstrapLeaderPump() + await this.bootstrapLeaderPumpUntilReady(generation) return } @@ -2244,6 +2250,34 @@ export class PathwaysBuilder< await this.stopLeaderRuntime() } + /** + * Keeps retrying the leader bootstrap while this instance still holds the + * lease. Giving up would leave the lease held with no pump running, so no + * instance in the cluster would deliver events until a restart. + */ + private async bootstrapLeaderPumpUntilReady(generation: number): Promise { + const isCurrentLeader = () => generation === this.leadershipGeneration && this.clusterManager?.isLeader === true + + for (let attempt = 1;; attempt++) { + try { + await this.bootstrapLeaderPump() + return + } catch (err) { + if (!isCurrentLeader()) return + + const delays = this.leaderBootstrapRetryDelaysMs + const delayMs = delays[Math.min(attempt, delays.length) - 1] + this.logger.warn("Leader bootstrap failed, retrying", { + attempt, + delayMs, + error: err instanceof Error ? err.message : String(err), + }) + await new Promise((resolve) => setTimeout(resolve, delayMs)) + if (!isCurrentLeader()) return + } + } + } + private buildRegistrations(): ProvisionerRegistration[] { return Object.keys(this.pathways).map((key) => { const [flowType, eventType] = key.split("/") diff --git a/src/pathways/pump/pathway-pump.ts b/src/pathways/pump/pathway-pump.ts index de3b173..80fff02 100644 --- a/src/pathways/pump/pathway-pump.ts +++ b/src/pathways/pump/pathway-pump.ts @@ -281,7 +281,19 @@ export class PathwayPump { for (const meta of groups.values()) { this.groupMeta.set(groupKey(meta.flowType, meta.pumpGroup), meta) - await this.startPumpForGroup(meta) + try { + await this.startPumpForGroup(meta) + } catch (error) { + this.logger.error( + "Initial data pump startup failed; retrying", + error instanceof Error ? error : new Error(String(error)), + { + flowType: meta.flowType, + pumpGroup: meta.pumpGroup, + }, + ) + if (this.running) this.scheduleRestart(meta) + } } } @@ -392,13 +404,26 @@ export class PathwayPump { this.pumps.set(key, pump) - await pump.start((error?: Error) => { - if (error) { - this.logger.error(`Data pump error`, error, { flowType, pumpGroup }) - if (!this.running) return - this.scheduleRestart(meta) + try { + await pump.start((error?: Error) => { + if (error) { + this.logger.error(`Data pump error`, error, { flowType, pumpGroup }) + if (!this.running) return + this.scheduleRestart(meta) + } + }) + } catch (error) { + try { + await pump.stop() + } catch { + this.logger.warn("Failed to stop partially started pump", { + flowType, + pumpGroup, + }) } - }) + if (this.pumps.get(key) === pump) this.pumps.delete(key) + throw error + } // Successful start: reset the per-group attempt counter. this.restartAttempts.set(key, 0) diff --git a/tests/pump-initial-start-recovery.test.ts b/tests/pump-initial-start-recovery.test.ts new file mode 100644 index 0000000..361cefb --- /dev/null +++ b/tests/pump-initial-start-recovery.test.ts @@ -0,0 +1,100 @@ +import { assertEquals } from "https://deno.land/std@0.224.0/assert/mod.ts" +import { FakeTime } from "https://deno.land/std@0.224.0/testing/time.ts" +import { FlowcoreDataPump } from "@flowcore/data-pump" +import { PathwayPump } from "../src/pathways/pump/pathway-pump.ts" + +async function flushStartupPromises() { + for (let step = 0; step < 20; step++) await Promise.resolve() +} + +Deno.test({ + name: "initial source startup failure stops the half-started pump, starts peers, and retries to recovery", + sanitizeOps: false, + sanitizeResources: false, + fn: async () => { + const time = new FakeTime() + const original = FlowcoreDataPump.create + const created: Array<{ flow: string; stopped: boolean }> = [] + let failures = 0 + FlowcoreDataPump.create = ((options: { dataSource: { flowType: string } }) => { + const record = { flow: options.dataSource.flowType, stopped: false } + created.push(record) + return { + async start() { + if (record.flow === "first.0" && failures++ < 2) throw new Error("DataCoreFetchCommand: transient 502") + }, + async stop() { + record.stopped = true + }, + } + }) as unknown as typeof FlowcoreDataPump.create + const pump = new PathwayPump({ stateManagerFactory: () => ({ getState: () => null, setState: () => {} }) }) + pump.configure({ + tenant: "synthetic", + dataCore: "synthetic", + apiKey: "fc_synthetic_synthetic", + baseUrl: "https://invalid.example", + processEvent: async () => {}, + }) + try { + await pump.start([{ flowType: "first.0", eventType: "first.created.0" }, { + flowType: "second.0", + eventType: "second.created.0", + }]) + assertEquals(created.map((item) => item.flow), ["first.0", "second.0"]) + assertEquals(created[0].stopped, true) + await time.tickAsync(1_001) + await flushStartupPromises() + assertEquals(created.length, 3) + assertEquals(created[2].stopped, true) + await time.tickAsync(2_001) + await flushStartupPromises() + assertEquals(created.length, 4) + assertEquals(created[3].stopped, false) + await pump.stop() + await time.tickAsync(60_000) + assertEquals(created.length, 4) + assertEquals(created.every((item) => item.stopped), true) + } finally { + await pump.stop() + FlowcoreDataPump.create = original + time.restore() + } + }, +}) + +Deno.test({ + name: "stop cancels the initial-start retry before it creates another source pump", + sanitizeOps: false, + sanitizeResources: false, + fn: async () => { + const time = new FakeTime() + const original = FlowcoreDataPump.create + let starts = 0 + FlowcoreDataPump.create = (() => ({ + async start() { + starts++ + throw new Error("temporary 401") + }, + async stop() {}, + })) as unknown as typeof FlowcoreDataPump.create + const pump = new PathwayPump({ stateManagerFactory: () => ({ getState: () => null, setState: () => {} }) }) + pump.configure({ + tenant: "synthetic", + dataCore: "synthetic", + apiKey: "fc_synthetic_synthetic", + baseUrl: "https://invalid.example", + processEvent: async () => {}, + }) + try { + await pump.start([{ flowType: "first.0", eventType: "first.created.0" }]) + await pump.stop() + await time.tickAsync(60_000) + assertEquals(starts, 1) + } finally { + await pump.stop() + FlowcoreDataPump.create = original + time.restore() + } + }, +}) diff --git a/tests/pump-runtime-behavior.test.ts b/tests/pump-runtime-behavior.test.ts index dfe5254..157d5fb 100644 --- a/tests/pump-runtime-behavior.test.ts +++ b/tests/pump-runtime-behavior.test.ts @@ -842,3 +842,116 @@ Deno.test({ }) }, }) + +function stubLeaderBootstrap(failingRegistrations: number) { + const pump = { starts: 0, stops: 0, registrations: 0 } + const stubs = [ + stub(PathwayProvisioner.prototype, "provision", async () => {}), + stub(PathwayPump.prototype, "start", async function (this: PathwayPump) { + pump.starts++ + ;(this as unknown as { running: boolean }).running = true + }), + stub(PathwayPump.prototype, "stop", async function (this: PathwayPump) { + if (!(this as unknown as { running: boolean }).running) return + pump.stops++ + ;(this as unknown as { running: boolean }).running = false + }), + stub(PathwayPump.prototype, "setPulseConfig", async () => {}), + stub(CommandPoller.prototype, "start", () => {}), + stub(CommandPoller.prototype, "stop", () => {}), + stub(globalThis, "fetch", async (_input, init) => { + if (((init as RequestInit | undefined)?.method ?? "GET") === "GET") { + return new Response(JSON.stringify({ deliveryState: "active", deliveryPauseTargets: null }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }) + } + pump.registrations++ + if (pump.registrations <= failingRegistrations) { + return new Response("502 Bad Gateway", { status: 502 }) + } + return new Response(JSON.stringify({ pathwayId: crypto.randomUUID(), status: "created" }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }) + }), + ] + return { pump, restore: () => stubs.forEach((s) => s.restore()) } +} + +async function createFollowerBuilder() { + const builder = createBuilder({ + runtimeEnv: "production", + pathwayName: "virtual-service", + pathwayMode: "virtual", + autoProvision: { pathway: true }, + // One attempt per bootstrap so the counts below measure leader-level retries. + provisionRetry: { maxAttempts: 1 }, + }) + const clusterManager = { isRunning: true, isLeader: false } + const internals = builder as unknown as { + clusterManager: typeof clusterManager + leaderBootstrapRetryDelaysMs: number[] + handleLeadershipChange(isLeader: boolean): Promise + } + internals.clusterManager = clusterManager + internals.leaderBootstrapRetryDelaysMs = [1, 1, 1] + await builder.startPump(createPumpOptions()) + return { builder, clusterManager, internals } +} + +Deno.test({ + name: "PathwaysBuilder leader bootstrap retries", + sanitizeResources: false, + sanitizeOps: false, + fn: async (t) => { + await t.step("a new leader retries a failed pathway registration until the pump runs", async () => { + const { pump, restore } = stubLeaderBootstrap(2) + try { + const { builder, clusterManager, internals } = await createFollowerBuilder() + + clusterManager.isLeader = true + await internals.handleLeadershipChange(true) + + assertEquals(pump.registrations, 3) + assertEquals(builder.pump?.isRunning, true) + } finally { + restore() + } + }) + + await t.step("retries stop once leadership is lost", async () => { + const { pump, restore } = stubLeaderBootstrap(Number.POSITIVE_INFINITY) + try { + const { builder, clusterManager, internals } = await createFollowerBuilder() + + clusterManager.isLeader = true + const bootstrap = internals.handleLeadershipChange(true) + clusterManager.isLeader = false + await bootstrap + + assertEquals(pump.registrations, 1) + assertEquals(builder.pump?.isRunning, false) + } finally { + restore() + } + }) + + await t.step("a superseded bootstrap stops retrying after leadership is regained", async () => { + const { pump, restore } = stubLeaderBootstrap(1) + try { + const { builder, clusterManager, internals } = await createFollowerBuilder() + + clusterManager.isLeader = true + const first = internals.handleLeadershipChange(true) + const second = internals.handleLeadershipChange(true) + await Promise.all([first, second]) + + assertEquals(pump.registrations, 2) + assertEquals(builder.pump?.isRunning, true) + } finally { + restore() + } + }) + }, +})