From a4b5c30b5ee26bdd9f1117493acc4ff4a630c4bc Mon Sep 17 00:00:00 2001 From: cyphercodes Date: Sun, 2 Aug 2026 08:46:47 +0300 Subject: [PATCH 1/2] fix(cloudflare): retry catalog load after failure --- src/server/cloudflare.test.ts | 39 +++++++++++++++++++++++++++++++++++ src/server/cloudflare.ts | 28 ++++++++++++++++++++----- 2 files changed, 62 insertions(+), 5 deletions(-) diff --git a/src/server/cloudflare.test.ts b/src/server/cloudflare.test.ts index edbc1be2..0e34b84b 100644 --- a/src/server/cloudflare.test.ts +++ b/src/server/cloudflare.test.ts @@ -25,6 +25,45 @@ describe("cloudflare worker", () => { vi.restoreAllMocks(); }); + it("deduplicates concurrent app creation and retries after a transient catalog failure", async () => { + vi.resetModules(); + const { default: isolatedWorker } = await import("./cloudflare.ts"); + const fallback = memoryAssets(chunkedCatalog()); + let indexAttempts = 0; + const assets: AssetsBinding = { + async fetch(request) { + if (new URL(request.url).pathname === "/catalog/index.json" && ++indexAttempts === 1) { + return new Response("upstream", { status: 500 }); + } + return fallback.fetch(request); + }, + }; + const env: CloudflareEnv = { + DB: new UnusedD1Database(), + TRANSIT_FILES: new UnusedR2Bucket(), + ASSETS: assets, + }; + const request = (): Request => new Request("https://catalog-retry.example.com/api/auth/session"); + const failures = await Promise.allSettled([ + isolatedWorker.fetch(request(), env, createExecutionContext()), + isolatedWorker.fetch(request(), env, createExecutionContext()), + ]); + + for (const failure of failures) { + expect(failure).toMatchObject({ + status: "rejected", + reason: { + message: "Cloudflare asset catalog request failed: /catalog/index.json returned 500", + }, + }); + } + expect(indexAttempts).toBe(1); + + const response = await isolatedWorker.fetch(request(), env, createExecutionContext()); + expect(response.status).toBe(200); + expect(indexAttempts).toBe(2); + }); + it("writes connection logs to console", async () => { const info = vi.spyOn(console, "info").mockImplementation(() => {}); const response = await worker.fetch( diff --git a/src/server/cloudflare.ts b/src/server/cloudflare.ts index 9d70d694..6897138c 100644 --- a/src/server/cloudflare.ts +++ b/src/server/cloudflare.ts @@ -33,11 +33,19 @@ export default { setPrivateNetworkAccessAllowed(parsePrivateNetworkAccessFlag(env.OOMOL_CONNECT_ALLOW_PRIVATE_NETWORK)); const publicOrigin = resolvePublicOrigin(request, env); const cacheKey = createCacheKey(env, publicOrigin); - if (!cachedApp || cachedApp.key !== cacheKey) { - cachedApp = { key: cacheKey, app: createCloudflareApp(env, publicOrigin) }; + let appPromise = cachedApp?.key === cacheKey ? cachedApp.app : undefined; + if (!appPromise) { + const createdApp = createCloudflareApp(env, publicOrigin); + cachedApp = { key: cacheKey, app: createdApp }; + void createdApp.catch(() => { + if (cachedApp?.app === createdApp) { + cachedApp = undefined; + } + }); + appPromise = createdApp; } - const { app } = await cachedApp.app; + const { app } = await appPromise; const response = await app.fetch(request, env); if (response.status === 404 && env.ASSETS && shouldServeAsset(request)) { return env.ASSETS.fetch(request); @@ -119,10 +127,20 @@ function writeWorkerLog(level: "error" | "info" | "warn"): (fields: unknown, mes } function loadCatalogOnce(assets: AssetsBinding): Promise { - catalogPromise ??= loadCatalogFromAssets(assets, { + if (catalogPromise) { + return catalogPromise; + } + + const createdCatalog = loadCatalogFromAssets(assets, { executableServices: Object.keys(executorModules), }); - return catalogPromise; + catalogPromise = createdCatalog; + void createdCatalog.catch(() => { + if (catalogPromise === createdCatalog) { + catalogPromise = undefined; + } + }); + return createdCatalog; } function createSecretCodec(encryptionKey: string | undefined): Promise { From ba3b600c1b22479d245f5e425f7369ca233c4e0a Mon Sep 17 00:00:00 2001 From: Kevin Cui Date: Sun, 2 Aug 2026 03:59:02 -0400 Subject: [PATCH 2/2] refactor(cloudflare): share one evict-on-rejection promise cache The app and catalog fixes each open-coded the same slot: memoize a promise, then drop it if it rejects so the isolate self-heals. A third slot in the same file, `cachedSecretCodec`, kept the original memoize-the-rejection shape, so a codec failure still poisons the isolate for its lifetime even though the app around it now retries. Route all three through `IsolatePromiseCache`, which memoizes one promise per key, hands the in-flight promise to concurrent callers, and clears the slot on rejection. Net effect: the third slot is fixed and the invariant lives in one place instead of being restated per caller. Unit-test the cache directly. The worker-level regression test cannot reach the guard that lets only the owning entry clear the slot, since that path needs an older promise to reject after a newer key replaced it; dropping the guard leaves `cloudflare.test.ts` green. --- src/server/cloudflare.test.ts | 4 + src/server/cloudflare.ts | 49 ++----- .../cloudflare/isolate-promise-cache.test.ts | 124 ++++++++++++++++++ .../cloudflare/isolate-promise-cache.ts | 39 ++++++ 4 files changed, 180 insertions(+), 36 deletions(-) create mode 100644 src/server/cloudflare/isolate-promise-cache.test.ts create mode 100644 src/server/cloudflare/isolate-promise-cache.ts diff --git a/src/server/cloudflare.test.ts b/src/server/cloudflare.test.ts index 0e34b84b..9e1ace7e 100644 --- a/src/server/cloudflare.test.ts +++ b/src/server/cloudflare.test.ts @@ -26,6 +26,10 @@ describe("cloudflare worker", () => { }); it("deduplicates concurrent app creation and retries after a transient catalog failure", async () => { + // The app and catalog caches live in module state, so this test needs an isolate of its own: + // the shared `worker` above has already cached a healthy catalog. The reimported copy also + // gets a fresh `guarded-fetch` module, which `vitest.setup.ts` no longer holds off real DNS — + // keep this instance on local routes that perform no provider egress. vi.resetModules(); const { default: isolatedWorker } = await import("./cloudflare.ts"); const fallback = memoryAssets(chunkedCatalog()); diff --git a/src/server/cloudflare.ts b/src/server/cloudflare.ts index 6897138c..8a4e899b 100644 --- a/src/server/cloudflare.ts +++ b/src/server/cloudflare.ts @@ -12,6 +12,7 @@ import { executorModules } from "../providers/registry.cloudflare.generated.ts"; import { isConsoleShellPath } from "./api/console-paths.ts"; import { loadCatalogFromAssets } from "./cloudflare/catalog-assets.ts"; import { readPositiveInteger, resolvePublicOrigin } from "./cloudflare/cloudflare-env.ts"; +import { IsolatePromiseCache } from "./cloudflare/isolate-promise-cache.ts"; import { createConnectApp } from "./connect-app.ts"; import { KVTransitFileService } from "./files/kv-transit-files.ts"; import { R2TransitFileService } from "./files/r2-transit-files.ts"; @@ -24,28 +25,15 @@ interface CloudflareExecutionContext { passThroughOnException(): void; } -let catalogPromise: Promise | undefined; -let cachedSecretCodec: { key: string; codec: Promise } | undefined; -let cachedApp: { key: string; app: Promise } | undefined; +const catalogCache = new IsolatePromiseCache(); +const secretCodecCache = new IsolatePromiseCache(); +const appCache = new IsolatePromiseCache(); export default { async fetch(request: Request, env: CloudflareEnv, _ctx: CloudflareExecutionContext): Promise { setPrivateNetworkAccessAllowed(parsePrivateNetworkAccessFlag(env.OOMOL_CONNECT_ALLOW_PRIVATE_NETWORK)); const publicOrigin = resolvePublicOrigin(request, env); - const cacheKey = createCacheKey(env, publicOrigin); - let appPromise = cachedApp?.key === cacheKey ? cachedApp.app : undefined; - if (!appPromise) { - const createdApp = createCloudflareApp(env, publicOrigin); - cachedApp = { key: cacheKey, app: createdApp }; - void createdApp.catch(() => { - if (cachedApp?.app === createdApp) { - cachedApp = undefined; - } - }); - appPromise = createdApp; - } - - const { app } = await appPromise; + const { app } = await appCache.get(createCacheKey(env, publicOrigin), () => createCloudflareApp(env, publicOrigin)); const response = await app.fetch(request, env); if (response.status === 404 && env.ASSETS && shouldServeAsset(request)) { return env.ASSETS.fetch(request); @@ -127,28 +115,17 @@ function writeWorkerLog(level: "error" | "info" | "warn"): (fields: unknown, mes } function loadCatalogOnce(assets: AssetsBinding): Promise { - if (catalogPromise) { - return catalogPromise; - } - - const createdCatalog = loadCatalogFromAssets(assets, { - executableServices: Object.keys(executorModules), - }); - catalogPromise = createdCatalog; - void createdCatalog.catch(() => { - if (catalogPromise === createdCatalog) { - catalogPromise = undefined; - } - }); - return createdCatalog; + // The catalog depends only on the assets binding, which is fixed for the isolate, so one slot + // under a constant key covers every request. + return catalogCache.get("", () => + loadCatalogFromAssets(assets, { + executableServices: Object.keys(executorModules), + }), + ); } function createSecretCodec(encryptionKey: string | undefined): Promise { - const key = encryptionKey ?? ""; - if (!cachedSecretCodec || cachedSecretCodec.key !== key) { - cachedSecretCodec = { key, codec: createWorkerSecretCodec(encryptionKey) }; - } - return cachedSecretCodec.codec; + return secretCodecCache.get(encryptionKey ?? "", () => createWorkerSecretCodec(encryptionKey)); } function createCacheKey(env: CloudflareEnv, publicOrigin: string): string { diff --git a/src/server/cloudflare/isolate-promise-cache.test.ts b/src/server/cloudflare/isolate-promise-cache.test.ts new file mode 100644 index 00000000..c57abf76 --- /dev/null +++ b/src/server/cloudflare/isolate-promise-cache.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from "vitest"; +import { IsolatePromiseCache } from "./isolate-promise-cache.ts"; + +describe("IsolatePromiseCache", () => { + it("creates once per key and reuses the resolved promise", async () => { + const cache = new IsolatePromiseCache(); + let creations = 0; + const create = async (): Promise => { + creations++; + return "value"; + }; + + await expect(cache.get("k", create)).resolves.toBe("value"); + await expect(cache.get("k", create)).resolves.toBe("value"); + expect(creations).toBe(1); + }); + + it("shares one in-flight promise between concurrent callers", async () => { + const cache = new IsolatePromiseCache(); + let creations = 0; + let release = (): void => {}; + const create = (): Promise => { + creations++; + return new Promise((resolve) => { + release = () => resolve("value"); + }); + }; + + const first = cache.get("k", create); + const second = cache.get("k", create); + expect(first).toBe(second); + release(); + await expect(Promise.all([first, second])).resolves.toEqual(["value", "value"]); + expect(creations).toBe(1); + }); + + it("evicts a rejected promise so the next caller retries", async () => { + const cache = new IsolatePromiseCache(); + let attempts = 0; + const create = async (): Promise => { + attempts++; + if (attempts === 1) { + throw new Error("transient"); + } + return "recovered"; + }; + + await expect(cache.get("k", create)).rejects.toThrow("transient"); + await expect(cache.get("k", create)).resolves.toBe("recovered"); + expect(attempts).toBe(2); + }); + + it("keeps rejecting concurrent callers of the failed attempt, then retries once", async () => { + const cache = new IsolatePromiseCache(); + let attempts = 0; + const create = async (): Promise => { + attempts++; + if (attempts === 1) { + throw new Error("transient"); + } + return "recovered"; + }; + + const settled = await Promise.allSettled([cache.get("k", create), cache.get("k", create)]); + expect(settled.map((result) => result.status)).toEqual(["rejected", "rejected"]); + expect(attempts).toBe(1); + await expect(cache.get("k", create)).resolves.toBe("recovered"); + expect(attempts).toBe(2); + }); + + it("replaces the slot when the key changes", async () => { + const cache = new IsolatePromiseCache(); + const create = (value: string) => async (): Promise => value; + + await expect(cache.get("a", create("first"))).resolves.toBe("first"); + await expect(cache.get("b", create("second"))).resolves.toBe("second"); + await expect(cache.get("a", create("third"))).resolves.toBe("third"); + }); + + it("does not evict a newer key when an older promise rejects afterwards", async () => { + const cache = new IsolatePromiseCache(); + let failOld = (): void => {}; + const oldEntry = cache.get( + "old", + () => + new Promise((_resolve, reject) => { + failOld = () => reject(new Error("stale")); + }), + ); + let newCreations = 0; + const createNew = async (): Promise => { + newCreations++; + return "fresh"; + }; + + await expect(cache.get("new", createNew)).resolves.toBe("fresh"); + failOld(); + await expect(oldEntry).rejects.toThrow("stale"); + + // The rejection belongs to the replaced entry, so the live "new" slot must still be memoized. + await expect(cache.get("new", createNew)).resolves.toBe("fresh"); + expect(newCreations).toBe(1); + }); + + it("does not report an unhandled rejection when nobody awaits the evicted promise", async () => { + const cache = new IsolatePromiseCache(); + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown): void => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandled); + try { + const rejected = cache.get("k", async () => { + throw new Error("transient"); + }); + await expect(rejected).rejects.toThrow("transient"); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } + }); +}); diff --git a/src/server/cloudflare/isolate-promise-cache.ts b/src/server/cloudflare/isolate-promise-cache.ts new file mode 100644 index 00000000..f10ab9ac --- /dev/null +++ b/src/server/cloudflare/isolate-promise-cache.ts @@ -0,0 +1,39 @@ +/** + * A one-slot promise cache scoped to a Workers isolate. + * + * An isolate is reused across many requests, so boot work is memoized in module state. Memoizing a + * *rejected* promise turns one transient failure into sustained errors until the isolate is + * recycled, with no self-healing in between. Each slot therefore drops itself as soon as its + * promise rejects, so the next request retries, while concurrent callers still share the in-flight + * promise and do the work once. + */ +export class IsolatePromiseCache { + private entry: IsolateCacheEntry | undefined; + + /** + * Return the cached promise for `key`, creating and memoizing it when the slot holds another key + * or is empty. `create` runs synchronously on a miss, so a burst of concurrent callers arriving + * before the first one settles all share the same promise. + */ + get(key: string, create: () => Promise): Promise { + if (this.entry?.key === key) { + return this.entry.value; + } + + const entry: IsolateCacheEntry = { key, value: create() }; + this.entry = entry; + // Evict only while this entry still owns the slot: a different key may have replaced it while + // the promise was in flight, and that newer entry must survive. + void entry.value.catch(() => { + if (this.entry === entry) { + this.entry = undefined; + } + }); + return entry.value; + } +} + +interface IsolateCacheEntry { + key: string; + value: Promise; +}