fix(workers): recover canceled cache initialization

server/adapters/cache/runtime-cache.test.ts

@@ -56,18 +56,18 @@ describe('runtime cache', () => {
     expect(await cache.get(stringPolicy, 'present')).toMatchObject({ value: 'value', tier: 'memory' })
   })
 
-  it('coalesces concurrent loads for the same key', async () => {
-    let release: (value: string) => void = () => undefined
-    const loader = vi.fn(() => new Promise<string>((resolve) => (release = resolve)))
+  it('does not reuse a request-bound loader that remains pending', async () => {
+    const loader = vi
+      .fn()
+      .mockImplementationOnce(() => new Promise<string>(() => undefined))
+      .mockResolvedValue('recovered')
     const cache = createRuntimeCache({ mode: 'memory' })
 
-    const first = cache.getOrLoad(stringPolicy, 'a', loader)
-    const second = cache.getOrLoad(stringPolicy, 'a', loader)
-    expect(loader).toHaveBeenCalledTimes(1)
+    void cache.getOrLoad(stringPolicy, 'a', loader)
+    const recovered = cache.getOrLoad(stringPolicy, 'a', loader)
 
-    release('shared')
-    expect(await first).toMatchObject({ value: 'shared', tier: 'source' })
-    expect(await second).toMatchObject({ value: 'shared', tier: 'coalesced' })
+    expect(loader).toHaveBeenCalledTimes(2)
+    await expect(recovered).resolves.toMatchObject({ value: 'recovered', tier: 'source' })
   })
 
   it('uses the shorter negative-cache TTL', async () => {

server/adapters/cache/runtime-cache.ts

@@ -44,7 +44,6 @@ export function createRuntimeCache(options: RuntimeCacheOptions): CacheService {
   }
 
   const stores = new Map<string, Map<string, MemoryEntry>>()
-  const loads = new Map<string, Promise<CacheResult<unknown>>>()
   const now = options.now ?? Date.now
 
   function memoryStore(namespace: string): Map<string, MemoryEntry> {
@@ -162,32 +161,18 @@ export function createRuntimeCache(options: RuntimeCacheOptions): CacheService {
       const inMemory = memoryGet(policy, key)
       if (inMemory !== undefined) return observed(policy.namespace, 'memory', startedAt, inMemory)
 
-      const loadKey = cacheKey(policy, key)
-      const existing = loads.get(loadKey) as Promise<CacheResult<T>> | undefined
-      if (existing) {
-        const result = await existing
-        return observed(policy.namespace, 'coalesced', startedAt, result.value)
+      if (usesDistributed(policy)) {
+        const distributed = await distributedGet(policy, key)
+        if (distributed !== undefined) return observed(policy.namespace, 'distributed', startedAt, distributed)
       }
 
-      const load = (async (): Promise<CacheResult<T>> => {
-        if (usesDistributed(policy)) {
-          const distributed = await distributedGet(policy, key)
-          if (distributed !== undefined) return { value: distributed, tier: 'distributed' }
-        }
-
-        const value = await loader()
-        const freshUntil = now() + ttlFor(policy, value)
-        memoryPut(policy, key, value, freshUntil)
-        if (usesDistributed(policy)) await distributedPut(policy, key, value, freshUntil)
-        return { value, tier: 'source' }
-      })()
-      loads.set(loadKey, load as Promise<CacheResult<unknown>>)
-      try {
-        const result = await load
-        return observed(policy.namespace, result.tier, startedAt, result.value)
-      } finally {
-        if (loads.get(loadKey) === load) loads.delete(loadKey)
-      }
+      // Loaders may perform request-bound I/O. Do not share their pending
+      // promises across Worker requests; only the resolved value is cacheable.
+      const value = await loader()
+      const freshUntil = now() + ttlFor(policy, value)
+      memoryPut(policy, key, value, freshUntil)
+      if (usesDistributed(policy)) await distributedPut(policy, key, value, freshUntil)
+      return observed(policy.namespace, 'source', startedAt, value)
     },
 
     async replace<T>(policy: CachePolicy<T>, key: string, value: T): Promise<void> {

server/usecases/ports/cache.ts

@@ -1,6 +1,6 @@
 export type CacheMode = 'off' | 'memory' | 'distributed'
 
-export type CacheTier = 'bypass' | 'memory' | 'distributed' | 'source' | 'coalesced'
+export type CacheTier = 'bypass' | 'memory' | 'distributed' | 'source'
 
 export interface CachePolicy<T> {
   namespace: string

vitest.config.ts

@@ -76,6 +76,7 @@ const isolatedCloudflareTests = [
   'server/http/site/storages.cf-test.ts',
   'server/http/site/system.cf-test.ts',
   'workers/bootstrap-image-domain.cf-test.ts',
+  'workers/bootstrap-init.cf-test.ts',
   'workers/bootstrap.cf-test.ts',
 ]
 

workers/bootstrap-init.cf-test.ts

@@ -0,0 +1,35 @@
+import { describe, expect, it, vi } from 'vitest'
+import { APP_INITIALIZATION_TIMEOUT_MS, appForRequest } from './bootstrap'
+
+describe('[CF] Worker auth initialization recovery', () => {
+  it('starts a fresh initialization after an earlier request reaches its deadline', async () => {
+    vi.useFakeTimers()
+    try {
+      const initialize = vi
+        .fn()
+        .mockImplementationOnce(() => new Promise(() => undefined))
+        .mockResolvedValue({ auth: {}, app: { fetch: vi.fn() } })
+      const keepAlive = vi.fn()
+      const runtime = {
+        authBySlot: new Map(),
+        appBySlot: new Map(),
+        appInitBySlot: new Map(),
+      } as Parameters<typeof appForRequest>[0]
+      const env = { BETTER_AUTH_SECRET: 'test-secret' } as Parameters<typeof appForRequest>[2]
+      const request = new Request('https://recovery.example.com/api/health')
+
+      const stuck = appForRequest(runtime, request, env, initialize, keepAlive)
+      const timedOut = expect(stuck).rejects.toThrow('Worker app initialization timed out')
+      expect(keepAlive).toHaveBeenCalledTimes(1)
+      await vi.advanceTimersByTimeAsync(APP_INITIALIZATION_TIMEOUT_MS)
+      await timedOut
+
+      const recovered = appForRequest(runtime, request, env, initialize, keepAlive)
+
+      expect(initialize).toHaveBeenCalledTimes(2)
+      await expect(recovered).resolves.toEqual({ fetch: expect.any(Function) })
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+})

workers/bootstrap.ts

@@ -37,6 +37,7 @@ const SHARE_TOKEN_RE = /^\/s\/([^/?#]+)/
 // allowing arbitrary Host headers to grow the cache. Changes to OAuth provider
 // configs or env vars take effect on isolate recycle.
 type AuthSlot = 'configured' | 'primary' | 'webdav'
+export const APP_INITIALIZATION_TIMEOUT_MS = 10_000
 
 interface WorkerRuntime {
   platform: ReturnType<typeof createCloudflarePlatform>
@@ -47,6 +48,19 @@ interface WorkerRuntime {
   appInitBySlot: Map<AuthSlot, Promise<ReturnType<typeof createApp>>>
 }
 
+type InitializeApp = (
+  runtime: WorkerRuntime,
+  env: Env,
+  baseURL: string,
+  trustedOrigins: string[],
+) => Promise<{ auth: Auth; app: ReturnType<typeof createApp> }>
+type KeepAlive = (promise: Promise<unknown>) => void
+
+const initializeApp: InitializeApp = async (runtime, env, baseURL, trustedOrigins) => {
+  const auth = await createAuth(runtime.platform, env.BETTER_AUTH_SECRET, baseURL, trustedOrigins, waitUntil)
+  return { auth, app: createApp(runtime.platform, auth, runtime.deps) }
+}
+
 let cachedRuntime: WorkerRuntime | undefined
 const responseCache = (
   caches as unknown as {
@@ -77,10 +91,12 @@ function runtimeFor(env: Env): WorkerRuntime {
   return cachedRuntime
 }
 
-async function appForRequest(
+export async function appForRequest(
   runtime: WorkerRuntime,
   request: Request,
   env: Env,
+  initialize: InitializeApp = initializeApp,
+  keepAlive: KeepAlive = waitUntil,
 ): Promise<ReturnType<typeof createApp>> {
   const origin = new URL(request.url).origin
   const webDavRequest = isPotentialWebDavPublicRequest(request.url)
@@ -98,19 +114,35 @@ async function appForRequest(
   const pendingApp = runtime.appInitBySlot.get(slot)
   if (pendingApp) return pendingApp
 
-  const appPromise = createAuth(runtime.platform, env.BETTER_AUTH_SECRET, baseURL, trustedOrigins, waitUntil).then(
-    (auth) => {
-      const app = createApp(runtime.platform, auth, runtime.deps)
-      runtime.authBySlot.set(slot, auth)
-      runtime.appBySlot.set(slot, app)
-      return app
-    },
-  )
+  // Keep the single cold-start initialization alive if its creating request is
+  // canceled, and bound it so a genuinely stuck dependency cannot poison the slot.
+  const appPromise = withTimeout(
+    initialize(runtime, env, baseURL, trustedOrigins),
+    APP_INITIALIZATION_TIMEOUT_MS,
+    'Worker app initialization timed out',
+  ).then(({ auth, app }) => {
+    runtime.authBySlot.set(slot, auth)
+    runtime.appBySlot.set(slot, app)
+    return app
+  })
   runtime.appInitBySlot.set(slot, appPromise)
+  keepAlive(appPromise.catch(() => undefined))
   try {
     return await appPromise
   } finally {
-    runtime.appInitBySlot.delete(slot)
+    if (runtime.appInitBySlot.get(slot) === appPromise) runtime.appInitBySlot.delete(slot)
+  }
+}
+
+async function withTimeout<T>(operation: Promise<T>, timeoutMs: number, message: string): Promise<T> {
+  let timeoutId: ReturnType<typeof setTimeout> | undefined
+  const timeout = new Promise<never>((_resolve, reject) => {
+    timeoutId = setTimeout(() => reject(new Error(message)), timeoutMs)
+  })
+  try {
+    return await Promise.race([operation, timeout])
+  } finally {
+    if (timeoutId !== undefined) clearTimeout(timeoutId)
   }
 }