From d88c23b9601c56f7d66accb32651b83cc66906da Mon Sep 17 00:00:00 2001 From: olavurellefsen Date: Tue, 6 Oct 2026 21:52:52 +0200 Subject: [PATCH 1/2] fix(cluster): keep retrying the leader bootstrap while holding the lease When an instance won the leader lease and registering the virtual pathway failed (e.g. a 502 from the control plane for longer than the provisioning retries cover), bootstrapLeaderPump stopped the pump and threw. The error was only logged, while the cluster manager kept renewing the lease, so no instance ran a pump until the process restarted. Retry the bootstrap with backoff (1s, 2s, 5s, 10s, then every 30s) for as long as this instance is still the current leader. A leadership change bumps a generation counter so stale retry loops stop. Seen in production on klokkan: events were written but not delivered for about five hours. Co-Authored-By: Claude --- src/pathways/builder.ts | 36 ++++++++- tests/pump-runtime-behavior.test.ts | 113 ++++++++++++++++++++++++++++ 2 files changed, 148 insertions(+), 1 deletion(-) diff --git a/src/pathways/builder.ts b/src/pathways/builder.ts index 193162e..310c965 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/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() + } + }) + }, +}) From 485e96f15435327c57f862cb8863c0d3db786fb6 Mon Sep 17 00:00:00 2001 From: olavurellefsen Date: Sun, 1 Nov 2026 14:04:14 +0000 Subject: [PATCH 2/2] fix(pump): recover partially started sources with bounded retries --- bun-tests/pathways-kv-expiry.test.ts | 1 + src/pathways/pump/pathway-pump.ts | 39 +++++++-- tests/pump-initial-start-recovery.test.ts | 100 ++++++++++++++++++++++ 3 files changed, 133 insertions(+), 7 deletions(-) create mode 100644 tests/pump-initial-start-recovery.test.ts 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/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() + } + }, +})