From bd68d17a297f04c1458bef903879a8758cc5121e Mon Sep 17 00:00:00 2001 From: James Date: Tue, 16 Jun 2026 14:50:39 +0100 Subject: [PATCH] fix: order regional cache writes --- .changeset/tidy-caches-wait.md | 7 + .../incremental-cache/regional-cache.spec.ts | 629 ++++++++++++++++++ .../incremental-cache/regional-cache.ts | 237 +++++-- 3 files changed, 822 insertions(+), 51 deletions(-) create mode 100644 .changeset/tidy-caches-wait.md create mode 100644 packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.spec.ts diff --git a/.changeset/tidy-caches-wait.md b/.changeset/tidy-caches-wait.md new file mode 100644 index 000000000..3e351281c --- /dev/null +++ b/.changeset/tidy-caches-wait.md @@ -0,0 +1,7 @@ +--- +"@opennextjs/cloudflare": patch +--- + +fix: order regional cache writes within an instance + +Prevent an older lazy backing-store refresh from replacing a newer regional cache generation written by the same Worker instance, and avoid extending unchanged entries. diff --git a/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.spec.ts b/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.spec.ts new file mode 100644 index 000000000..66110021d --- /dev/null +++ b/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.spec.ts @@ -0,0 +1,629 @@ +import type { CacheValue, IncrementalCache } from "@opennextjs/aws/types/overrides.js"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { withRegionalCache } from "./regional-cache.js"; + +const waitUntilPromises: Promise[] = []; + +vi.mock("../../cloudflare-context.js", () => ({ + getCloudflareContext: () => ({ + ctx: { + waitUntil: (promise: Promise) => { + waitUntilPromises.push(promise); + }, + }, + }), +})); + +const createValue = (body: string): CacheValue<"cache"> => + ({ + type: "app", + body, + revalidate: 1_800, + meta: {}, + }) as CacheValue<"cache">; + +const createCache = () => { + const entries = new Map(); + const cache = { + match: vi.fn(async (key: string) => entries.get(key)?.clone()), + put: vi.fn(async (key: string, response: Response) => { + entries.set(key, response.clone()); + }), + delete: vi.fn(async (key: string) => entries.delete(key)), + }; + + return { cache, entries }; +}; + +const readEntry = async (entries: Map) => { + const response = entries.values().next().value as Response | undefined; + return response + ? ((await response.clone().json()) as { lastModified: number; value: { body: string } }) + : undefined; +}; + +const expectStateReleased = (regionalCache: IncrementalCache) => { + const internals = regionalCache as unknown as { + cacheStates: Map; + pendingCacheOperations: Map; + }; + expect(internals.cacheStates.size).toBe(0); + expect(internals.pendingCacheOperations.size).toBe(0); +}; + +describe("RegionalCache", () => { + beforeEach(() => { + waitUntilPromises.length = 0; + globalThis.nextVersion = "16.2.6"; + }); + + afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it("does not replace a newer entry with an older lazy refresh in the same instance", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValueOnce(100).mockReturnValueOnce(200); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + expect(store.get).toHaveBeenCalledOnce(); + + await regionalCache.set("route", createValue("new")); + resolveStoreGet?.({ value: createValue("old"), lastModified: 100 }); + await Promise.all(waitUntilPromises); + + const entry = await readEntry(entries); + expect(entry?.lastModified).toBe(200); + expect(entry?.value.body).toBe("new"); + expect(cache.put).toHaveBeenCalledTimes(2); + expectStateReleased(regionalCache); + }); + + it("does not extend an unchanged entry lifetime", async () => { + const { cache } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn().mockResolvedValue({ value: createValue("same"), lastModified: 100 }), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("same")); + await regionalCache.get("route"); + await Promise.all(waitUntilPromises); + + expect(cache.put).toHaveBeenCalledTimes(1); + expectStateReleased(regionalCache); + }); + + it("writes a newer backing-store generation", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn().mockResolvedValue({ value: createValue("new"), lastModified: 200 }), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + await Promise.all(waitUntilPromises); + + const entry = await readEntry(entries); + expect(entry?.lastModified).toBe(200); + expect(entry?.value.body).toBe("new"); + expect(cache.put).toHaveBeenCalledTimes(2); + expectStateReleased(regionalCache); + }); + + it("keeps the later explicit set when timestamps are equal", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn(), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("first")); + await regionalCache.set("route", createValue("second")); + + const entry = await readEntry(entries); + expect(entry?.lastModified).toBe(100); + expect(entry?.value.body).toBe("second"); + expect(cache.put).toHaveBeenCalledTimes(2); + expectStateReleased(regionalCache); + }); + + it("does not restore a lazy refresh after deletion", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + await regionalCache.delete("route"); + + resolveStoreGet?.({ value: createValue("newer-store"), lastModified: 200 }); + await Promise.all(waitUntilPromises); + + expect(await readEntry(entries)).toBeUndefined(); + expect(cache.put).toHaveBeenCalledTimes(1); + expect(cache.delete).toHaveBeenCalledOnce(); + expectStateReleased(regionalCache); + }); + + it("does not populate a cache miss after a concurrent set", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + vi.spyOn(Date, "now").mockReturnValue(200); + const cacheMiss = regionalCache.get("route"); + await vi.waitFor(() => expect(store.get).toHaveBeenCalledOnce()); + await regionalCache.set("route", createValue("new")); + resolveStoreGet?.({ value: createValue("old"), lastModified: 100 }); + await cacheMiss; + await Promise.all(waitUntilPromises); + + const entry = await readEntry(entries); + expect(entry?.lastModified).toBe(200); + expect(entry?.value.body).toBe("new"); + expect(cache.put).toHaveBeenCalledTimes(1); + expectStateReleased(regionalCache); + }); + + it("does not restore a lazy refresh after a failed regional set", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValueOnce(100).mockReturnValueOnce(200); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + cache.put.mockRejectedValueOnce(new Error("cache unavailable")); + await regionalCache.set("route", createValue("new")); + resolveStoreGet?.({ value: createValue("store"), lastModified: 150 }); + await Promise.all(waitUntilPromises); + + expect((await readEntry(entries))?.lastModified).toBe(100); + expect(cache.put).toHaveBeenCalledTimes(2); + expectStateReleased(regionalCache); + }); + + it("does not restore a lazy refresh after a failed regional delete", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + cache.delete.mockRejectedValueOnce(new Error("cache unavailable")); + await regionalCache.delete("route"); + resolveStoreGet?.({ value: createValue("store"), lastModified: 200 }); + await Promise.all(waitUntilPromises); + + expect((await readEntry(entries))?.lastModified).toBe(100); + expect(cache.put).toHaveBeenCalledTimes(1); + expectStateReleased(regionalCache); + }); + + it("deletes malformed cache entries", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn(), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + const cacheKey = entries.keys().next().value as string; + entries.set(cacheKey, new Response("invalid-json")); + cache.delete.mockClear(); + + await expect(regionalCache.get("route")).resolves.toBeNull(); + expect(cache.delete).toHaveBeenCalledOnce(); + expectStateReleased(regionalCache); + }); + + it("releases per-key state after operations settle", async () => { + const { cache } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn(), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("value")); + + expectStateReleased(regionalCache); + }); + + it("releases per-key state when malformed cache deletion fails", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn(), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + const cacheKey = entries.keys().next().value as string; + entries.set(cacheKey, new Response("invalid-json")); + cache.delete.mockRejectedValueOnce(new Error("cache unavailable")); + + await expect(regionalCache.get("route")).resolves.toBeNull(); + expect(cache.delete).toHaveBeenCalledOnce(); + expectStateReleased(regionalCache); + }); + + it("handles cache-miss population failures without leaking state", async () => { + const { cache } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + cache.put.mockRejectedValueOnce(new Error("cache unavailable")); + + const store = { + name: "test-store", + get: vi.fn().mockResolvedValue({ value: createValue("value"), lastModified: 100 }), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + await expect(regionalCache.get("route")).resolves.toEqual({ + value: createValue("value"), + lastModified: 100, + }); + await Promise.all(waitUntilPromises); + + expectStateReleased(regionalCache); + }); + + it("keeps set invalidation active while the backing-store write is delayed", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreSet: (() => void) | undefined; + const storeSet = new Promise((resolve) => { + resolveStoreSet = resolve; + }); + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi + .fn() + .mockResolvedValueOnce(undefined) + .mockImplementationOnce(() => storeSet), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValueOnce(100).mockReturnValueOnce(200); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + const pendingSet = regionalCache.set("route", createValue("new")); + resolveStoreGet?.({ value: createValue("store"), lastModified: 150 }); + await Promise.all(waitUntilPromises); + resolveStoreSet?.(); + await pendingSet; + + expect((await readEntry(entries))?.lastModified).toBe(200); + expectStateReleased(regionalCache); + }); + + it("keeps delete invalidation active while the backing-store delete is delayed", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let resolveStoreDelete: (() => void) | undefined; + const storeDelete = new Promise((resolve) => { + resolveStoreDelete = resolve; + }); + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(() => storeDelete), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + const pendingDelete = regionalCache.delete("route"); + resolveStoreGet?.({ value: createValue("store"), lastModified: 200 }); + await Promise.all(waitUntilPromises); + resolveStoreDelete?.(); + await pendingDelete; + + expect(await readEntry(entries)).toBeUndefined(); + expectStateReleased(regionalCache); + }); + + it("does not refresh an old cache response observed before a concurrent set", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let releaseMatch: (() => void) | undefined; + const matchGate = new Promise((resolve) => { + releaseMatch = resolve; + }); + let markMatchCaptured: (() => void) | undefined; + const matchCaptured = new Promise((resolve) => { + markMatchCaptured = resolve; + }); + cache.match.mockImplementationOnce(async (key: string) => { + const response = entries.get(key)?.clone(); + markMatchCaptured?.(); + await matchGate; + return response; + }); + + const store = { + name: "test-store", + get: vi.fn().mockResolvedValue({ value: createValue("store"), lastModified: 150 }), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValueOnce(100).mockReturnValueOnce(200); + await regionalCache.set("route", createValue("old")); + const pendingGet = regionalCache.get("route"); + await matchCaptured; + await regionalCache.set("route", createValue("new")); + releaseMatch?.(); + await pendingGet; + await Promise.all(waitUntilPromises); + + const entry = await readEntry(entries); + expect(entry?.lastModified).toBe(200); + expect(entry?.value.body).toBe("new"); + expectStateReleased(regionalCache); + }); + + it("does not delete a concurrent set after observing an old malformed response", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + const store = { + name: "test-store", + get: vi.fn(), + set: vi.fn(), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { mode: "long-lived" }); + + vi.spyOn(Date, "now").mockReturnValueOnce(100).mockReturnValueOnce(200); + await regionalCache.set("route", createValue("old")); + const cacheKey = entries.keys().next().value as string; + entries.set(cacheKey, new Response("invalid-json")); + + let releaseMatch: (() => void) | undefined; + const matchGate = new Promise((resolve) => { + releaseMatch = resolve; + }); + let markMatchCaptured: (() => void) | undefined; + const matchCaptured = new Promise((resolve) => { + markMatchCaptured = resolve; + }); + cache.match.mockImplementationOnce(async (key: string) => { + const response = entries.get(key)?.clone(); + markMatchCaptured?.(); + await matchGate; + return response; + }); + + const pendingGet = regionalCache.get("route"); + await matchCaptured; + await regionalCache.set("route", createValue("new")); + releaseMatch?.(); + await pendingGet; + + const entry = await readEntry(entries); + expect(entry?.lastModified).toBe(200); + expect(entry?.value.body).toBe("new"); + expectStateReleased(regionalCache); + }); + + it("does not restore a lazy refresh after a backing-store set failure", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let rejectStoreSet: ((error: Error) => void) | undefined; + const storeSet = new Promise((_resolve, reject) => { + rejectStoreSet = reject; + }); + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi + .fn() + .mockResolvedValueOnce(undefined) + .mockImplementationOnce(() => storeSet), + delete: vi.fn(), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + const failedSet = regionalCache.set("route", createValue("new")); + resolveStoreGet?.({ value: createValue("store"), lastModified: 200 }); + await Promise.all(waitUntilPromises); + rejectStoreSet?.(new Error("store unavailable")); + await failedSet; + + expect((await readEntry(entries))?.lastModified).toBe(100); + expect(cache.put).toHaveBeenCalledTimes(1); + expectStateReleased(regionalCache); + }); + + it("does not restore a lazy refresh after a backing-store delete failure", async () => { + const { cache, entries } = createCache(); + vi.stubGlobal("caches", { open: vi.fn().mockResolvedValue(cache) }); + + let rejectStoreDelete: ((error: Error) => void) | undefined; + const storeDelete = new Promise((_resolve, reject) => { + rejectStoreDelete = reject; + }); + let resolveStoreGet: ((entry: Awaited>) => void) | undefined; + const storeGet = new Promise>>((resolve) => { + resolveStoreGet = resolve; + }); + const store = { + name: "test-store", + get: vi.fn(() => storeGet), + set: vi.fn(), + delete: vi.fn(() => storeDelete), + } satisfies IncrementalCache; + const regionalCache = withRegionalCache(store, { + mode: "long-lived", + shouldLazilyUpdateOnCacheHit: true, + }); + + vi.spyOn(Date, "now").mockReturnValue(100); + await regionalCache.set("route", createValue("old")); + await regionalCache.get("route"); + const failedDelete = regionalCache.delete("route"); + resolveStoreGet?.({ value: createValue("store"), lastModified: 200 }); + await Promise.all(waitUntilPromises); + rejectStoreDelete?.(new Error("store unavailable")); + await failedDelete; + + expect((await readEntry(entries))?.lastModified).toBe(100); + expect(cache.put).toHaveBeenCalledTimes(1); + expectStateReleased(regionalCache); + }); +}); diff --git a/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.ts b/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.ts index 2513d0e0d..1c09ba303 100644 --- a/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.ts +++ b/packages/cloudflare/src/api/overrides/incremental-cache/regional-cache.ts @@ -67,6 +67,17 @@ interface PutToCacheInput { key: string; cacheType?: CacheEntryType; entry: IncrementalCacheEntry; + expectedRevision?: number; +} + +interface CacheState { + revision: number; + references: number; +} + +interface CapturedCacheState { + state: CacheState; + revision: number; } /** @@ -85,6 +96,8 @@ class RegionalCache implements IncrementalCache { public name: string; protected localCache: Cache | undefined; + protected pendingCacheOperations = new Map>(); + protected cacheStates = new Map(); constructor( private store: IncrementalCache, @@ -127,44 +140,90 @@ class RegionalCache implements IncrementalCache { try { const cache = await this.getCacheInstance(); const urlKey = this.getCacheUrlKey(key, cacheType); + const capturedState = this.captureCacheState(urlKey); // Check for a cached entry as this will be faster than the store response. - const cachedResponse = await cache.match(urlKey); + let cachedResponse: Response | undefined; + try { + cachedResponse = await cache.match(urlKey); + } catch (e) { + this.releaseCacheState(urlKey, capturedState.state); + throw e; + } if (cachedResponse) { debugCache("RegionalCache", `get ${key} -> cached response`); + let cachedEntry: IncrementalCacheEntry; + try { + cachedEntry = await cachedResponse.json>(); + } catch (e) { + try { + await this.deleteFromCache(urlKey, capturedState.revision); + } finally { + this.releaseCacheState(urlKey, capturedState.state); + } + throw e; + } // Re-fetch from the store and update the regional cache in the background. // Note: this is only useful when the Cache API is not purged automatically. if (this.opts.shouldLazilyUpdateOnCacheHit) { getCloudflareContext().ctx.waitUntil( - this.store.get(key, cacheType).then(async (rawEntry) => { - const { value, lastModified } = rawEntry ?? {}; - - if (value && typeof lastModified === "number") { - await this.putToCache({ key, cacheType, entry: { value, lastModified } }); + (async () => { + try { + const rawEntry = await this.store.get(key, cacheType); + const { value, lastModified } = rawEntry ?? {}; + + if (value && typeof lastModified === "number" && lastModified > cachedEntry.lastModified) { + await this.putToCache({ + key, + cacheType, + entry: { value, lastModified }, + expectedRevision: capturedState.revision, + }); + } + } catch (e) { + error(`Failed to lazily update regional cache for ${key}`, e); + } finally { + this.releaseCacheState(urlKey, capturedState.state); } - }) + })() ); + } else { + this.releaseCacheState(urlKey, capturedState.state); } - const responseJson: Record = await cachedResponse.json(); - return { - ...responseJson, + ...cachedEntry, shouldBypassTagCache: this.opts.bypassTagCacheOnCacheHit, }; } - const rawEntry = await this.store.get(key, cacheType); + let rawEntry: WithLastModified> | null; + try { + rawEntry = await this.store.get(key, cacheType); + } catch (e) { + this.releaseCacheState(urlKey, capturedState.state); + throw e; + } const { value, lastModified } = rawEntry ?? {}; - if (!value || typeof lastModified !== "number") return null; + if (!value || typeof lastModified !== "number") { + this.releaseCacheState(urlKey, capturedState.state); + return null; + } debugCache("RegionalCache", `get ${key} -> put to cache`); // Update the locale cache after retrieving from the store. getCloudflareContext().ctx.waitUntil( - this.putToCache({ key, cacheType, entry: { value, lastModified } }) + this.putToCache({ + key, + cacheType, + entry: { value, lastModified }, + expectedRevision: capturedState.revision, + }) + .catch((e) => error(`Failed to populate regional cache for ${key}`, e)) + .finally(() => this.releaseCacheState(urlKey, capturedState.state)) ); return { value, lastModified }; @@ -182,18 +241,26 @@ class RegionalCache implements IncrementalCache { try { debugCache("RegionalCache", `set ${key}`); - await this.store.set(key, value, cacheType); - - await this.putToCache({ - key, - cacheType, - entry: { - value, - // Note: `Date.now()` returns the time of the last IO rather than the actual time. - // See https://developers.cloudflare.com/workers/reference/security-model/ - lastModified: Date.now(), - }, - }); + const urlKey = this.getCacheUrlKey(key, cacheType); + const capturedState = this.captureCacheState(urlKey); + const cacheState = capturedState.state; + cacheState.revision++; + try { + await this.store.set(key, value, cacheType); + await this.putToCache({ + key, + cacheType, + entry: { + value, + // Note: `Date.now()` returns the time of the last IO rather than the actual time. + // See https://developers.cloudflare.com/workers/reference/security-model/ + lastModified: Date.now(), + }, + }); + } finally { + cacheState.revision++; + this.releaseCacheState(urlKey, cacheState); + } } catch (e) { error(`Failed to set the regional cache`, e); } @@ -201,13 +268,19 @@ class RegionalCache implements IncrementalCache { async delete(key: string): Promise { debugCache("RegionalCache", `delete ${key}`); + const urlKey = this.getCacheUrlKey(key); + const capturedState = this.captureCacheState(urlKey); + const cacheState = capturedState.state; + cacheState.revision++; try { await this.store.delete(key); - const cache = await this.getCacheInstance(); - await cache.delete(this.getCacheUrlKey(key)); + await this.deleteFromCache(urlKey); } catch (e) { error("Failed to delete from regional cache", e); + } finally { + cacheState.revision++; + this.releaseCacheState(urlKey, cacheState); } } @@ -223,31 +296,93 @@ class RegionalCache implements IncrementalCache { return "http://cache.local" + `/${buildId}/${key}`.replace(/\/+/g, "/") + `.${cacheType ?? "cache"}`; } - protected async putToCache({ key, cacheType, entry }: PutToCacheInput): Promise { + protected async putToCache({ key, cacheType, entry, expectedRevision }: PutToCacheInput): Promise { const urlKey = this.getCacheUrlKey(key, cacheType); - const cache = await this.getCacheInstance(); - - const age = - this.opts.mode === "short-lived" - ? ONE_MINUTE_IN_SECONDS - : entry.value.revalidate || this.opts.defaultLongLivedTtlSec || THIRTY_MINUTES_IN_SECONDS; - - // We default to the entry key if no tags are found. - // so that we can also revalidate page router based entry this way. - const tags = getTagsFromCacheEntry(entry) ?? [key]; - await cache.put( - urlKey, - new Response(JSON.stringify(entry), { - headers: new Headers({ - "cache-control": `max-age=${age}`, - ...(tags.length > 0 - ? { - "cache-tag": tags.join(","), - } - : {}), - }), - }) - ); + return this.enqueueCacheOperation(urlKey, async (cacheState) => { + if (expectedRevision !== undefined && cacheState.revision !== expectedRevision) return; + + const cache = await this.getCacheInstance(); + const age = + this.opts.mode === "short-lived" + ? ONE_MINUTE_IN_SECONDS + : entry.value.revalidate || this.opts.defaultLongLivedTtlSec || THIRTY_MINUTES_IN_SECONDS; + + // We default to the entry key if no tags are found. + // so that we can also revalidate page router based entry this way. + const tags = getTagsFromCacheEntry(entry) ?? [key]; + await cache.put( + urlKey, + new Response(JSON.stringify(entry), { + headers: new Headers({ + "cache-control": `max-age=${age}`, + ...(tags.length > 0 + ? { + "cache-tag": tags.join(","), + } + : {}), + }), + }) + ); + cacheState.revision++; + }); + } + + protected async deleteFromCache(urlKey: string, expectedRevision?: number): Promise { + return this.enqueueCacheOperation(urlKey, async (cacheState) => { + if (expectedRevision !== undefined && cacheState.revision !== expectedRevision) return; + const cache = await this.getCacheInstance(); + await cache.delete(urlKey); + cacheState.revision++; + }); + } + + protected getCacheState(urlKey: string): CacheState { + let cacheState = this.cacheStates.get(urlKey); + if (!cacheState) { + cacheState = { revision: 0, references: 0 }; + this.cacheStates.set(urlKey, cacheState); + } + return cacheState; + } + + protected captureCacheState(urlKey: string): CapturedCacheState { + const state = this.getCacheState(urlKey); + state.references++; + return { state, revision: state.revision }; + } + + protected releaseCacheState(urlKey: string, state: CacheState): void { + state.references--; + this.cleanupCacheState(urlKey, state); + } + + protected async enqueueCacheOperation( + urlKey: string, + operation: (cacheState: CacheState) => Promise + ): Promise { + const pendingOperation = + this.pendingCacheOperations.get(urlKey)?.catch(() => undefined) ?? Promise.resolve(); + const cacheOperation = pendingOperation.then(() => operation(this.getCacheState(urlKey))); + this.pendingCacheOperations.set(urlKey, cacheOperation); + try { + await cacheOperation; + } finally { + if (this.pendingCacheOperations.get(urlKey) === cacheOperation) { + this.pendingCacheOperations.delete(urlKey); + } + const cacheState = this.cacheStates.get(urlKey); + if (cacheState) this.cleanupCacheState(urlKey, cacheState); + } + } + + protected cleanupCacheState(urlKey: string, state: CacheState): void { + if ( + this.cacheStates.get(urlKey) === state && + state.references === 0 && + !this.pendingCacheOperations.has(urlKey) + ) { + this.cacheStates.delete(urlKey); + } } }