From 60985470d988e1a1ad54642ac6f246e2a7230712 Mon Sep 17 00:00:00 2001 From: Globe Tracker Assistant Date: Thu, 8 Oct 2026 13:34:21 +0100 Subject: [PATCH] fix: reclaim expired Bun completion markers and share KV initialization --- .github/workflows/bun-completion-expiry.yml | 25 ++ bun-tests/pathways-kv-expiry.test.ts | 264 ++++++++++++++++++++ deno.json | 2 +- deno.lock | 8 +- src/pathways/internal-pathway.state.ts | 12 +- src/pathways/kv/bun-kv-adapter.ts | 17 ++ 6 files changed, 320 insertions(+), 8 deletions(-) create mode 100644 .github/workflows/bun-completion-expiry.yml create mode 100644 bun-tests/pathways-kv-expiry.test.ts diff --git a/.github/workflows/bun-completion-expiry.yml b/.github/workflows/bun-completion-expiry.yml new file mode 100644 index 0000000..0ca2ab4 --- /dev/null +++ b/.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 diff --git a/bun-tests/pathways-kv-expiry.test.ts b/bun-tests/pathways-kv-expiry.test.ts new file mode 100644 index 0000000..63afa3c --- /dev/null +++ b/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 | 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) + }) + }) +} diff --git a/deno.json b/deno.json index 0a8a808..f382c77 100644 --- a/deno.json +++ b/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", diff --git a/deno.lock b/deno.lock index 7924109..a0dfd21 100644 --- a/deno.lock +++ b/deno.lock @@ -22,7 +22,7 @@ "jsr:@std/path@~0.225.2": "0.225.2", "jsr:@ts-morph/bootstrap@0.24": "0.24.0", "jsr:@ts-morph/common@0.24": "0.24.0", - "npm:@flowcore/data-pump@~0.23.3": "0.23.3", + "npm:@flowcore/data-pump@~0.23.4": "0.23.4", "npm:@flowcore/sdk-transformer-core@^2.5.1": "2.5.1", "npm:@flowcore/sdk@^4.2.4": "4.2.4", "npm:bun-sqlite-key-value@^1.13.1": "1.13.1_typescript@5.9.3", @@ -144,8 +144,8 @@ "@date-fns/utc@2.1.1": { "integrity": "sha512-SlJDfG6RPeEX8wEVv6ZB3kak4MmbtyiI2qX/5zuKdordbrhB/iaJ58GVMZgJ6P1sJaM1gMgENFYYeg1JWrCFrA==" }, - "@flowcore/data-pump@0.23.3": { - "integrity": "sha512-J31pnN9E9zcf1FkZFHuKfy/lW83eLVaGY9/Z8VPwb/7h2oeSv43SmXPueRL5FbhpXPgpPrRqfHXtbx/RerLM5w==", + "@flowcore/data-pump@0.23.4": { + "integrity": "sha512-qXRahSk7q8lm56+cLT7ZJF0QNjg2/lKX7btbzvoZiOXcRblor1Y5tFHm0Z0tXG0UwMa9URrC53RJIGZmmw+r0g==", "dependencies": [ "@date-fns/utc", "@flowcore/sdk@6.1.0", @@ -372,7 +372,7 @@ "workspace": { "dependencies": [ "jsr:@deno/dnt@~0.41.3", - "npm:@flowcore/data-pump@~0.23.3", + "npm:@flowcore/data-pump@~0.23.4", "npm:@flowcore/sdk-transformer-core@^2.5.1", "npm:@flowcore/sdk@^4.2.4", "npm:bun-sqlite-key-value@^1.13.1", diff --git a/src/pathways/internal-pathway.state.ts b/src/pathways/internal-pathway.state.ts index cd109df..47285b0 100644 --- a/src/pathways/internal-pathway.state.ts +++ b/src/pathways/internal-pathway.state.ts @@ -17,6 +17,7 @@ export class InternalPathwayState implements PathwayState { * @private */ private kv: KvAdapter | null = null + private kvPromise?: Promise /** * Gets or initializes the KV adapter @@ -25,10 +26,15 @@ export class InternalPathwayState implements PathwayState { * @returns The KV adapter instance */ private async getKv(): Promise { - 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 } /** diff --git a/src/pathways/kv/bun-kv-adapter.ts b/src/pathways/kv/bun-kv-adapter.ts index 60f893a..19a3093 100644 --- a/src/pathways/kv/bun-kv-adapter.ts +++ b/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(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) } }