fix: reclaim expired Bun completion markers and share KV initialization

.github/workflows/bun-completion-expiry.yml

@@ -0,0 +1,25 @@
+name: Bun completion expiry
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  bun-completion-expiry:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - uses: denoland/setup-deno@v2
+        with:
+          deno-version: v2.x
+      - uses: oven-sh/setup-bun@v2
+        with:
+          bun-version: 1.3.13
+      - run: deno install --frozen-lockfile
+      - run: deno task build:npm
+      - run: bun test bun-tests/pathways-kv-expiry.test.ts

bun-tests/pathways-kv-expiry.test.ts

@@ -0,0 +1,264 @@
+import { afterEach, describe, expect, spyOn, test } from "bun:test"
+import { resolve } from "node:path"
+import { z } from "../npm/node_modules/zod/index.js"
+
+// Exercise both published runtime paths, not a reimplementation of the adapter.
+const root = resolve(import.meta.dir, "../npm")
+const stores: { close(): void }[] = []
+let clock: ReturnType<typeof spyOn> | undefined
+afterEach(() => {
+  clock?.mockRestore()
+  clock = undefined
+  for (const store of stores.splice(0)) store.close()
+})
+
+for (const format of ["esm", "script"]) {
+  describe(`Pathways ${format} Bun completion expiry`, () => {
+    async function state() {
+      const { InternalPathwayState } = await import(`${root}/${format}/pathways/internal-pathway.state.js`)
+      const result = new InternalPathwayState()
+      await result.isProcessed("initial-probe")
+      expect(result.kv.constructor.name).toBe("BunKvAdapter")
+      stores.push(result.kv.store)
+      return result
+    }
+
+    test("physically reclaims unique expired IDs and reuses SQLite pages", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const s = await state()
+      const pages: number[] = []
+      const rows: number[] = []
+      for (let cycle = 0; cycle < 12; cycle++) {
+        for (let i = 0; i < 5000; i++) {
+          await s.setProcessed(`event-${String(cycle).padStart(2, "0")}-${String(i).padStart(5, "0")}`)
+        }
+        rows.push(s.kv.store.getCount())
+        pages.push(s.kv.store.db.query("PRAGMA page_count").get().page_count)
+        now += 360_001
+      }
+      console.log(JSON.stringify({ format, rows, pages, rss: process.memoryUsage().rss }))
+      expect(Math.max(...rows)).toBe(5000)
+      expect(Math.max(...pages.slice(3)) - Math.min(...pages.slice(3))).toBeLessThanOrEqual(3)
+      await s.isProcessed("never-seen")
+      expect(s.kv.store.getCount()).toBe(0)
+      expect(s.kv.store.db.query("PRAGMA freelist_count").get().freelist_count).toBeGreaterThan(0)
+    })
+
+    test("bounds a continuous stream across overlapping TTL windows without early eviction", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const s = await state()
+      let highWater = 0
+      for (let second = 0; second < 1800; second++) {
+        now = 1_000_000 + second * 1000
+        for (let i = 0; i < 10; i++) await s.setProcessed(`stream-${second}-${i}`)
+        if (second >= 300) expect(await s.isProcessed(`stream-${second - 300}-0`)).toBe(true)
+        if (second > 300) expect(await s.isProcessed(`stream-${second - 301}-0`)).toBe(false)
+        highWater = Math.max(highWater, s.kv.store.getCount())
+      }
+      expect(highWater).toBeLessThanOrEqual(3610)
+      console.log(JSON.stringify({ format, continuousStreamHighWater: highWater }))
+    })
+
+    test("retries a rejected initialization and handles clock rollback", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const { InternalPathwayState } = await import(`${root}/${format}/pathways/internal-pathway.state.js`)
+      const s = new InternalPathwayState()
+      s.kvPromise = Promise.reject(new Error("synthetic initialization failure"))
+      const rejected = await Promise.allSettled(Array.from({ length: 32 }, () => s.getKv()))
+      expect(rejected.every((result) => result.status === "rejected")).toBe(true)
+      expect(s.kvPromise).toBeUndefined()
+      const recovered = await Promise.all(Array.from({ length: 32 }, () => s.getKv()))
+      expect(new Set(recovered).size).toBe(1)
+      await s.setProcessed("retained")
+      stores.push(s.kv.store)
+      const purge = spyOn(s.kv.store, "deleteExpired")
+      try {
+        now -= 1000
+        expect(await s.isProcessed("retained")).toBe(true)
+        expect(purge).toHaveBeenCalledTimes(1)
+      } finally {
+        purge.mockRestore()
+      }
+    })
+
+    test("creates no background timer requiring shutdown", async () => {
+      const interval = spyOn(globalThis, "setInterval")
+      const timeout = spyOn(globalThis, "setTimeout")
+      try {
+        const s = await state()
+        await s.setProcessed("local-only")
+        expect(interval).not.toHaveBeenCalled()
+        expect(timeout).not.toHaveBeenCalled()
+      } finally {
+        interval.mockRestore()
+        timeout.mockRestore()
+      }
+    })
+
+    test("keeps the existing strict expiry boundary and full five-minute TTL", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const s = await state()
+      await s.setProcessed("duplicate")
+      s.kv.set("permanent", true)
+      now += 299_999
+      expect(await s.isProcessed("duplicate")).toBe(true)
+      now++
+      expect(await s.isProcessed("duplicate")).toBe(true)
+      now++
+      expect(await s.isProcessed("duplicate")).toBe(false)
+      expect(s.kv.get("permanent")).toBe(true)
+      await s.setProcessed("duplicate")
+      expect(await s.isProcessed("duplicate")).toBe(true)
+    })
+
+    test("throttles scans and reclaims on first access after idle without timers", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const s = await state()
+      const purge = spyOn(s.kv.store, "deleteExpired")
+      try {
+        for (let i = 0; i < 100; i++) await s.setProcessed(`id-${i}`)
+        expect(purge).toHaveBeenCalledTimes(0)
+        now += 60_000
+        await s.isProcessed("unknown")
+        expect(purge).toHaveBeenCalledTimes(1)
+        now += 600_000
+        expect(s.kv.store.getCount()).toBe(100)
+        await s.setProcessed("new")
+        expect(s.kv.store.getCount()).toBe(1)
+        expect(purge).toHaveBeenCalledTimes(2)
+      } finally {
+        purge.mockRestore()
+      }
+    })
+
+    test("concurrent initialization shares one adapter and loses no completions", async () => {
+      const { InternalPathwayState } = await import(`${root}/${format}/pathways/internal-pathway.state.js`)
+      const s = new InternalPathwayState()
+      const adapters = await Promise.all(Array.from({ length: 32 }, () => s.getKv()))
+      for (const adapter of new Set(adapters)) stores.push(adapter.store)
+      expect(new Set(adapters).size).toBe(1)
+      await Promise.all(Array.from({ length: 32 }, (_, i) => s.setProcessed(`parallel-${i}`)))
+      for (let i = 0; i < 32; i++) expect(await s.isProcessed(`parallel-${i}`)).toBe(true)
+    })
+
+    test("cleanup failure preserves reads and writes, throttles failures and later recovers", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const s = await state()
+      await s.setProcessed("old")
+      await s.setProcessed("untouched-expired")
+      now += 360_001
+      const purge = spyOn(s.kv.store, "deleteExpired").mockImplementationOnce(() => {
+        throw new Error("synthetic cleanup failure")
+      })
+      try {
+        await s.setProcessed("new")
+        expect(s.kv.store.getCount()).toBe(3)
+        expect(await s.isProcessed("new")).toBe(true)
+        expect(await s.isProcessed("old")).toBe(false)
+        for (let i = 0; i < 100; i++) await s.setProcessed(`during-failure-${i}`)
+        expect(purge).toHaveBeenCalledTimes(1)
+        now += 60_000
+        expect(await s.isProcessed("new")).toBe(true)
+        expect(s.kv.store.getCount()).toBe(101)
+        expect(purge).toHaveBeenCalledTimes(2)
+      } finally {
+        purge.mockRestore()
+      }
+    })
+
+    for (const handlerFails of [false, true]) {
+      test(`real builder preserves handler retry behavior during persistent cleanup failure (handlerFails=${handlerFails})`, async () => {
+        let now = 1_000_000
+        clock = spyOn(Date, "now").mockImplementation(() => now)
+        const s = await state()
+        // The published CJS builder cannot load its ESM-only file-type dependency.
+        // Exercise the application's ESM builder with each real state/adapter format.
+        const { PathwaysBuilder } = await import(`${root}/esm/pathways/builder.js`)
+        const builder = new PathwaysBuilder({
+          tenant: "synthetic",
+          dataCore: "synthetic",
+          apiKey: "fc_synthetic_only",
+          baseUrl: "https://example.invalid",
+          runtimeEnv: "test",
+          pathwayMode: "virtual",
+          autoProvision: false,
+        }).withPathwayState(s)
+        builder.register({
+          flowType: "test.0",
+          eventType: "event.0",
+          schema: z.object({}),
+          writable: false,
+          subscribe: false,
+          maxRetries: 1,
+          retryDelayMs: 0,
+        })
+        let deliveries = 0
+        const handlerError = new Error("synthetic handler failure")
+        builder.handle("test.0/event.0", async () => {
+          await Promise.resolve()
+          deliveries++
+          // Cleanup becomes due after delivery, exactly where completion is written.
+          now += 60_000
+          if (handlerFails) throw handlerError
+        })
+        const purge = spyOn(s.kv.store, "deleteExpired").mockImplementation(() => {
+          throw new Error("synthetic cleanup failure")
+        })
+        try {
+          const pending = builder.process("test.0/event.0", { eventId: "delivery", payload: {}, metadata: {} })
+          if (handlerFails) await expect(pending).rejects.toBe(handlerError)
+          else await pending
+          expect(deliveries).toBe(handlerFails ? 2 : 1)
+          // Current main retains terminal failures for redelivery; only success is marked.
+          expect(await s.isProcessed("delivery")).toBe(!handlerFails)
+          expect(purge).toHaveBeenCalledTimes(1)
+          now += 60_000
+          expect(await s.isProcessed("delivery")).toBe(!handlerFails)
+          expect(purge).toHaveBeenCalledTimes(2)
+          if (handlerFails) {
+            await expect(builder.process("test.0/event.0", {
+              eventId: "delivery",
+              payload: {},
+              metadata: {},
+            })).rejects.toBe(handlerError)
+            expect(deliveries).toBe(4)
+            expect(await s.isProcessed("delivery")).toBe(false)
+          }
+        } finally {
+          purge.mockRestore()
+        }
+      })
+    }
+
+    test("real KV read and completion write failures still reject", async () => {
+      let now = 1_000_000
+      clock = spyOn(Date, "now").mockImplementation(() => now)
+      const s = await state()
+      now += 60_000
+      const purge = spyOn(s.kv.store, "deleteExpired").mockImplementation(() => {
+        throw new Error("synthetic cleanup failure")
+      })
+      const write = spyOn(s.kv.store, "set").mockImplementation(() => {
+        throw new Error("synthetic write failure")
+      })
+      const read = spyOn(s.kv.store, "get").mockImplementation(() => {
+        throw new Error("synthetic read failure")
+      })
+      try {
+        await expect(s.setProcessed("not-completed")).rejects.toThrow("synthetic write failure")
+        await expect(s.isProcessed("not-completed")).rejects.toThrow("synthetic read failure")
+      } finally {
+        purge.mockRestore()
+        write.mockRestore()
+        read.mockRestore()
+      }
+      expect(await s.isProcessed("not-completed")).toBe(false)
+    })
+  })
+}

