From 074a9af315902a223760528ddb9d7208fa2e4043 Mon Sep 17 00:00:00 2001 From: Thomas Durieux <5577568+tdurieux@users.noreply.github.com> Date: Mon, 2 Nov 2026 13:58:25 +0000 Subject: [PATCH] fix: fence shared transformations and cancel abandoned cache work --- docs/shared-cache.md | 5 +++ src/core/blob-cache.ts | 4 ++- src/core/cached-response.ts | 29 ++++++++++++++--- src/core/transformed-cache.ts | 12 +++++-- test/shared-blob-cache.test.js | 19 +++++++++++ test/shared-cache-http.test.js | 59 +++++++++++++++++++++++++++++++++- 6 files changed, 118 insertions(+), 10 deletions(-) diff --git a/docs/shared-cache.md b/docs/shared-cache.md index 3847171..34dfa8f 100644 --- a/docs/shared-cache.md +++ b/docs/shared-cache.md @@ -65,6 +65,11 @@ Publication installs a reader lease before releasing its transaction, including duplicate blobs. Decisions retain sniffed text classification for consistent MIME headers on repeat reads; binary responses omit the range-denial header. +Shared result production and delivery use the same repository lifecycle checks as +the legacy transformed cache. Client disconnects cancel the active transformer and +its worker signal, releasing spool reservations. Completed evictions are committed +even when pinned readers still prevent a new admission. + The HTTP fast path applies to individual file responses backed by the shared source pool, including small text files. Decisions include content identity, path, project context, all transformation options, generation/revision, mask, hostname and engine diff --git a/src/core/blob-cache.ts b/src/core/blob-cache.ts index 64d3898..dfcb776 100644 --- a/src/core/blob-cache.ts +++ b/src/core/blob-cache.ts @@ -287,9 +287,11 @@ export class BlobCache { if (this.usage() + reserved + reservation > this.options.budget) { this.evictTo(Math.max(0, Math.min(this.options.budget * (this.options.lowWater ?? 0.9), this.options.budget - reserved - reservation))); } - if (this.usage() + reserved + reservation > this.options.budget) throw new CacheAdmissionError("cache_budget_busy"); + // Eviction unlinks files: commit its metadata even if admission still fails. + if (this.usage() + reserved + reservation > this.options.budget) return null; this.db.prepare("INSERT INTO fills VALUES(?,?,?,?)").run(key, token, Date.now() + 120000, reservation); return true; }, true); + if (owned === null) throw new CacheAdmissionError("cache_budget_busy"); if (owned) break; this.metric("fill_wait"); if (Date.now() >= deadline) throw new CacheAdmissionError("fill_timeout"); diff --git a/src/core/cached-response.ts b/src/core/cached-response.ts index 600802d..b938dbe 100644 --- a/src/core/cached-response.ts +++ b/src/core/cached-response.ts @@ -5,6 +5,7 @@ import { pipeline } from "stream/promises"; import { Response } from "express"; import { performance } from "perf_hooks"; import config from "../config"; +import { assertTransformScopeActive } from "./transformed-cache"; import { BlobStream, CacheAdmissionError, resultBlobCache } from "./blob-cache"; import { AnonymizeTransformer, classifyByName } from "./anonymize-utils"; import { observeCache } from "./cache-observability"; @@ -18,13 +19,18 @@ export function decisionKey(digest: string, options: unknown) { export async function sendSharedContent(content: Readable, res: Response, options: ConstructorParameters[0]): Promise { try { return await sendCached(content, res, options); } - catch (error) { if (error instanceof CacheAdmissionError) return false; throw error; } + catch (error) { if (error instanceof CacheAdmissionError) return false; content.destroy(); throw error; } } async function sendCached(content: Readable, res: Response, options: ConstructorParameters[0]): Promise { const source = (content as BlobStream).cachedBlob; const results = resultBlobCache(); if (!source || !results) return false; + const assertActive = async () => { + if (res.destroyed) throw new Error("response_cancelled"); + await assertTransformScopeActive(options); + if (res.destroyed) throw new Error("response_cancelled"); + }; const encoding = res.req.acceptsEncodings("gzip", "identity"); res.vary("Accept-Encoding"); if (!encoding) { content.destroy(); res.status(406).end(); return true; } @@ -33,6 +39,7 @@ async function sendCached(content: Readable, res: Response, const send = async (store: typeof results, digest: string, isText: boolean, onHit?: () => void) => { const stream = await store.open(digest, encoding === "gzip"); if (!stream) return false; + try { await assertActive(); } catch (error) { stream.destroy(); throw error; } const record = stream.cachedBlob!.record; onHit?.(); content.destroy(); @@ -63,30 +70,42 @@ async function sendCached(content: Readable, res: Response, let processing: Readable | undefined; let transformer: AnonymizeTransformer | undefined; const started = performance.now(); - const abort = () => { if (!res.writableFinished) processing?.destroy(new Error("response_cancelled")); }; + const abort = () => { + if (!res.writableFinished) { + const error = new Error("response_cancelled"); + transformer?.destroy(error); + processing?.destroy(error); + } + }; res.once("close", abort); try { const output = await results.getOrFill(`render-${key}`, async () => { + await assertActive(); processing = await source.store.open(source.record.digest); if (!processing) throw new CacheAdmissionError("source_evicted"); + if (res.destroyed) throw new Error("response_cancelled"); const transform = transformer = new AnonymizeTransformer({ ...options, decisionManaged: true }); const stream = new PassThrough(); void pipeline(processing, transform, stream).catch(error => stream.destroy(error)); return stream; }, async proof => { + await assertActive(); const unchanged = source.record.digest === `git-${proof.gitSha}` || source.record.digest === `sha256-${proof.sha256}`; observeCache(unchanged ? "rewrite_unchanged" : "rewrite_changed", source.record.size, performance.now() - started); if (unchanged) { results.saveDecision(key, source.record.digest, false, !!transformer?.isText); throw new Unchanged(); } return `sha256-${proof.sha256}`; }, source.record.size); const digest = output.cachedBlob!.record.digest; - results.saveDecision(key, digest, true, true); // Keep the result pinned until its HTTP reader has been opened. - try { if (await send(results, digest, true)) return true; } finally { output.destroy(); } + try { + await assertActive(); + results.saveDecision(key, digest, true, true); + if (await send(results, digest, true)) return true; + } finally { output.destroy(); } return false; } catch (error) { if (error instanceof Unchanged) return send(source.store, source.record.digest, !!transformer?.isText); if (error instanceof CacheAdmissionError) { observeCache("decision_bypass", source.record.size); return false; } content.destroy(); throw error; - } finally { res.removeListener("close", abort); processing?.destroy(); } + } finally { res.removeListener("close", abort); transformer?.destroy(); processing?.destroy(); } } diff --git a/src/core/transformed-cache.ts b/src/core/transformed-cache.ts index 8b6292f..8f78469 100644 --- a/src/core/transformed-cache.ts +++ b/src/core/transformed-cache.ts @@ -26,6 +26,14 @@ async function scopeActive(scope?: CacheScope) { && scope.revision === current.contentCacheRevision; } +/** Shared and legacy result caches must enforce the same repository lifecycle. */ +export async function assertTransformScopeActive(options: unknown) { + const settings = options as { repoId?: string; cacheGeneration?: string; cacheRevision?: string } | undefined; + const scope = typeof settings?.repoId === "string" + ? { repoId: settings.repoId, generation: settings.cacheGeneration, revision: settings.cacheRevision } : undefined; + if (!await scopeActive(scope)) throw new AnonymousError("repository_changed", { httpStatus: 409 }); +} + async function removeEntry(metadata: string) { await Promise.all([metadata, metadata.replace(/\.json$/, ".data")].map(path => fs.rm(path, { force: true }))); } @@ -36,9 +44,7 @@ export async function transformedFile(digest: string, options: unknown, output: const settings = options as { repoId?: string; cacheGeneration?: string; cacheRevision?: string } | undefined; const scope: CacheScope | undefined = typeof settings?.repoId === "string" ? { repoId: settings.repoId, generation: settings.cacheGeneration, revision: settings.cacheRevision } : undefined; - const assertActive = async () => { - if (!await scopeActive(scope)) throw new AnonymousError("repository_changed", { httpStatus: 409 }); - }; + const assertActive = () => assertTransformScopeActive(options); const produceActive = async () => { await assertActive(); const changed = await produce(); diff --git a/test/shared-blob-cache.test.js b/test/shared-blob-cache.test.js index 6b3efc5..1da7367 100644 --- a/test/shared-blob-cache.test.js +++ b/test/shared-blob-cache.test.js @@ -262,6 +262,25 @@ describe("shared content-addressed blob pool", function () { assert.equal(cache.disk.prepare("SELECT COUNT(*) n FROM reservations").get().n, 0); } finally { workers.forEach(w => w.kill()); } }); + it("commits completed evictions when pinned readers still prevent admission", async () => { + for (let i = 0; i < 11; i++) await read(await cache.getOrFill("evict-" + i, async () => Readable.from("content-" + i))); + const pinned = await Promise.all(Array.from({ length: 10 }, (_, i) => cache.lookup("evict-" + i))); + let release, started; + const began = new Promise(resolve => { started = resolve; }); + const pending = cache.getOrFill("pending", async () => { + started(); await new Promise(resolve => { release = resolve; }); return Readable.from("pending content"); + }); + try { + await began; + await assert.rejects(cache.getOrFill("denied", async () => Readable.from("new")), + error => error instanceof CacheAdmissionError && error.message === "cache_budget_busy"); + assert.equal(cache.stats().blobs, 10); assert.equal(cache.stats().allocatedBytes, 10 * 4096); + assert.equal(cache.db.prepare("SELECT 1 FROM aliases WHERE key='evict-10'").get(), undefined); + for (const row of cache.db.prepare("SELECT * FROM blobs").all()) + await fs.access(join(root, "blobs", row.digest.split("-")[1].slice(0, 2), row.digest, row.file)); + release(); await read(await pending); + } finally { release?.(); await pending.catch(() => {}); pinned.forEach(stream => stream.destroy()); } + }); it("does not publish failed or hash-invalid fills and releases reservations", async () => { await assert.rejects(cache.getOrFill("invalid", async () => Readable.from("bad"), async () => { throw Error("invalid_hash"); }), /invalid_hash/); assert.equal(cache.stats().blobs, 0); assert.equal(cache.stats().reservedBytes, 0); diff --git a/test/shared-cache-http.test.js b/test/shared-cache-http.test.js index e9b0c30..cb30da5 100644 --- a/test/shared-cache-http.test.js +++ b/test/shared-cache-http.test.js @@ -3,6 +3,7 @@ const { promises: fs } = require("node:fs"); const { tmpdir } = require("node:os"); const { join } = require("node:path"); const http = require("node:http"); +const { setTimeout } = require("node:timers"); const { Readable } = require("node:stream"); const { createHash } = require("node:crypto"); const { gunzipSync } = require("node:zlib"); @@ -47,7 +48,7 @@ describe("shared cache HTTP and telemetry integration", function () { const handled = await sendSharedContent(input, res, { filePath, repoId: req.params.project, terms: req.params.mode === "changed" ? ["Public=>Private"] : ["Alice"] }); if (!handled) res.send(await consume(input)); - } catch (e) { if (!res.headersSent) res.status(500).end(e.message); else res.destroy(e); } + } catch (e) { if (!res.headersSent) res.status(e.httpStatus || 500).end(e.message); else res.destroy(e); } }); await new Promise(resolve => { server = app.listen(0, "127.0.0.1", resolve); }); port = server.address().port; }); @@ -85,6 +86,62 @@ describe("shared cache HTTP and telemetry integration", function () { assert.equal(calls, 1); } finally { GitHubStream.prototype.downloadWithFallback = original; } }); + for (const change of ["expired", "removed", "revision"]) it("rejects a large rewrite retired during worker execution: " + change, async () => { + const db = require("../src/server/database"), model = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default; + const pool = require("../src/core/anonymization-pool"), worker = pool.anonymizeOnWorker, find = model.findOne, connected = db.isConnected; + const current = { status: "ready" }; let release, started; + const began = new Promise(resolve => { started = resolve; }), held = new Promise(resolve => { release = resolve; }); + body = Buffer.from("Public documentation.\n".repeat(20000)); + db.isConnected = true; + model.findOne = () => ({ select() { return this; }, lean() { return this; }, exec: async () => current }); + pool.anonymizeOnWorker = async message => { + started(); await held; + await fs.writeFile(message.output, (await fs.readFile(message.input, "utf8")).replaceAll("Public", "Private")); + return true; + }; + let pending; + try { + pending = request(port, "/a/changed", "identity", headers); await began; + if (change === "revision") current.contentCacheRevision = "new"; else current.status = change; + release(); const r = await pending; + assert.equal(r.status, 409); assert.equal(r.body.toString(), "repository_changed"); + assert.equal(resultBlobCache().stats().blobs, 0); assert.equal(resultBlobCache().stats().decisions, 0); + } finally { release(); await pending?.catch(() => {}); pool.anonymizeOnWorker = worker; model.findOne = find; db.isConnected = connected; } + }); + it("rechecks repository lifecycle on a decision hit before sending bytes", async () => { + await request(port, "/a/changed", "identity", headers); + const db = require("../src/server/database"), model = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default; + const results = resultBlobCache(), open = results.open.bind(results), find = model.findOne, connected = db.isConnected; + const current = { status: "ready" }; db.isConnected = true; + model.findOne = () => ({ select() { return this; }, lean() { return this; }, exec: async () => current }); + results.open = async (...args) => { const stream = await open(...args); current.status = "removed"; return stream; }; + try { + const r = await request(port, "/a/changed", "identity", headers); + assert.equal(r.status, 409); assert.equal(r.body.toString(), "repository_changed"); + } finally { results.open = open; model.findOne = find; db.isConnected = connected; } + }); + it("cancels an active worker and releases spools when the client disconnects", async () => { + const pool = require("../src/core/anonymization-pool"), worker = pool.anonymizeOnWorker; + const baseline = pool.getAnonymizationPoolStats().spoolBytes; + let started, aborted, rejectWork, client; + const began = new Promise(resolve => { started = resolve; }), cancelled = new Promise(resolve => { aborted = resolve; }); + body = Buffer.from("Public documentation.\n".repeat(20000)); + pool.anonymizeOnWorker = (_message, _size, signal) => new Promise((_resolve, reject) => { + rejectWork = reject; started(); + signal.addEventListener("abort", () => { aborted(); reject(Error("anonymization_cancelled")); }, { once: true }); + }); + try { + client = http.get({ host: "127.0.0.1", port, path: "/a/changed", headers }); client.on("error", () => {}); + await began; assert.ok(pool.getAnonymizationPoolStats().spoolBytes > baseline); + client.destroy(); + await Promise.race([cancelled, new Promise((_, reject) => setTimeout(() => reject(Error("worker was not cancelled")), 2000))]); + for (let i = 0; i < 100 && (pool.getAnonymizationPoolStats().spoolBytes !== baseline || resultBlobCache().stats().reservedBytes); i++) + await new Promise(resolve => setTimeout(resolve, 10)); + assert.equal(pool.getAnonymizationPoolStats().spoolBytes, baseline); + assert.equal(resultBlobCache().stats().reservedBytes, 0); + assert.equal(resultBlobCache().stats().blobs, 0); + } finally { client?.destroy(); rejectWork?.(Error("test cleanup")); pool.anonymizeOnWorker = worker; } + }); for (const changed of [false, true]) it("preserves sniffed text MIME on initial and cached responses: " + changed, async () => { for (const encoding of ["gzip", "identity"]) { const r = await request(port, "/a/" + (changed ? "changed" : "plain") + "?name=data.unknownext", encoding, headers);