Skip to content

perf(cache): remove eager response store tag index #3211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
james-elicx wants to merge 1 commit into from
Closed

perf(cache): remove eager response store tag index #3211

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 11 additions & 98 deletions packages/workers-response-store/src/metadata-do.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,6 @@ type WriteReservation = {
revision: number;
};

type TagIndexEntry = {
keyHash: string;
tag: string;
};

export type CacheMetadataStub = DurableObjectStub & {
beginWrite(keyHash: string, cacheKey: string): Promise<number>;
reserveWrite(
Expand Down Expand Up @@ -63,7 +58,6 @@ export type CacheMetadataStub = DurableObjectStub & {
getEntriesMatching(options: ResponseStoreRefreshOptions): Promise<StoredEntry[]>;
purgeMatching(options: ResponseStorePurgeOptions): Promise<PurgedEntry[]>;
inspect(): Promise<StoredEntry[]>;
inspectTagIndex(): Promise<TagIndexEntry[]>;
};

type EntryRow = Record<string, SqlStorageValue> & {
Expand All @@ -85,11 +79,6 @@ type EntryRow = Record<string, SqlStorageValue> & {
tombstoned: number;
};

type TagRow = Record<string, SqlStorageValue> & {
key_hash: string;
tag: string;
};

const MAX_SQL_PARAMETERS = 100;
const ORPHAN_RETENTION_MS = 60 * 60 * 1000;
const ORPHAN_CLEANUP_LIMIT = 100;
Expand Down Expand Up @@ -212,38 +201,6 @@ export class CacheMetadata extends DurableObject<CacheMetadataEnv> {
);
CREATE INDEX IF NOT EXISTS pending_objects_created_at ON pending_objects(created_at);
`);

const hasBackfilledTagIndex =
ctx.storage.sql
.exec<{ version: number }>(
"SELECT version FROM metadata_schema_migrations WHERE version = 1",
)
.toArray().length > 0;

if (!hasBackfilledTagIndex) {
ctx.storage.transactionSync(() => {
const rows = ctx.storage.sql
.exec<{ key_hash: string; cache_tags: string | null }>(
`SELECT key_hash, cache_tags FROM entries
WHERE tombstoned = 0 AND active_revision IS NOT NULL`,
)
.toArray();

for (const row of rows) {
const tags = JSON.parse(row.cache_tags ?? "[]") as string[];

for (const tag of normalizeTags(tags)) {
ctx.storage.sql.exec(
"INSERT OR IGNORE INTO entry_tags (tag, key_hash) VALUES (?, ?)",
tag,
row.key_hash,
);
}
}

ctx.storage.sql.exec("INSERT INTO metadata_schema_migrations (version) VALUES (1)");
});
}
});
}

Expand All @@ -266,43 +223,18 @@ export class CacheMetadata extends DurableObject<CacheMetadataEnv> {
.toArray();
}

const matches = new Map<string, EntryRow>();
const tags = normalizeTags(options.tags ?? []);

for (let offset = 0; offset < tags.length; offset += MAX_SQL_PARAMETERS) {
const batch = tags.slice(offset, offset + MAX_SQL_PARAMETERS);
const placeholders = batch.map(() => "?").join(", ");
const rows = this.ctx.storage.sql
.exec<EntryRow>(
`SELECT DISTINCT entries.* FROM entries
INNER JOIN entry_tags ON entry_tags.key_hash = entries.key_hash
WHERE entries.tombstoned = 0 AND entries.active_revision IS NOT NULL
AND entry_tags.tag IN (${placeholders})`,
...batch,
)
.toArray();

for (const row of rows) {
matches.set(row.key_hash, row);
}
}

const tags = new Set(normalizeTags(options.tags ?? []));
const prefixes = options.pathPrefixes ?? [];
if (prefixes.length) {
const rows = this.ctx.storage.sql
.exec<EntryRow>(
"SELECT * FROM entries WHERE tombstoned = 0 AND active_revision IS NOT NULL",
)
.toArray();

for (const row of rows) {
if (prefixes.some((prefix) => row.cache_key.startsWith(prefix))) {
matches.set(row.key_hash, row);
}
}
}

return [...matches.values()];
return this.ctx.storage.sql
.exec<EntryRow>("SELECT * FROM entries WHERE tombstoned = 0 AND active_revision IS NOT NULL")
.toArray()
.filter(
(row) =>
prefixes.some((prefix) => row.cache_key.startsWith(prefix)) ||
normalizeTags(JSON.parse(row.cache_tags ?? "[]") as string[]).some((tag) =>
tags.has(tag),
),
);
}

async trackPendingObjects(objectKeys: string[], createdAt: number): Promise<void> {
Expand Down Expand Up @@ -541,17 +473,6 @@ export class CacheMetadata extends DurableObject<CacheMetadataEnv> {
);

const published = update.rowsWritten === 1;
if (published) {
this.ctx.storage.sql.exec("DELETE FROM entry_tags WHERE key_hash = ?", keyHash);

for (const tag of normalizeTags(metadata.cacheTags)) {
this.ctx.storage.sql.exec(
"INSERT INTO entry_tags (tag, key_hash) VALUES (?, ?)",
tag,
keyHash,
);
}
}

const entry: StoredEntry = {
keyHash,
Expand Down Expand Up @@ -654,7 +575,6 @@ export class CacheMetadata extends DurableObject<CacheMetadataEnv> {
"DELETE FROM revalidation_claims WHERE key_hash = ?",
row.key_hash,
);
this.ctx.storage.sql.exec("DELETE FROM entry_tags WHERE key_hash = ?", row.key_hash);
}

return matches.map((row) => ({
Expand All @@ -671,11 +591,4 @@ export class CacheMetadata extends DurableObject<CacheMetadataEnv> {
.toArray();
return storedEntriesFromRows(rows);
}

inspectTagIndex(): TagIndexEntry[] {
return this.ctx.storage.sql
.exec<TagRow>("SELECT key_hash, tag FROM entry_tags ORDER BY tag, key_hash")
.toArray()
.map((row) => ({ keyHash: row.key_hash, tag: row.tag }));
}
}
17 changes: 3 additions & 14 deletions packages/workers-response-store/tests/e2e.test.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,10 +113,6 @@ async function metadata() {
return (await metadataStub()).inspect();
}

async function tagIndex() {
return (await metadataStub()).inspectTagIndex();
}

async function r2Objects() {
const bucket = await mf.getR2Bucket("CACHE_BODIES", "user-worker");
return bucket.list();
Expand Down Expand Up @@ -422,7 +418,7 @@ test("refresh accepts more tag selectors than one SQLite parameter batch", async
assert.equal(await (await read("/refresh-many-tags")).text(), "refreshed");
});

test("refresh and purge use a reverse tag-to-entry index that follows publication", async () => {
test("refresh and purge select entries from their stored tags", async () => {
await put("/tag-index", "seed", {
tags: ["Original", "Shared"],
revalidator: {
Expand All @@ -432,25 +428,18 @@ test("refresh and purge use a reverse tag-to-entry index that follows publicatio
},
});

assert.deepEqual(
(await tagIndex()).map(({ tag }) => tag),
["original", "shared"],
);
assert.deepEqual((await metadata())[0].cacheTags, ["Original", "Shared"]);
assert.deepEqual((await refreshSelectors({ tags: ["ORIGINAL"] })).json, {
backingStoreUpdated: true,
edgePurgeAccepted: false,
});
assert.deepEqual(
(await tagIndex()).map(({ tag }) => tag),
["replacement"],
);
assert.deepEqual((await metadata())[0].cacheTags, ["Replacement"]);
assert.deepEqual((await refreshSelectors({ tags: ["original"] })).json, {
backingStoreUpdated: false,
edgePurgeAccepted: false,
});

await purge({ tags: ["REPLACEMENT"] });
assert.deepEqual(await tagIndex(), []);
assert.equal((await read("/tag-index")).status, 404);
assert.ok((await (await metadataStub()).getTagExpiration(["replacement"])) > 0);
});
Expand Down
Loading