deno.json

@@ -27,7 +27,7 @@
     "file-type": "npm:file-type@^21.0.0",
     "node-cache": "npm:node-cache@^5.1.2",
     "rxjs": "npm:rxjs@^7.8.1",
-    "@flowcore/data-pump": "npm:@flowcore/data-pump@^0.23.3",
+    "@flowcore/data-pump": "npm:@flowcore/data-pump@^0.23.4",
     "@flowcore/sdk": "npm:@flowcore/sdk@^4.2.4",
     "postgres": "npm:postgres@^3.4.3",
     "ws": "npm:ws@^8.18.0",

src/pathways/internal-pathway.state.ts

@@ -17,6 +17,7 @@ export class InternalPathwayState implements PathwayState {
    * @private
    */
   private kv: KvAdapter | null = null
+  private kvPromise?: Promise<KvAdapter>
 
   /**
    * Gets or initializes the KV adapter
@@ -25,10 +26,15 @@ export class InternalPathwayState implements PathwayState {
    * @returns The KV adapter instance
    */
   private async getKv(): Promise<KvAdapter> {
-    if (!this.kv) {
-      this.kv = await createKvAdapter()
+    if (this.kv) return this.kv
+    const pending = this.kvPromise ??= createKvAdapter()
+    try {
+      this.kv = await pending
+      return this.kv
+    } finally {
+      // Share initialization with all callers, but allow retry after rejection.
+      if (this.kvPromise === pending) this.kvPromise = undefined
     }
-    return this.kv
   }
 
   /**

src/pathways/kv/bun-kv-adapter.ts

@@ -12,6 +12,21 @@ export class BunKvAdapter implements KvAdapter {
    * @private
    */
   private store: BunSqliteKeyValue
+  private static readonly PURGE_INTERVAL_MS = 60_000
+  private lastPurgeAttempt = Date.now()
+
+  private purgeExpired(): void {
+    const now = Date.now()
+    if (now >= this.lastPurgeAttempt && now - this.lastPurgeAttempt < BunKvAdapter.PURGE_INTERVAL_MS) return
+    // Throttle attempts, including failures. Maintenance must not turn an already
+    // delivered event's completion write into a handler retry and duplicate delivery.
+    this.lastPurgeAttempt = now
+    try {
+      this.store.deleteExpired()
+    } catch {
+      // Best effort only; actual KV read/write errors still propagate below.
+    }
+  }
 
   /**
    * Creates a new in-memory Bun KV adapter
@@ -28,6 +43,7 @@ export class BunKvAdapter implements KvAdapter {
    * @returns The stored value or null if not found
    */
   get<T>(key: string): T | null {
+    this.purgeExpired()
     const value = this.store.get(key)
     return value as T | null
   }
@@ -40,6 +56,7 @@ export class BunKvAdapter implements KvAdapter {
    * @param ttlMs Time-to-live in milliseconds
    */
   set(key: string, value: unknown, ttlMs: number): void {
+    this.purgeExpired()
     this.store.set(key, value, ttlMs)
   }
 }