diff --git a/server/adapters/cache/runtime-cache.test.ts b/server/adapters/cache/runtime-cache.test.ts index 2fac9861..00889641 100644 --- a/server/adapters/cache/runtime-cache.test.ts +++ b/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((resolve) => (release = resolve))) + it('does not reuse a request-bound loader that remains pending', async () => { + const loader = vi + .fn() + .mockImplementationOnce(() => new Promise(() => 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 () => { diff --git a/server/adapters/cache/runtime-cache.ts b/server/adapters/cache/runtime-cache.ts index 7a70d33d..245151a7 100644 --- a/server/adapters/cache/runtime-cache.ts +++ b/server/adapters/cache/runtime-cache.ts @@ -44,7 +44,6 @@ export function createRuntimeCache(options: RuntimeCacheOptions): CacheService { } const stores = new Map>() - const loads = new Map>>() const now = options.now ?? Date.now function memoryStore(namespace: string): Map { @@ -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> | 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> => { - 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>) - 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(policy: CachePolicy, key: string, value: T): Promise { diff --git a/server/usecases/ports/cache.ts b/server/usecases/ports/cache.ts index 6819da44..c4d696ab 100644 --- a/server/usecases/ports/cache.ts +++ b/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 { namespace: string diff --git a/vitest.config.ts b/vitest.config.ts index e734447f..f5b94f16 100644 --- a/vitest.config.ts +++ b/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', ] diff --git a/workers/bootstrap-init.cf-test.ts b/workers/bootstrap-init.cf-test.ts new file mode 100644 index 00000000..7e09497b --- /dev/null +++ b/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[0] + const env = { BETTER_AUTH_SECRET: 'test-secret' } as Parameters[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() + } + }) +}) diff --git a/workers/bootstrap.ts b/workers/bootstrap.ts index 4480ccb1..0207b798 100644 --- a/workers/bootstrap.ts +++ b/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 @@ -47,6 +48,19 @@ interface WorkerRuntime { appInitBySlot: Map>> } +type InitializeApp = ( + runtime: WorkerRuntime, + env: Env, + baseURL: string, + trustedOrigins: string[], +) => Promise<{ auth: Auth; app: ReturnType }> +type KeepAlive = (promise: Promise) => 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> { 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(operation: Promise, timeoutMs: number, message: string): Promise { + let timeoutId: ReturnType | undefined + const timeout = new Promise((_resolve, reject) => { + timeoutId = setTimeout(() => reject(new Error(message)), timeoutMs) + }) + try { + return await Promise.race([operation, timeout]) + } finally { + if (timeoutId !== undefined) clearTimeout(timeoutId) } }