From 63d8febe9506156263076c3e1341e6dfdcc851f6 Mon Sep 17 00:00:00 2001 From: James Date: Thu, 10 Sep 2026 11:15:33 +0100 Subject: [PATCH] refactor(cache): reduce response store metadata calls --- .../workers-response-store/src/binding.ts | 139 +++++-------- .../workers-response-store/src/metadata-do.ts | 194 ++++++++++++++---- .../workers-response-store/tests/e2e.test.mjs | 7 +- 3 files changed, 204 insertions(+), 136 deletions(-) diff --git a/packages/workers-response-store/src/binding.ts b/packages/workers-response-store/src/binding.ts index 7f706a977..2959119d5 100644 --- a/packages/workers-response-store/src/binding.ts +++ b/packages/workers-response-store/src/binding.ts @@ -102,6 +102,7 @@ type CacheKey = { }; type WriteReservation = CacheKey & { + objectKey: string; revision: number; }; @@ -111,6 +112,7 @@ type StoreResult = { }; type PublicationResult = { + entry: StoredEntry | null; published: boolean; previousObjectKey?: string; }; @@ -207,8 +209,6 @@ const MISS_HEADERS = { }; const BACKGROUND_REVALIDATION_LEASE_MS = 30_000; -const ORPHAN_RETENTION_MS = 60 * 60 * 1000; -const ORPHAN_CLEANUP_LIMIT = 100; const R2_DELETE_BATCH_SIZE = 1_000; const CACHE_PURGE_BATCH_SIZE = 100; const MAX_CACHE_TAG_HEADER_BYTES = 16 * 1024; @@ -339,6 +339,24 @@ export class ResponseStoreBinding extends WorkerEntrypoint< return accepted; } + private objectKeyPrefix(keyHash: string): string { + return ["runtime-cache", this.getVersionId(), keyHash].join("/"); + } + + private async reserveWrite( + metadata: CacheMetadataStub, + keyHash: string, + cacheKey: string, + ): Promise { + const reservation = await metadata.reserveWrite( + keyHash, + cacheKey, + this.objectKeyPrefix(keyHash), + Date.now(), + ); + return { cacheKey, keyHash, ...reservation }; + } + private logCleanupFailure(objectKey: string, error: unknown): void { console.error( JSON.stringify({ @@ -349,10 +367,7 @@ export class ResponseStoreBinding extends WorkerEntrypoint< ); } - private async deletePendingObjects( - metadata: CacheMetadataStub, - objectKeys: string[], - ): Promise { + private async deleteObjects(objectKeys: string[]): Promise { if (!objectKeys.length) { return; } @@ -360,47 +375,12 @@ export class ResponseStoreBinding extends WorkerEntrypoint< for (const batch of batches(objectKeys, R2_DELETE_BATCH_SIZE)) { try { await this.env.CACHE_BODIES.delete(batch); - await metadata.finishPendingObjects(batch); } catch (error) { this.logCleanupFailure(batch.join(","), error); } } } - private async trackPendingObjects( - metadata: CacheMetadataStub, - objectKeys: string[], - createdAt: number, - ): Promise { - try { - await metadata.trackPendingObjects(objectKeys, createdAt); - } catch (bulkError) { - try { - for (const objectKey of objectKeys) { - await metadata.trackPendingObject(objectKey, createdAt); - } - } catch (fallbackError) { - throw new AggregateError([bulkError, fallbackError], "Failed to track pending R2 objects"); - } - } - } - - private async cleanupExpiredPendingObjects(metadata: CacheMetadataStub): Promise { - try { - const objectKeys = await metadata.listExpiredPendingObjects( - Date.now() - ORPHAN_RETENTION_MS, - ORPHAN_CLEANUP_LIMIT, - ); - if (!objectKeys.length) { - return; - } - - await this.deletePendingObjects(metadata, objectKeys); - } catch (error) { - this.logCleanupFailure("expired-pending-objects", error); - } - } - private async readStoredResponse(entry: StoredEntry, now = Date.now()): Promise { const object = await this.env.CACHE_BODIES.get(entry.objectKey); if (!object) { @@ -456,10 +436,10 @@ export class ResponseStoreBinding extends WorkerEntrypoint< revalidator: ResponseStorePutOptions["revalidator"], reservation?: WriteReservation, ): Promise { - const { cacheKey, keyHash } = reservation ?? (await this.deriveCacheKey(request)); - const revision = reservation?.revision ?? (await metadata.beginWrite(keyHash, cacheKey)); - - const objectKey = ["runtime-cache", this.getVersionId(), keyHash, String(revision)].join("/"); + const cacheKey = reservation ?? (await this.deriveCacheKey(request)); + const write = + reservation ?? (await this.reserveWrite(metadata, cacheKey.keyHash, cacheKey.cacheKey)); + const { keyHash, objectKey, revision } = write; const now = Date.now(); const policy = deriveCachePolicy(response.headers, now); @@ -493,51 +473,30 @@ export class ResponseStoreBinding extends WorkerEntrypoint< responseMetadataInR2: true, }; - await this.trackPendingObjects(metadata, [objectKey], now); - - try { - // RPC-transferred Response streams do not retain the fixed-length marker - // required by R2's single-part put API. Materialise only in the cache - // Worker; bodies are never stored in the metadata Durable Object. - const body = response.body ? await response.arrayBuffer() : new ArrayBuffer(0); - await this.env.CACHE_BODIES.put(objectKey, body, { - customMetadata: { - status: String(response.status), - createdAt: String(policy.createdAt), - initialAge: String(policy.initialAge), - }, - }); - } catch (error) { - await this.deletePendingObjects(metadata, [objectKey]); - throw error; - } + // RPC-transferred Response streams do not retain the fixed-length marker + // required by R2's single-part put API. Materialise only in the cache + // Worker; bodies are never stored in the metadata Durable Object. + const body = response.body ? await response.arrayBuffer() : new ArrayBuffer(0); + await this.env.CACHE_BODIES.put(objectKey, body, { + customMetadata: { + status: String(response.status), + createdAt: String(policy.createdAt), + initialAge: String(policy.initialAge), + }, + }); - let publication: PublicationResult; - try { - publication = await metadata.publish(keyHash, revision, candidate); - } catch (error) { - await this.deletePendingObjects(metadata, [objectKey]); - throw error; - } + const publication: PublicationResult = await metadata.publish(keyHash, revision, candidate); if (!publication.published) { - await this.deletePendingObjects(metadata, [objectKey]); - return { published: false, entry: await metadata.getEntry(keyHash) }; + await this.deleteObjects([objectKey]); + return { published: false, entry: publication.entry }; } - await metadata - .finishPendingObjects([objectKey]) - .catch((error) => this.logCleanupFailure(objectKey, error)); - - const previousObjectKey = publication.previousObjectKey; - if (previousObjectKey && previousObjectKey !== objectKey) { - await this.trackPendingObjects(metadata, [previousObjectKey], Date.now()).catch((error) => - this.logCleanupFailure(previousObjectKey, error), - ); - await this.deletePendingObjects(metadata, [previousObjectKey]); + if (publication.previousObjectKey && publication.previousObjectKey !== objectKey) { + await this.deleteObjects([publication.previousObjectKey]); } - return { published: true, entry: await metadata.getEntry(keyHash) }; + return { published: true, entry: publication.entry }; } private async regenerateEntry( @@ -560,11 +519,8 @@ export class ResponseStoreBinding extends WorkerEntrypoint< } const cacheRequest = new Request(`https://runtime-cache.invalid${entry.cacheKey}`); - const writeReservation = reservation ?? { - cacheKey: entry.cacheKey, - keyHash: entry.keyHash, - revision: await metadata.beginWrite(entry.keyHash, entry.cacheKey), - }; + const writeReservation = + reservation ?? (await this.reserveWrite(metadata, entry.keyHash, entry.cacheKey)); const response = await origin.regenerate({ request: cacheRequest, @@ -594,6 +550,7 @@ export class ResponseStoreBinding extends WorkerEntrypoint< entry.keyHash, entry.activeRevision, entry.cacheKey, + this.objectKeyPrefix(entry.keyHash), Date.now(), BACKGROUND_REVALIDATION_LEASE_MS, ); @@ -605,6 +562,7 @@ export class ResponseStoreBinding extends WorkerEntrypoint< await this.regenerateEntry(metadata, entry, "swr", { cacheKey: entry.cacheKey, keyHash: entry.keyHash, + objectKey: claim.objectKey, revision: claim.revision, }); } catch (error) { @@ -668,7 +626,6 @@ export class ResponseStoreBinding extends WorkerEntrypoint< options: ResponseStorePutOptions = {}, ): Promise { const metadata = this.getMetadata(); - this.ctx.waitUntil(this.cleanupExpiredPendingObjects(metadata)); const result = await this.storeResponse(metadata, request, response, options.revalidator); if (!result.published || !result.entry) { @@ -743,9 +700,7 @@ export class ResponseStoreBinding extends WorkerEntrypoint< } if (purged.length > 0) { - const objectKeys = purged.map((entry) => entry.objectKey); - await this.trackPendingObjects(metadata, objectKeys, Date.now()); - await this.deletePendingObjects(metadata, objectKeys); + await this.deleteObjects(purged.map((entry) => entry.objectKey)); } return { backingStoreUpdated: true, edgePurgeAccepted }; diff --git a/packages/workers-response-store/src/metadata-do.ts b/packages/workers-response-store/src/metadata-do.ts index eaf083f68..8dd6c3c34 100644 --- a/packages/workers-response-store/src/metadata-do.ts +++ b/packages/workers-response-store/src/metadata-do.ts @@ -11,14 +11,21 @@ import type { type RevalidationClaim = { claimId: string; + objectKey: string; revision: number; }; type PublicationResult = { + entry: StoredEntry | null; published: boolean; previousObjectKey?: string; }; +type WriteReservation = { + objectKey: string; + revision: number; +}; + type TagIndexEntry = { keyHash: string; tag: string; @@ -26,10 +33,17 @@ type TagIndexEntry = { export type CacheMetadataStub = DurableObjectStub & { beginWrite(keyHash: string, cacheKey: string): Promise; + reserveWrite( + keyHash: string, + cacheKey: string, + objectKeyPrefix: string, + createdAt: number, + ): Promise; claimRevalidation( keyHash: string, activeRevision: number, cacheKey: string, + objectKeyPrefix: string, now: number, leaseMs: number, ): Promise; @@ -38,6 +52,7 @@ export type CacheMetadataStub = DurableObjectStub & { trackPendingObjects(objectKeys: string[], createdAt: number): Promise; finishPendingObjects(objectKeys: string[]): Promise; listExpiredPendingObjects(cutoff: number, limit: number): Promise; + sweepExpiredPendingObjects(cutoff?: number): Promise; publish( keyHash: string, revision: number, @@ -76,6 +91,12 @@ type TagRow = Record & { }; const MAX_SQL_PARAMETERS = 100; +const ORPHAN_RETENTION_MS = 60 * 60 * 1000; +const ORPHAN_CLEANUP_LIMIT = 100; + +type CacheMetadataEnv = { + CACHE_BODIES: R2Bucket; +}; function normalizeTags(tags: string[]): string[] { const normalized = tags.map((tag) => tag.trim().toLowerCase()).filter(Boolean); @@ -138,8 +159,10 @@ function storedEntriesFromRows(rows: EntryRow[]): StoredEntry[] { return entries; } -export class CacheMetadata extends DurableObject> { - constructor(ctx: DurableObjectState, env: Record) { +export class CacheMetadata extends DurableObject { + private cleanupAlarmKnown = false; + + constructor(ctx: DurableObjectState, env: CacheMetadataEnv) { super(ctx, env); void ctx.blockConcurrencyWhile(async () => { ctx.storage.sql.exec(` @@ -224,6 +247,16 @@ export class CacheMetadata extends DurableObject> { }); } + private async ensureCleanupAlarm(createdAt: number): Promise { + if (this.cleanupAlarmKnown) return; + + const current = await this.ctx.storage.getAlarm(); + if (current === null) { + await this.ctx.storage.setAlarm(createdAt + ORPHAN_RETENTION_MS); + } + this.cleanupAlarmKnown = true; + } + private findMatchingEntryRows(options: ResponseStorePurgeOptions): EntryRow[] { if (options.purgeEverything) { return this.ctx.storage.sql @@ -272,7 +305,7 @@ export class CacheMetadata extends DurableObject> { return [...matches.values()]; } - trackPendingObjects(objectKeys: string[], createdAt: number): void { + async trackPendingObjects(objectKeys: string[], createdAt: number): Promise { if (!objectKeys.length) { return; } @@ -286,10 +319,11 @@ export class CacheMetadata extends DurableObject> { ); } }); + await this.ensureCleanupAlarm(createdAt); } - trackPendingObject(objectKey: string, createdAt: number): void { - this.trackPendingObjects([objectKey], createdAt); + trackPendingObject(objectKey: string, createdAt: number): Promise { + return this.trackPendingObjects([objectKey], createdAt); } finishPendingObjects(objectKeys: string[]): void { @@ -320,14 +354,45 @@ export class CacheMetadata extends DurableObject> { .map((row) => row.object_key); } - claimRevalidation( + async sweepExpiredPendingObjects(cutoff = Date.now() - ORPHAN_RETENTION_MS): Promise { + const objectKeys = this.listExpiredPendingObjects(cutoff, ORPHAN_CLEANUP_LIMIT); + if (objectKeys.length) { + await this.env.CACHE_BODIES.delete(objectKeys); + this.finishPendingObjects(objectKeys); + } + + this.cleanupAlarmKnown = false; + const next = this.ctx.storage.sql + .exec<{ created_at: number | null }>( + `SELECT MIN(pending_objects.created_at) AS created_at FROM pending_objects + LEFT JOIN entries + ON entries.object_key = pending_objects.object_key AND entries.tombstoned = 0 + WHERE entries.object_key IS NULL`, + ) + .one().created_at; + if (objectKeys.length === ORPHAN_CLEANUP_LIMIT) { + await this.ctx.storage.setAlarm(Date.now()); + this.cleanupAlarmKnown = true; + } else if (next !== null) { + await this.ensureCleanupAlarm(next); + } + + return objectKeys.length; + } + + async alarm(): Promise { + await this.sweepExpiredPendingObjects(); + } + + async claimRevalidation( keyHash: string, activeRevision: number, cacheKey: string, + objectKeyPrefix: string, now: number, leaseMs: number, - ): RevalidationClaim | null { - return this.ctx.storage.transactionSync(() => { + ): Promise { + const claim = this.ctx.storage.transactionSync(() => { const entry = this.ctx.storage.sql .exec<{ active_revision: number | null; latest_revision: number; tombstoned: number }>( `SELECT active_revision, latest_revision, tombstoned @@ -352,6 +417,7 @@ export class CacheMetadata extends DurableObject> { const revision = entry.latest_revision + 1; const claimId = crypto.randomUUID(); + const objectKey = `${objectKeyPrefix}/${revision}`; this.ctx.storage.sql.exec( "UPDATE entries SET latest_revision = ? WHERE key_hash = ? AND active_revision = ?", @@ -370,9 +436,16 @@ export class CacheMetadata extends DurableObject> { now, now + leaseMs, ); + this.ctx.storage.sql.exec( + "INSERT OR REPLACE INTO pending_objects (object_key, created_at) VALUES (?, ?)", + objectKey, + now, + ); - return { claimId, revision }; + return { claimId, objectKey, revision }; }); + if (claim) await this.ensureCleanupAlarm(now); + return claim; } finishRevalidation(keyHash: string, claimId: string): void { @@ -383,46 +456,65 @@ export class CacheMetadata extends DurableObject> { ); } - beginWrite(keyHash: string, cacheKey: string): number { - return this.ctx.storage.transactionSync(() => { - const current = this.ctx.storage.sql - .exec<{ latest_revision: number }>( - "SELECT latest_revision FROM entries WHERE key_hash = ?", - keyHash, - ) - .toArray()[0]; - const revision = (current?.latest_revision ?? 0) + 1; + private reserveRevision(keyHash: string, cacheKey: string): number { + const current = this.ctx.storage.sql + .exec<{ latest_revision: number }>( + "SELECT latest_revision FROM entries WHERE key_hash = ?", + keyHash, + ) + .toArray()[0]; + const revision = (current?.latest_revision ?? 0) + 1; - if (!current) { - this.ctx.storage.sql.exec( - "INSERT INTO entries (key_hash, cache_key, latest_revision, tombstoned) VALUES (?, ?, ?, 1)", - keyHash, - cacheKey, - revision, - ); - } else { - this.ctx.storage.sql.exec( - "UPDATE entries SET cache_key = ?, latest_revision = ? WHERE key_hash = ?", - cacheKey, - revision, - keyHash, - ); - } + if (!current) { + this.ctx.storage.sql.exec( + "INSERT INTO entries (key_hash, cache_key, latest_revision, tombstoned) VALUES (?, ?, ?, 1)", + keyHash, + cacheKey, + revision, + ); + } else { + this.ctx.storage.sql.exec( + "UPDATE entries SET cache_key = ?, latest_revision = ? WHERE key_hash = ?", + cacheKey, + revision, + keyHash, + ); + } + + return revision; + } - return revision; + beginWrite(keyHash: string, cacheKey: string): number { + return this.ctx.storage.transactionSync(() => this.reserveRevision(keyHash, cacheKey)); + } + + async reserveWrite( + keyHash: string, + cacheKey: string, + objectKeyPrefix: string, + createdAt: number, + ): Promise { + const reservation = this.ctx.storage.transactionSync(() => { + const revision = this.reserveRevision(keyHash, cacheKey); + const objectKey = `${objectKeyPrefix}/${revision}`; + this.ctx.storage.sql.exec( + "INSERT OR REPLACE INTO pending_objects (object_key, created_at) VALUES (?, ?)", + objectKey, + createdAt, + ); + return { objectKey, revision }; }); + await this.ensureCleanupAlarm(createdAt); + return reservation; } publish(keyHash: string, revision: number, metadata: CandidateMetadata): PublicationResult { return this.ctx.storage.transactionSync(() => { const current = this.ctx.storage.sql - .exec<{ latest_revision: number; object_key: string | null }>( - "SELECT latest_revision, object_key FROM entries WHERE key_hash = ?", - keyHash, - ) + .exec("SELECT * FROM entries WHERE key_hash = ?", keyHash) .toArray()[0]; if (!current || current.latest_revision !== revision) { - return { published: false }; + return { entry: current ? storedEntryFromRow(current) : null, published: false }; } const responseMetadataIsInR2 = metadata.responseMetadataInR2 === true; @@ -461,7 +553,31 @@ export class CacheMetadata extends DurableObject> { } } + const entry: StoredEntry = { + keyHash, + cacheKey: current.cache_key, + activeRevision: revision, + latestRevision: revision, + objectKey: metadata.objectKey, + statusText: metadata.statusText, + responseHeaders: metadata.responseHeaders, + freshUntil: metadata.freshUntil, + swrUntil: metadata.swrUntil, + revalidator: metadata.revalidator, + cacheTags: metadata.cacheTags, + ...(responseMetadataIsInR2 + ? {} + : { + legacyResponseMetadata: { + status: metadata.status, + createdAt: metadata.createdAt, + initialAge: metadata.initialAge, + }, + }), + }; + return { + entry: published ? entry : null, published, ...(current.object_key ? { previousObjectKey: current.object_key } : {}), }; diff --git a/packages/workers-response-store/tests/e2e.test.mjs b/packages/workers-response-store/tests/e2e.test.mjs index cdf0c0300..7a339d019 100644 --- a/packages/workers-response-store/tests/e2e.test.mjs +++ b/packages/workers-response-store/tests/e2e.test.mjs @@ -520,7 +520,7 @@ test("a newer put wins and the superseded candidate is cleaned up", async () => assert.equal((await r2Objects()).objects.length, 1); }); -test("retention cleanup removes orphaned candidates without deleting active R2 objects", async () => { +test("retention sweep removes orphaned candidates without deleting active R2 objects", async () => { await put("/active-cleanup", "active"); const bucket = await mf.getR2Bucket("CACHE_BODIES", "user-worker"); const stub = await metadataStub(); @@ -530,10 +530,7 @@ test("retention cleanup removes orphaned candidates without deleting active R2 o await stub.trackPendingObject(activeObjectKey, 0); await stub.trackPendingObjects([orphanObjectKey], 0); - await put("/cleanup-trigger", "trigger"); - for (let attempt = 0; attempt < 20 && (await bucket.head(orphanObjectKey)); attempt++) { - await new Promise((resolve) => setTimeout(resolve, 25)); - } + assert.equal(await stub.sweepExpiredPendingObjects(1), 1); assert.equal(await bucket.head(orphanObjectKey), null); assert.notEqual(await bucket.head(activeObjectKey), null);