fix: fence shared transformations and cancel abandoned cache work

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

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");

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<typeof AnonymizeTransformer>[0]): Promise<boolean> {
   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<typeof AnonymizeTransformer>[0]): Promise<boolean> {
   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(); }
 }

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();

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);

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);