fix(cloudflare): retry catalog load after failure (#255)

## Summary

- clear rejected Cloudflare app and catalog promises so a recovered
asset binding is retried
- evict only the promise that is still cached, preserving concurrent
request deduplication
- cover concurrent failure sharing and a successful retry through
`worker.fetch`

Fixes #246

## Validation

- `npx vitest run src/server/cloudflare.test.ts` (5 tests)
- `npm run fix-check`
- `npm test` (66 files, 704 tests)
- `npm run lint`
- `npm run format`
- `git diff --check`

---------

Co-authored-by: cyphercodes <cyphercodes@users.noreply.github.com>
Co-authored-by: Kevin Cui <bh@bugs.cc>

Author: cyphercodes

src/server/cloudflare.test.ts

@@ -25,6 +25,49 @@ describe("cloudflare worker", () => {
     vi.restoreAllMocks();
   });
 
+  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());
+    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(

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,20 +25,15 @@ interface CloudflareExecutionContext {
   passThroughOnException(): void;
 }
 
-let catalogPromise: Promise<CatalogStore> | undefined;
-let cachedSecretCodec: { key: string; codec: Promise<ISecretCodec> } | undefined;
-let cachedApp: { key: string; app: Promise<ConnectApp> } | undefined;
+const catalogCache = new IsolatePromiseCache<CatalogStore>();
+const secretCodecCache = new IsolatePromiseCache<ISecretCodec>();
+const appCache = new IsolatePromiseCache<ConnectApp>();
 
 export default {
   async fetch(request: Request, env: CloudflareEnv, _ctx: CloudflareExecutionContext): Promise<Response> {
     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) };
-    }
-
-    const { app } = await cachedApp.app;
+    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);
@@ -119,18 +115,17 @@ function writeWorkerLog(level: "error" | "info" | "warn"): (fields: unknown, mes
 }
 
 function loadCatalogOnce(assets: AssetsBinding): Promise<CatalogStore> {
-  catalogPromise ??= loadCatalogFromAssets(assets, {
-    executableServices: Object.keys(executorModules),
-  });
-  return catalogPromise;
+  // 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<ISecretCodec> {
-  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 {

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<string>();
+    let creations = 0;
+    const create = async (): Promise<string> => {
+      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<string>();
+    let creations = 0;
+    let release = (): void => {};
+    const create = (): Promise<string> => {
+      creations++;
+      return new Promise<string>((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<string>();
+    let attempts = 0;
+    const create = async (): Promise<string> => {
+      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<string>();
+    let attempts = 0;
+    const create = async (): Promise<string> => {
+      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<string>();
+    const create = (value: string) => async (): Promise<string> => 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<string>();
+    let failOld = (): void => {};
+    const oldEntry = cache.get(
+      "old",
+      () =>
+        new Promise<string>((_resolve, reject) => {
+          failOld = () => reject(new Error("stale"));
+        }),
+    );
+    let newCreations = 0;
+    const createNew = async (): Promise<string> => {
+      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<string>();
+    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);
+    }
+  });
+});

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<T> {
+  private entry: IsolateCacheEntry<T> | 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<T>): Promise<T> {
+    if (this.entry?.key === key) {
+      return this.entry.value;
+    }
+
+    const entry: IsolateCacheEntry<T> = { 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<T> {
+  key: string;
+  value: Promise<T>;
+}