fix(pump): retry leader and source bootstrap failures

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.

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,20 +2234,50 @@ export class PathwaysBuilder<
   }
 
   private async handleLeadershipChange(isLeader: boolean): Promise<void> {
+    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
     }
 
     this.logger.info("Lost leadership, stopping leader runtime")
     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<void> {
+    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("/")

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)

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()
+    }
+  },
+})

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("<html>502 Bad Gateway</html>", { 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<void>
+  }
+  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()
+      }
+    })
+  },
+})