diff --git a/apps/api/src/read/candidates.ts b/apps/api/src/read/candidates.ts index 75d5d22..29435c1 100644 --- a/apps/api/src/read/candidates.ts +++ b/apps/api/src/read/candidates.ts @@ -18,7 +18,13 @@ candidates.get("/v2/collections/candidates", (c) => cached( c, "collection-candidates:chosen:2", - { ttl: 86400, edge: 300, swr: 86400, staleKey: "collection-candidates:chosen:1" }, + { + ttl: 86400, + edge: 300, + swr: 86400, + staleKey: "collection-candidates:chosen:1", + durableWrite: true, + }, async (): Promise> => { const seeds = await collectorCandidateSeeds(c.env.CORE_DB, SEED_LIMIT); const chosen = new Map(); diff --git a/apps/api/src/read/respond.ts b/apps/api/src/read/respond.ts index 4afe4a1..6a12b77 100644 --- a/apps/api/src/read/respond.ts +++ b/apps/api/src/read/respond.ts @@ -21,6 +21,7 @@ export const J = (c: Ctx, body: unknown, ttl = 30) => }); const CACHE_REFRESH_LEASE_SECONDS = 60; +const CACHE_WRITE_RETRY_DELAYS_MS = [250, 750, 2_000, 4_000] as const; function primaryCacheDatabase(db: D1Database): D1Database { const session = db as unknown as { withSession?: (mode: string) => D1Database }; @@ -36,6 +37,29 @@ export async function claimCacheRefresh(db: D1Database, key: string, now: number return (result.meta.rows_written ?? 0) === 1; } +const wait = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds)); + +/** + * An expensive producer can briefly saturate D1, causing the immediately following cache INSERT to + * fail. Retrying only that explicit overload prevents a successful result from becoming an endless + * cold-cache recomputation loop; schema/auth/logic errors still fail immediately. + */ +export async function retryD1CacheWrite( + write: () => Promise, + pause: (milliseconds: number) => Promise = wait, +): Promise { + for (let attempt = 0; ; attempt++) { + try { + await write(); + return; + } catch (error) { + const overloaded = error instanceof Error && error.message.includes("D1 DB is overloaded"); + if (!overloaded || attempt >= CACHE_WRITE_RETRY_DELAYS_MS.length) throw error; + await pause(CACHE_WRITE_RETRY_DELAYS_MS[attempt]!); + } + } +} + /** * D1-backed response cache (the `cache` table: key, body, ctype, expires_at) with stale-while-revalidate. * Layer 2 above the per-colo edge cache: it PERSISTS across colos and cold edge, so a heavy aggregation @@ -55,10 +79,12 @@ export async function cached( * stale body serves immediately and the primary computes in the background, so a version * bump never opens a cold window for readers. */ staleKey?: string; + /** Require a successful cache publication before returning an expensive cold result. */ + durableWrite?: boolean; }, producer: () => Promise, ): Promise { - const { ttl, edge = Math.min(ttl, 30), swr = ttl, staleKey } = opts; + const { ttl, edge = Math.min(ttl, 30), swr = ttl, staleKey, durableWrite = false } = opts; // Public query producers keep using the replica-local database from the route context. Cache rows // and their refresh lease use the primary so every colo observes one global coordination point. const cacheDb = primaryCacheDatabase(c.env.CORE_DB); @@ -71,15 +97,18 @@ export async function cached( }); const write = async (): Promise => { const body = JSON.stringify(await producer()); - await cacheDb - .prepare( - `INSERT INTO cache (key,body,ctype,expires_at,refreshing_until) VALUES (?,?,'application/json',?,0) - ON CONFLICT(key) DO UPDATE SET body=excluded.body, ctype=excluded.ctype, - expires_at=excluded.expires_at,refreshing_until=0`, - ) - .bind(key, body, Math.floor(Date.now() / 1000) + ttl) - .run() - .catch(() => {}); + const persist = async () => { + await cacheDb + .prepare( + `INSERT INTO cache (key,body,ctype,expires_at,refreshing_until) VALUES (?,?,'application/json',?,0) + ON CONFLICT(key) DO UPDATE SET body=excluded.body, ctype=excluded.ctype, + expires_at=excluded.expires_at,refreshing_until=0`, + ) + .bind(key, body, Math.floor(Date.now() / 1000) + ttl) + .run(); + }; + if (durableWrite) await retryD1CacheWrite(persist); + else await persist().catch(() => {}); return body; }; const hit = await cacheDb diff --git a/apps/api/tests/response-cache.test.ts b/apps/api/tests/response-cache.test.ts index adecb2c..bdf1cc2 100644 --- a/apps/api/tests/response-cache.test.ts +++ b/apps/api/tests/response-cache.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { DatabaseSync } from "node:sqlite"; import { test } from "node:test"; -import { cached, claimCacheRefresh, type Ctx } from "#api/read/respond"; +import { cached, claimCacheRefresh, retryD1CacheWrite, type Ctx } from "#api/read/respond"; class Statement { private args: unknown[] = []; @@ -52,6 +52,35 @@ test("only one stale-cache request owns a refresh lease", async () => { sqlite.close(); }); +test("durable cache publication retries only transient D1 overload", async () => { + let attempts = 0; + const delays: number[] = []; + await retryD1CacheWrite( + async () => { + attempts++; + if (attempts < 3) throw new Error("D1_ERROR: D1 DB is overloaded. Requests queued for too long."); + }, + async (milliseconds) => { + delays.push(milliseconds); + }, + ); + assert.equal(attempts, 3); + assert.deepEqual(delays, [250, 750]); + + attempts = 0; + await assert.rejects( + retryD1CacheWrite( + async () => { + attempts++; + throw new Error("D1_ERROR: no such table: cache"); + }, + async () => {}, + ), + /no such table/, + ); + assert.equal(attempts, 1); +}); + test("a fresh hit is edge-cacheable for the configured window", async () => { const future = Math.floor(Date.now() / 1000) + 1_000; const { sqlite, db } = cacheDatabase([["k:v2", `{"n":2}`, future]]);