merge: reconcile paid hosting with shared cache and main

test/anonymize-regex-prefix.test.js

@@ -0,0 +1,52 @@
+const { expect } = require("chai");
+require("ts-node/register/transpile-only");
+const { ContentAnonimizer } = require("../src/core/anonymize-utils");
+
+function unfiltered(terms, text) {
+  const a = new ContentAnonimizer({ terms });
+  a.termPresence = null;
+  for (const term of a.compiledTerms) delete term.requiredPrefix;
+  return a.anonymize(text);
+}
+
+describe("regex required-prefix searches", function () {
+  it("keeps RE2 results for assertions, escaped punctuation, optional atoms and quantifiers", function () {
+    const terms = [String.raw`\b[Aa]uthor\b`, String.raw`\Bauthor\B`, String.raw`author\.example\.org`,
+      "Research[ _-]?Institute", "a?b?c?d?e?required", "a??bc", "a+b", "a*b", "a{0,2}b", "a{2}b",
+      String.raw`\(Author\)`, String.raw`\ba?\b`, "[a-z]Name", "[^x]Name", "a|b", "(a)?b", "^Author", "Author$",
+      String.raw`Author\s+Name`, String.raw`Author\d+`, String.raw`Author\x41`, "Σ?Author", "😀?Author"];
+    const text = "Author author xauthorx author.example.org authorXexampleYorg Research-Institute Research Institute " +
+      "ResearchInstitute required aaaaab ab b aabc abc (Author) aName xName Author Name Author12 AuthorA ΣAuthor 😀Author";
+    for (const term of terms) {
+      const a = new ContentAnonimizer({ terms: [term] });
+      expect(a.anonymize(text), term).to.equal(unfiltered([term], text));
+    }
+  });
+
+  it("preserves cascades, anchors and external boundaries after skipping unmatched prefixes", function () {
+    const cases = [
+      [[String.raw`\bAuthor\b=>Research Institute`, "Research[ _-]?Institute=>Hidden"], "Author"],
+      [["a?bc=>Hidden"], "xbc abc bc abcx abc"],
+      [["a+b=>Hidden"], "xaaab aaab aabx ab"],
+      [[String.raw`\b[Aa]uthor\b=>Hidden`], "éAuthor Authoré Author xAuthor Author"],
+      [[String.raw`\bAuthor\b=>Hidden`], "😀Author\nAuthor😀"],
+      [["a?=>X"], "xyz"],
+      [["a{0}=>X"], "xyz"],
+      [[String.raw`\bAuthor\b=>First`, "^First=>Second", "Second$=>Last"], "Author"],
+    ];
+    for (const [terms, text] of cases) {
+      expect(new ContentAnonimizer({ terms }).anonymize(text), JSON.stringify(terms)).to.equal(unfiltered(terms, text));
+    }
+  });
+
+  it("processes sparse boundary and optional regex rules in a large file within the worker deadline", function () {
+    this.timeout(20000);
+    const body = "const value = [1,2,3,4,5]; // ordinary application data\n".repeat(290000);
+    const input = `Author Research Project\n${body}\nAuthor`;
+    const terms = [String.raw`\b[Aa]uthor\b=>Hidden`, "Research[ _-]?Institute=>Organization",
+      ...Array.from({ length: 24 }, (_, i) => String.raw`\bResearcher${i}\b=>Hidden`),
+      String.raw`research\.example\.org=>Website`];
+    const a = new ContentAnonimizer({ terms });
+    expect(a.anonymize(input, 10000)).to.equal(`Hidden Research Project\n${body}\nHidden`);
+  });
+});

test/download-header-performance.test.js

@@ -72,6 +72,17 @@ describe("download headers and website revalidation", function () {
     expect(response.headers["content-disposition"]).to.include("attachment");
     expect(response.headers["x-content-type-options"]).to.equal("nosniff");
   });
+  it("uses representation-safe validators for shared-cache HEAD and revalidation without reading bytes", async function () {
+    const previous = config.SHARED_CACHE_ENABLED; config.SHARED_CACHE_ENABLED = true;
+    try {
+      const path = "/api/repo/fixture/file/data.txt";
+      const head = await fetch(path, "HEAD", { "accept-encoding": "gzip" });
+      expect(head.status).to.equal(200); expect(head.headers.etag).to.match(/^W\/"f-/);
+      expect(head.headers.vary).to.include("Accept-Encoding"); expect(head.body).to.equal("");
+      const fresh = await fetch(path, "GET", { "if-none-match": head.headers.etag, "accept-encoding": "identity" });
+      expect(fresh.status).to.equal(304); expect(transfers).to.equal(0);
+    } finally { config.SHARED_CACHE_ENABLED = previous; }
+  });
   it("preserves content gates for HEAD", async function () {
     repo.options.pdf = false;
     expect((await fetch("/api/repo/fixture/file/private.pdf", "HEAD")).status).to.equal(403);

test/download-progress-fences.test.js

@@ -0,0 +1,78 @@
+const { expect } = require("chai");
+const { setImmediate } = require("timers");
+require("ts-node/register/transpile-only");
+const db = require("../src/server/database");
+const Model = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default;
+const Repository = require("../src/core/Repository").default;
+const gh = require("../src/core/GitHubUtils");
+const sift = require("sift").default;
+
+describe("download progress publication fences", function () {
+  const restores = [];
+  function stub(object, key, value) { const before = object[key]; restores.push(() => { object[key] = before; }); object[key] = value; }
+  afterEach(() => { while (restores.length) restores.pop()(); });
+  function setup() {
+    const model = new Model({ repoId: "progress-test", status: "download", statusDate: new Date(1000),
+      owner: "507f1f77bcf86cd799439011", anonymizeDate: new Date(500),
+      source: { repositoryName: "owner/repo", commit: "abc", branch: "main" }, options: { terms: [], expirationMode: "never" } });
+    const repo = new Repository(model); repo.protectLifecycle = true;
+    const stored = model.toObject();
+    stub(db, "isConnected", true);
+    stub(Model, "updateOne", (filter, update) => ({ exec: async () => {
+      if (!sift(filter)(stored)) return { matchedCount: 0 };
+      Object.assign(stored, update.$set); return { matchedCount: 1 };
+    } }));
+    return { repo, stored };
+  }
+
+  it("keeps a captured publication fence valid when a queue progress tick runs", async function () {
+    const { repo, stored } = setup(); let tick;
+    stub(global, "setInterval", callback => { tick = callback; return 1; });
+    stub(global, "clearInterval", () => {});
+    stub(db, "connect", async () => {});
+    stub(db, "getRepository", async () => repo);
+    stub(gh, "getToken", async () => "test-token");
+    stub(gh, "getRedisGateResetAt", async () => 0);
+    stub(repo, "resetSate", async () => {});
+    stub(repo, "anonymize", async progress => {
+      const fence = { status: stored.status, statusDate: stored.statusDate };
+      progress("Writing files"); tick();
+      await new Promise(resolve => setImmediate(resolve));
+      expect(sift(fence)(stored), "progress invalidated the publication fence").to.equal(true);
+      expect(stored.statusMessage).to.equal("Writing files");
+    });
+    await require("../src/queue/processes/downloadRepository").default({ data: { repoId: repo.repoId }, updateProgress: async () => {} });
+    expect(stored.status).to.equal("ready");
+  });
+
+  it("keeps fresh long-running downloads busy without changing statusDate", async function () {
+    const { repo, stored } = setup(); const before = repo.model.statusDate;
+    await repo.updateProgressMessage("Writing files");
+    expect(stored.statusDate).to.deep.equal(before);
+    expect(repo.model.statusDate).to.deep.equal(before);
+    expect(repo.statusActivityDate.getTime()).to.be.greaterThan(Date.now() - 1000);
+    try { await repo.check(); throw new Error("expected not ready"); }
+    catch (error) { expect(error.message).to.equal("repository_not_ready"); }
+  });
+
+  for (const replacement of ["ready", "removed", "download"]) {
+    it(`does not overwrite progress for a newer ${replacement} state`, async function () {
+      const { repo, stored } = setup();
+      stored.status = replacement; stored.statusDate = new Date(2000); stored.statusMessage = "new operation";
+      await repo.updateProgressMessage("old progress");
+      expect(stored.statusMessage).to.equal("new operation");
+      expect(repo.model.statusMessage).not.to.equal("old progress");
+      expect(stored).not.to.have.property("progressDate");
+    });
+  }
+
+  it("does not apply a completed progress write to a locally advanced status", async function () {
+    const { repo } = setup();
+    stub(Model, "updateOne", () => ({ exec: async () => {
+      repo.model.status = "ready"; repo.model.statusDate = new Date(2000); repo.model.statusMessage = "done";
+      return { matchedCount: 1 };
+    } }));
+    await repo.updateProgressMessage("old progress");
+    expect(repo.model.statusMessage).to.equal("done");
+  });
+});

test/fixtures/shared-cache-worker.js

@@ -0,0 +1,28 @@
+require("ts-node/register/transpile-only");
+const { setTimeout } = require("node:timers");
+const { Readable } = require("node:stream");
+const { BlobCache } = require("../../src/core/blob-cache");
+require("../../src/config").default.CACHE_MONITORING_ENABLED = false;
+const cache = new BlobCache(JSON.parse(process.env.BLOB_TEST_OPTIONS));
+if (process.env.BLOB_TEST_CRASH_AFTER_RENAME) {
+  const fs = require("node:fs").promises, rename = fs.rename.bind(fs);
+  fs.rename = async (...args) => {
+    await rename(...args);
+    process.send({ renamed: args[1] });
+    await new Promise(() => {}); // Parent kills the process before index publication.
+  };
+}
+if (process.env.BLOB_TEST_DISK_BYTES) require("node:fs").statfsSync = () => ({ bavail: Number(process.env.BLOB_TEST_DISK_BYTES), bsize: 1 });
+process.send("ready");
+process.once("message", async () => {
+  try {
+    const stream = await cache.getOrFill("multiprocess", async () => {
+      process.send("fill");
+      if (process.env.BLOB_TEST_HOLD_FILL) await new Promise(resolve => process.once("message", resolve));
+      else await new Promise(resolve => setTimeout(resolve, 100));
+      return Readable.from("shared across processes");
+    });
+    let body = ""; for await (const chunk of stream) body += chunk;
+    cache.close(); process.send({ body }); process.disconnect();
+  } catch (error) { process.send({ error: error.message }); process.exitCode = 1; process.disconnect(); }
+});

test/monitoring-access.test.js

@@ -16,9 +16,9 @@ describe("performance report access", function () {
     server = await new Promise(resolve => { const s = app.listen(0, "127.0.0.1", () => resolve(s)); });
   });
   after(async function () { server.closeAllConnections(); await new Promise(resolve => server.close(resolve)); utils.getUser = original; });
-  async function request(user) {
+  async function request(user, path = "/api/admin/performance") {
     return new Promise((resolve, reject) => {
-      http.get({ host: "127.0.0.1", port: server.address().port, path: "/api/admin/performance", headers: user ? { "x-fixture-user": user } : {} }, res => {
+      http.get({ host: "127.0.0.1", port: server.address().port, path, headers: user ? { "x-fixture-user": user } : {} }, res => {
         let body = ""; res.on("data", chunk => { body += chunk; }); res.on("end", () => resolve({ status: res.statusCode, body, headers: res.headers }));
       }).on("error", reject);
     });
@@ -29,4 +29,12 @@ describe("performance report access", function () {
     const result = await request("admin"); expect(result.status).to.equal(200);
     expect(JSON.parse(result.body).available).to.equal(false); expect(result.headers["cache-control"]).to.equal("no-store");
   });
+  it("restricts cache telemetry to admins and disables response caching", async function () {
+    expect((await request(undefined, "/api/admin/cache")).status).to.equal(401);
+    expect((await request("regular", "/api/admin/cache")).status).to.equal(401);
+    const result = await request("admin", "/api/admin/cache");
+    expect(result.status).to.equal(200); expect(result.headers["cache-control"]).to.equal("no-store");
+    expect(JSON.parse(result.body)).to.have.property("available");
+  });
+
 });

test/shared-blob-cache.test.js

@@ -0,0 +1,363 @@
+const { strict: assert } = require("node:assert");
+const { promises: fs } = require("node:fs");
+const { tmpdir } = require("node:os");
+const { join } = require("node:path");
+const { Readable } = require("node:stream");
+const { createHash, randomBytes } = require("node:crypto");
+const { setTimeout: delay } = require("node:timers/promises");
+const { gunzipSync } = require("node:zlib");
+require("ts-node/register/transpile-only");
+const { BlobCache, CacheAdmissionError } = require("../src/core/blob-cache");
+const config = require("../src/config").default;
+const read = async stream => { const chunks = []; for await (const b of stream) chunks.push(Buffer.from(b)); return Buffer.concat(chunks); };
+const digest = b => "git-" + createHash("sha1").update(`blob ${b.length}\0`).update(b).digest("hex");
+
+describe("shared content-addressed blob pool", function () {
+  this.timeout(10000);
+  let root, cache, oldMonitoring;
+  beforeEach(async () => { oldMonitoring = config.CACHE_MONITORING_ENABLED; config.CACHE_MONITORING_ENABLED = false;
+    root = await fs.mkdtemp(join(tmpdir(), "shared-blob-test-"));
+    cache = new BlobCache({ root, budget: 65536, maxObject: 4096, gzip: true }); });
+  afterEach(async () => { cache.close(); config.CACHE_MONITORING_ENABLED = oldMonitoring; await fs.rm(root, { recursive: true, force: true }); });
+  it("stores one verified blob across project aliases and independently reads gzip and identity", async () => {
+    const b = Buffer.from("shared text\n".repeat(300));
+    const verify = async p => { assert.equal("git-" + p.gitSha, digest(b)); return digest(b); };
+    assert.deepEqual(await read(await cache.getOrFill("project-a", async () => Readable.from(b), verify)), b);
+    assert.deepEqual(await read(await cache.getOrFill("project-b", async () => Readable.from(b), verify)), b);
+    assert.equal(cache.stats().blobs, 1); assert.ok(cache.stats().storedBytes < b.length / 2);
+    assert.deepEqual(gunzipSync(await read(await cache.open(digest(b), true))), b);
+    assert.deepEqual(await read(await cache.lookup("project-a")), b);
+  });
+  it("keeps summary counters exact across connections, replacement, rollback and migration", async () => {
+    const check = store => {
+      const expected = store.db.prepare(`SELECT COUNT(*) blobs,COALESCE(SUM(size),0) originalBytes,
+        COALESCE(SUM(stored),0) storedBytes,COALESCE(SUM(allocated),0) allocatedBytes,
+        (SELECT COUNT(*) FROM readers) activeReaders,(SELECT COUNT(*) FROM decisions) decisions,
+        (SELECT COUNT(*) FROM aliases) aliases,(SELECT COUNT(*) FROM history) history,
+        (SELECT COALESCE(SUM(reserved),0) FROM fills) reservedBytes FROM blobs`).get();
+      const actual = store.db.prepare("SELECT * FROM summary WHERE id=1").get();
+      for (const [key, value] of Object.entries(expected)) assert.equal(actual[key], value, key);
+    };
+    const stream = await cache.getOrFill("counter-a", async () => Readable.from("same content"));
+    check(cache);
+    const other = new BlobCache(cache.options);
+    try {
+      await read(await other.getOrFill("counter-b", async () => Readable.from("same content")));
+      other.saveDecision("rule", stream.cachedBlob.record.digest, false);
+      other.saveDecision("rule", stream.cachedBlob.record.digest, true);
+      check(cache); check(other);
+      other.db.exec("BEGIN; INSERT OR REPLACE INTO history VALUES('counter-a',0,1); DELETE FROM decisions; ROLLBACK;");
+      check(cache);
+      await read(stream); await delay(20); check(other);
+      other.db.prepare("DELETE FROM blobs").run(); check(cache);
+    } finally { other.close(); }
+    // Simulate an index created before the summary migration.
+    for (const row of cache.db.prepare("SELECT name FROM sqlite_master WHERE type='trigger' AND name LIKE 'summary_%'").all())
+      cache.db.exec('DROP TRIGGER "' + row.name + '"');
+    cache.db.exec("DROP TABLE summary");
+    cache.close(); cache = new BlobCache(cache.options); check(cache);
+  });
+  it("upgrades old decisions without inventing a sniffed content classification", () => {
+    cache.saveDecision("old", "git-" + "a".repeat(40), false);
+    cache.db.exec("ALTER TABLE decisions DROP COLUMN isText");
+    cache.close(); cache = new BlobCache(cache.options);
+    assert.equal(cache.decision("old").isText, undefined);
+    cache.saveDecision("old", "git-" + "a".repeat(40), false, true);
+    assert.equal(cache.decision("old").isText, true);
+    assert.equal(cache.stats().decisions, 1);
+  });
+  it("uses durable commits for reservations and publication while keeping hit bookkeeping disposable", async () => {
+    assert.equal(cache.db.prepare("PRAGMA synchronous").get().synchronous, 1);
+    const original = cache.db.exec.bind(cache.db), commits = [];
+    cache.db.exec = sql => {
+      if (sql === "COMMIT") commits.push(Number(cache.db.prepare("PRAGMA synchronous").get().synchronous));
+      return original(sql);
+    };
+    await read(await cache.getOrFill("durability", async () => Readable.from("durable bytes")));
+    assert.equal(commits.filter(x => x === 2).length, 3);
+    commits.length = 0;
+    await read(await cache.getOrFill("durability", async () => { throw Error("unexpected miss"); }));
+    assert.ok(commits.length > 0); assert.ok(commits.every(x => x === 1));
+    cache.options.budget = 1;
+    await assert.rejects(cache.getOrFill("no-space", async () => Readable.from("x")), CacheAdmissionError);
+    assert.equal(cache.db.prepare("PRAGMA synchronous").get().synchronous, 1);
+  });
+  it("coalesces fills across independent index connections", async () => {
+    const other = new BlobCache(cache.options); let calls = 0;
+    try {
+      const produce = async () => { calls++; await delay(50); return Readable.from("complete output"); };
+      const streams = await Promise.all(Array.from({ length: 8 }, (_, i) => (i % 2 ? other : cache).getOrFill("same", produce)));
+      const values = await Promise.all(streams.map(read)); assert.equal(calls, 1);
+      assert.ok(values.every(b => b.toString() === "complete output"));
+    } finally { other.close(); }
+  });
+  it("coalesces fills across separate operating-system processes", async function () {
+    this.timeout(45000);
+    const { fork } = require("node:child_process");
+    const workers = [0, 1].map(() => fork(join(__dirname, "fixtures/shared-cache-worker.js"), [], {
+      env: { ...process.env, NODE_ENV: "test", BLOB_TEST_OPTIONS: JSON.stringify(cache.options) }, stdio: ["ignore", "ignore", "pipe", "ipc"]
+    }));
+    let fills = 0;
+    const readiness = workers.map(worker => new Promise((resolve, reject) => {
+      let stderr = "";
+      worker.stderr.on("data", chunk => { stderr = (stderr + chunk).slice(-4000); });
+      worker.on("message", message => { if (message === "ready") resolve(); });
+      worker.on("error", reject);
+      worker.on("exit", code => reject(Error(`worker exited ${code}: ${stderr}`)));
+    }));
+    try {
+      const results = workers.map(worker => new Promise((resolve, reject) => {
+        worker.on("message", message => {
+          if (message === "fill") fills++;
+          if (message.error) reject(Error(message.error));
+          if (message.body) resolve(message.body);
+        });
+        worker.on("error", reject);
+        worker.on("exit", code => { if (code) reject(Error("worker exited " + code)); });
+      }));
+      await Promise.all(readiness);
+      workers.forEach(worker => worker.send("start"));
+      assert.deepEqual(await Promise.all(results), ["shared across processes", "shared across processes"]);
+      assert.equal(fills, 1); assert.equal(cache.stats().blobs, 1);
+    } finally { workers.forEach(worker => worker.kill()); }
+  });
+  it("recovers a producer killed after rename without losing its disk reservation", async function () {
+    this.timeout(45000);
+    const { fork } = require("node:child_process");
+    const worker = fork(join(__dirname, "fixtures/shared-cache-worker.js"), [], {
+      env: { ...process.env, NODE_ENV: "test", BLOB_TEST_OPTIONS: JSON.stringify(cache.options), BLOB_TEST_CRASH_AFTER_RENAME: "1" },
+      stdio: ["ignore", "ignore", "pipe", "ipc"]
+    });
+    try {
+      const renamed = await new Promise((resolve, reject) => {
+        let stderr = "";
+        worker.stderr.on("data", chunk => { stderr = (stderr + chunk).slice(-4000); });
+        worker.on("error", reject);
+        worker.on("exit", code => reject(Error("premature exit " + code + ": " + stderr)));
+        worker.on("message", message => {
+          if (message === "ready") worker.send("start");
+          if (message.renamed) resolve(message.renamed);
+          if (message.error) reject(Error(message.error));
+        });
+      });
+      const exited = new Promise(resolve => worker.once("exit", resolve));
+      worker.kill("SIGKILL"); await exited;
+      assert.ok(cache.stats().reservedBytes > 0);
+      assert.equal(cache.disk.prepare("SELECT COUNT(*) n FROM reservations").get().n, 1);
+      assert.equal(cache.stats().blobs, 0);
+      assert.equal(cache.db.prepare("SELECT COUNT(*) n FROM publications").get().n, 1);
+      await fs.access(renamed);
+      cache.db.exec("UPDATE fills SET expires=1");
+      await cache.maintain();
+      await assert.rejects(fs.access(renamed), { code: "ENOENT" });
+      assert.equal(cache.stats().reservedBytes, 0);
+      assert.equal(cache.disk.prepare("SELECT COUNT(*) n FROM reservations").get().n, 0);
+      assert.equal(cache.db.prepare("SELECT COUNT(*) n FROM publications").get().n, 0);
+      assert.equal((await read(await cache.getOrFill("multiprocess", async () => Readable.from("recovered")))).toString(), "recovered");
+    } finally { worker.kill(); }
+  });
+  it("pins a duplicate publication while another connection tries to evict it", async () => {
+    await read(await cache.getOrFill("original", async () => Readable.from("identical")));
+    const other = new BlobCache({ ...cache.options, budget: 8192, maxObject: 1024 });
+    const unlink = fs.unlink.bind(fs); let competed = false;
+    fs.unlink = async path => {
+      if (!competed && path.includes("/blobs/") && !cache.db.prepare("SELECT 1 FROM blobs WHERE file=?").get(require("node:path").basename(path))) {
+        competed = true;
+        await assert.rejects(other.getOrFill("competitor", async () => Readable.from("different")), CacheAdmissionError);
+      }
+      return unlink(path);
+    };
+    try {
+      assert.equal((await read(await cache.getOrFill("duplicate", async () => Readable.from("identical")))).toString(), "identical");
+      assert.ok(competed);
+      await delay(20); assert.equal(cache.stats().activeReaders, 0);
+    } finally { fs.unlink = unlink; other.close(); }
+  });
+  it("turns SQLite write contention into a cache bypass and preserves existing data", async () => {
+    await read(await cache.getOrFill("locked", async () => Readable.from("available")));
+    const other = new BlobCache(cache.options);
+    other.db.exec("BEGIN IMMEDIATE");
+    try {
+      await assert.rejects(cache.getOrFill("locked", async () => { throw Error("unexpected fill"); }),
+        error => error instanceof CacheAdmissionError && error.message === "cache_index_busy");
+    } finally { other.db.exec("ROLLBACK"); other.close(); }
+    assert.equal((await read(await cache.lookup("locked"))).toString(), "available");
+    await delay(20); assert.equal(cache.stats().activeReaders, 0);
+  });
+  it("retries a contended reader heartbeat while its confirmed lease is valid", async () => {
+    await read(await cache.getOrFill("heartbeat", async () => Readable.from("still readable")));
+    const other = new BlobCache(cache.options);
+    const interval = global.setInterval, now = Date.now; let tick, reader;
+    global.setInterval = fn => { tick = fn; return interval(fn, 10000000); };
+    try { reader = await cache.lookup("heartbeat"); } finally { global.setInterval = interval; }
+    const base = now();
+    try {
+      other.db.exec("BEGIN IMMEDIATE"); Date.now = () => base + 31000; tick();
+      assert.equal(reader.destroyed, false);
+      other.db.exec("ROLLBACK"); Date.now = () => base + 32000; tick();
+      assert.ok(Number(cache.db.prepare("SELECT MAX(expires) n FROM readers").get().n) >= base + 152000);
+      Date.now = now;
+      assert.equal((await read(reader)).toString(), "still readable");
+    } finally { Date.now = now; reader.destroy(); if (other.db.isTransaction) other.db.exec("ROLLBACK"); other.close(); }
+  });
+  it("ends a contended reader once its confirmed lease expires", async () => {
+    await read(await cache.getOrFill("expiry", async () => Readable.from("data")));
+    const other = new BlobCache(cache.options), interval = global.setInterval, now = Date.now;
+    let tick, reader;
+    global.setInterval = fn => { tick = fn; return interval(fn, 10000000); };
+    try { reader = await cache.lookup("expiry"); } finally { global.setInterval = interval; }
+    const error = new Promise(resolve => reader.once("error", resolve)), base = now();
+    try {
+      other.db.exec("BEGIN IMMEDIATE"); Date.now = () => base + 121000; tick();
+      assert.equal((await error).message, "cache_reader_lease_lost");
+    } finally { Date.now = now; other.db.exec("ROLLBACK"); reader.destroy(); other.close(); }
+  });
+  for (const outcome of ["recover", "expire", "replaced"]) {
+    it(`handles fill heartbeat contention and ${outcome}`, async () => {
+      const other = new BlobCache(cache.options), interval = global.setInterval, now = Date.now;
+      let tick, started;
+      const ready = new Promise(resolve => { started = resolve; });
+      const input = new Readable({ read() {} });
+      global.setInterval = fn => { tick = fn; return interval(fn, 10000000); };
+      const fill = cache.getOrFill("fill-heartbeat", async () => { started(); return input; });
+      // Attach rejection handling before deliberately terminating the producer.
+      const result = fill.then(read);
+      const checked = outcome === "recover" ? result : assert.rejects(result, /cache_fill_lease_lost/);
+      try {
+        await ready; global.setInterval = interval; await delay(10);
+        const base = now();
+        other.db.exec("BEGIN IMMEDIATE"); Date.now = () => base + 31000; tick();
+        assert.equal(input.destroyed, false);
+        other.db.exec("ROLLBACK");
+        if (outcome === "recover") {
+          Date.now = () => base + 32000; tick();
+          assert.ok(Number(cache.db.prepare("SELECT expires FROM fills WHERE key=?").get("fill-heartbeat").expires) >= base + 152000);
+          Date.now = now; input.push("completed after contention"); input.push(null);
+          assert.equal((await checked).toString(), "completed after contention");
+        } else {
+          if (outcome === "replaced") other.db.prepare("UPDATE fills SET token=? WHERE key=?").run("replacement", "fill-heartbeat");
+          Date.now = () => base + (outcome === "expire" ? 121000 : 32000); tick();
+          assert.equal(input.destroyed, true);
+          Date.now = now; await checked;
+          assert.equal(cache.stats().blobs, 0);
+        }
+      } finally {
+        global.setInterval = interval; Date.now = now;
+        if (other.db.isTransaction) other.db.exec("ROLLBACK");
+        input.destroy(); await checked.catch(() => {}); other.close();
+      }
+    });
+  }
+  it("coordinates physical headroom between source and result pools", async () => {
+    const native = require("node:fs"), statfs = native.statfsSync;
+    const other = new BlobCache({ ...cache.options, root: join(root, "results"), reservationRoot: root, freeBytes: 1000 });
+    cache.options.freeBytes = 1000;
+    native.statfsSync = () => ({ bavail: 18000, bsize: 1 });
+    let release, started;
+    const pending = new Promise(resolve => { started = resolve; });
+    const fill = cache.getOrFill("source-pending", async () => {
+      started(); await new Promise(resolve => { release = resolve; }); return Readable.from("source");
+    });
+    try {
+      await pending;
+      await assert.rejects(other.getOrFill("result-pending", async () => Readable.from("result")),
+        error => error instanceof CacheAdmissionError && error.message === "disk_headroom");
+      release(); await read(await fill);
+      assert.equal((await read(await other.getOrFill("result-pending", async () => Readable.from("result")))).toString(), "result");
+      assert.equal(cache.disk.prepare("SELECT COUNT(*) n FROM reservations").get().n, 0);
+    } finally { release?.(); await fill.catch(() => {}); native.statfsSync = statfs; other.close(); }
+  });
+  it("shares physical headroom reservations across separate producer processes", async function () {
+    this.timeout(45000);
+    const { fork } = require("node:child_process");
+    const workers = ["source-process", "result-process"].map(name => fork(join(__dirname, "fixtures/shared-cache-worker.js"), [], {
+      env: { ...process.env, NODE_ENV: "test", BLOB_TEST_DISK_BYTES: "18000", BLOB_TEST_HOLD_FILL: "1",
+        BLOB_TEST_OPTIONS: JSON.stringify({ ...cache.options, root: join(root, name), reservationRoot: root, freeBytes: 1000 }) },
+      stdio: ["ignore", "ignore", "pipe", "ipc"]
+    }));
+    try {
+      const ready = [], outcomes = [], bodies = [];
+      for (const worker of workers) {
+        let stderr = ""; worker.stderr.on("data", b => { stderr = (stderr + b).slice(-4000); });
+        ready.push(new Promise((resolve, reject) => {
+          worker.on("message", m => { if (m === "ready") resolve(); });
+          worker.on("error", reject); worker.on("exit", code => reject(Error("worker exited " + code + ": " + stderr)));
+        }));
+        outcomes.push(new Promise((resolve, reject) => {
+          worker.on("message", m => { if (m === "fill" || m.error) resolve(m); }); worker.on("error", reject);
+        }));
+        bodies.push(new Promise(resolve => worker.on("message", m => { if (m.body || m.error) resolve(m); })));
+      }
+      await Promise.all(ready); workers.forEach(w => w.send("start"));
+      const values = await Promise.all(outcomes);
+      assert.equal(values.filter(v => v === "fill").length, 1);
+      assert.equal(values.find(v => v.error)?.error, "disk_headroom");
+      workers[values.indexOf("fill")].send("release");
+      const completed = await Promise.all(bodies);
+      assert.equal(completed.find(v => v.body)?.body, "shared across processes");
+      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);
+    assert.deepEqual(await fs.readdir(join(root, "tmp")), []);
+    await assert.rejects(cache.getOrFill("large", async () => Readable.from(Buffer.alloc(5000))), CacheAdmissionError);
+    assert.equal(cache.stats().reservedBytes, 0);
+  });
+  it("evicts least-recently-read data while retaining active readers and respecting the budget", async () => {
+    const data = randomBytes(4000);
+    for (let i = 0; i < 12; i++) { const b = Buffer.concat([Buffer.from([i]), data]); await read(await cache.getOrFill("file" + i, async () => Readable.from(b))); }
+    const pinned = await cache.lookup("file0"); assert.ok(pinned);
+    const pinnedDigest = pinned.cachedBlob.record.digest;
+    cache.db.prepare("UPDATE blobs SET accessed=1 WHERE digest<>?").run(pinnedDigest);
+    for (let i = 12; i < 20; i++) { const b = Buffer.concat([Buffer.from([i]), data]); await read(await cache.getOrFill("file" + i, async () => Readable.from(b))); }
+    assert.ok(cache.stats().storedBytes + cache.stats().reservedBytes <= cache.options.budget);
+    assert.ok(cache.db.prepare("SELECT 1 FROM blobs WHERE digest=?").get(pinnedDigest));
+    assert.equal((await read(pinned)).length, 4001);
+    assert.equal(await cache.lookup("file1"), undefined);
+    assert.deepEqual(await read(await cache.getOrFill("file1", async () => Readable.from("refetched"))), Buffer.from("refetched"));
+  });
+  it("recovers an expired producer and removes its temporary files before admitting new work", async () => {
+    cache.db.prepare("INSERT INTO fills VALUES(?,?,?,?)").run("dead", "dead", 1, 60000);
+    await fs.writeFile(join(root, "tmp", "dead.raw"), "partial");
+    const orphanDigest = "sha256-" + "a".repeat(64);
+    const orphanDir = join(root, "blobs", "aa", orphanDigest);
+    await fs.mkdir(orphanDir, { recursive: true }); await fs.writeFile(join(orphanDir, "dead.raw"), "unindexed");
+    cache.db.prepare("INSERT INTO publications VALUES(?,?,?)").run("dead", orphanDigest, "dead.raw");
+    await read(await cache.getOrFill("new", async () => Readable.from("new")));
+    assert.equal(cache.stats().reservedBytes, 0); assert.deepEqual(await fs.readdir(join(root, "tmp")), []);
+    assert.deepEqual(await fs.readdir(orphanDir), []);
+  });
+  it("invalidates a missing or truncated file and refetches it", async () => {
+    const stream = await cache.getOrFill("missing", async () => Readable.from("hello"));
+    const record = stream.cachedBlob.record; await read(stream);
+    const p = join(root, "blobs", record.digest.split("-")[1].slice(0, 2), record.digest, record.file);
+    await fs.writeFile(p, "x"); assert.equal(await cache.lookup("missing"), undefined);
+    assert.deepEqual(await read(await cache.getOrFill("missing", async () => Readable.from("hello"))), Buffer.from("hello"));
+  });
+  it("bounds decision retention and expires old transformation settings", () => {
+    cache.saveDecision("rule-v1", "git-" + "a".repeat(40), false);
+    assert.equal(cache.decision("rule-v1").changed, false); assert.equal(cache.decision("rule-v2"), undefined);
+    cache.db.prepare("UPDATE decisions SET expires=1").run(); assert.equal(cache.decision("rule-v1"), undefined);
+  });
+});

test/shared-cache-http.test.js

@@ -0,0 +1,267 @@
+const { strict: assert } = require("node:assert");
+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");
+const express = require("express");
+require("ts-node/register/transpile-only");
+const config = require("../src/config").default;
+const { sharedBlobCache, resultBlobCache, closeBlobCaches } = require("../src/core/blob-cache");
+const { sendSharedContent } = require("../src/core/cached-response");
+const { closeCacheMetrics, cacheMetricsReport, observeCache } = require("../src/core/cache-observability");
+const { monitorRequests } = require("../src/core/request-monitoring");
+const GitHubStream = require("../src/core/source/GitHubStream").default;
+const sha = b => createHash("sha1").update(`blob ${b.length}\0`).update(b).digest("hex");
+const consume = async s => { const chunks = []; for await (const b of s) chunks.push(Buffer.from(b)); return Buffer.concat(chunks); };
+function request(port, path, encoding = "gzip", extra = {}) {
+  return new Promise((resolve, reject) => {
+    const req = http.get({ host: "127.0.0.1", port, path, headers: { "accept-encoding": encoding, ...extra } }, res => {
+      consume(res).then(body => resolve({ status: res.statusCode, headers: res.headers, body }), reject);
+    }); req.on("error", reject);
+  });
+}
+describe("shared cache HTTP and telemetry integration", function () {
+  this.timeout(15000);
+  let root, old, server, port, fills, body;
+  beforeEach(async () => {
+    old = { ...config }; closeBlobCaches(); closeCacheMetrics();
+    root = await fs.mkdtemp(join(tmpdir(), "shared-http-"));
+    Object.assign(config, { FOLDER: root, SHARED_CACHE_PATH: join(root, "cache"), SHARED_CACHE_ENABLED: true,
+      SHARED_CACHE_BYTES: 16 * 1024 ** 2, SHARED_CACHE_RESULT_BYTES: 8 * 1024 ** 2, SHARED_CACHE_MAX_FILE_BYTES: 1024 ** 2,
+      SHARED_CACHE_FREE_BYTES: 0, CACHE_MONITORING_ENABLED: true });
+    fills = 0; body = Buffer.from("Public documentation with no identifying names.\n".repeat(100));
+    const app = express(); app.use(monitorRequests("api", undefined, { info() {}, warn() {} }));
+    app.get("/:project/:mode", async (req, res) => {
+      if (req.headers.authorization !== "test-access") return res.status(403).end();
+      const source = new GitHubStream({ repoId: req.params.project, organization: "org", repoName: "repo", commit: "commit", getToken: async () => "" });
+      source.downloadWithFallback = async () => { fills++; return Readable.from(body); };
+      source.downloadFile = () => Readable.from("invalid pointer");
+      const filePath = String(req.query.name || "README.md");
+      const mime = require("mime-types").lookup(filePath);
+      if (mime) res.type(mime);
+      try {
+        const input = await source.getFileContentCache(filePath, req.params.project, () => ({ sha: sha(body), size: body.length }));
+        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(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;
+  });
+  afterEach(async () => {
+    await new Promise(resolve => server.close(resolve)); closeBlobCaches(); closeCacheMetrics(); Object.assign(config, old);
+    await fs.rm(root, { recursive: true, force: true });
+  });
+  const headers = { authorization: "test-access" };
+  it("shares source content across projects without bypassing authorization", async () => {
+    assert.equal((await request(port, "/a/plain")).status, 403); assert.equal(fills, 0);
+    const a = await request(port, "/a/plain", "gzip", headers);
+    assert.equal(a.status, 200); assert.equal(a.headers["content-encoding"], "gzip"); assert.deepEqual(gunzipSync(a.body), body);
+    const b = await request(port, "/b/plain", "identity", headers);
+    assert.deepEqual(b.body, body); assert.equal(b.headers["content-encoding"], undefined); assert.equal(fills, 1);
+    assert.equal(sharedBlobCache().stats().blobs, 1); assert.equal(resultBlobCache().stats().blobs, 0);
+    assert.equal((await request(port, "/b/plain", "gzip")).status, 403);
+  });
+  it("fetches public submodule content from its origin without forwarding parent credentials", async () => {
+    const original = GitHubStream.prototype.downloadWithFallback;
+    const upstream = { organization: "public-org", repoName: "dependency", commit: "a".repeat(40), path: "src/data.txt" };
+    let calls = 0;
+    GitHubStream.prototype.downloadWithFallback = async function (token, digest, path) {
+      calls++;
+      assert.equal(token, ""); assert.equal(digest, sha(body)); assert.equal(path, upstream.path);
+      assert.equal(this.data.organization, upstream.organization); assert.equal(this.data.repoName, upstream.repoName);
+      assert.equal(this.data.commit, upstream.commit);
+      return Readable.from(body);
+    };
+    try {
+      const source = new GitHubStream({ repoId: "parent", organization: "private-org", repoName: "parent",
+        commit: "parent-commit", getToken: async () => "parent-secret" });
+      const meta = () => ({ sha: sha(body), size: body.length, upstream });
+      assert.deepEqual(await consume(await source.getFileContentCache("vendor/src/data.txt", "parent", meta)), body);
+      assert.deepEqual(await consume(await source.getFileContentCache("vendor/src/data.txt", "parent", meta)), body);
+      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);
+      assert.equal(r.status, 200); assert.match(r.headers["content-type"], /^text\/plain/);
+      const decoded = r.headers["content-encoding"] === "gzip" ? gunzipSync(r.body) : r.body;
+      assert.equal(decoded.toString(), changed ? body.toString().replaceAll("Public", "Private") : body.toString());
+    }
+  });
+  for (const name of ["video.mp4", "audio.mp3", "binary.unknownext"]) it("omits the range-denial header for binary " + name, async () => {
+    body = Buffer.from([0, 1, 2, 0, 255]);
+    for (let i = 0; i < 2; i++) {
+      const r = await request(port, "/a/plain?name=" + name, "identity", headers);
+      assert.equal(r.status, 200); assert.equal(r.headers["accept-ranges"], undefined);
+      assert.notEqual(r.headers["content-type"], "text/plain; charset=utf-8"); assert.deepEqual(r.body, body);
+    }
+  });
+  it("bypasses a contended source index without failing a download", async () => {
+    await request(port, "/a/plain", "identity", headers);
+    const { BlobCache } = require("../src/core/blob-cache");
+    const other = new BlobCache(sharedBlobCache().options); other.db.exec("BEGIN IMMEDIATE");
+    try {
+      const r = await request(port, "/a/plain", "identity", headers);
+      assert.equal(r.status, 200); assert.deepEqual(r.body, body); assert.equal(fills, 2);
+    } finally { other.db.exec("ROLLBACK"); other.close(); }
+  });
+  it("bypasses a contended result index while retaining the source fallback", async () => {
+    await request(port, "/a/plain", "identity", headers);
+    const { BlobCache } = require("../src/core/blob-cache");
+    const other = new BlobCache(resultBlobCache().options); other.db.exec("BEGIN IMMEDIATE");
+    try {
+      const r = await request(port, "/a/plain", "identity", headers);
+      assert.equal(r.status, 200); assert.deepEqual(r.body, body); assert.equal(fills, 1);
+    } finally { other.db.exec("ROLLBACK"); other.close(); }
+  });
+  for (const bypass of [false, true]) for (const streamed of [false, true]) {
+    it("normalizes upstream permission errors for " + JSON.stringify({ bypass, streamed }), async () => {
+      if (bypass) sharedBlobCache().options.budget = 1;
+      const source = new GitHubStream({ repoId: "errors", organization: "org", repoName: "repo", commit: "commit", getToken: async () => "" });
+      source.downloadWithFallback = async () => {
+        const error = Object.assign(Error("forbidden"), { response: { statusCode: 403 } });
+        if (!streamed) throw error;
+        return Readable.from((async function* () { yield Buffer.alloc(0); throw error; })());
+      };
+      await assert.rejects(async () => consume(await source.getFileContentCache("data.txt", "errors", () => ({ sha: sha(body), size: body.length }))),
+        error => error.message === "file_not_accessible" && error.httpStatus === 403);
+    });
+  }
+  it("rejects truncated admission-bypass downloads before successful completion", async () => {
+    sharedBlobCache().options.budget = 1;
+    const source = new GitHubStream({ repoId: "short", organization: "org", repoName: "repo", commit: "commit", getToken: async () => "" });
+    source.downloadWithFallback = async () => Readable.from("short");
+    await assert.rejects(async () => consume(await source.getFileContentCache("data.txt", "short", () => ({ sha: sha(body), size: 20 }))),
+      error => error.message === "storage_write_size_mismatch" && error.httpStatus === 502);
+  });
+  it("classifies parent repository misses before returning a shared-download error", async () => {
+    const misses = require("../src/core/source/github-content-miss"), original = misses.classifyContentMiss;
+    let classified = 0;
+    misses.classifyContentMiss = async () => { classified++; return "repo_not_found"; };
+    try {
+      const source = new GitHubStream({ repoId: "missing-parent", organization: "org", repoName: "repo", commit: "missing-parent", getToken: async () => "" });
+      source.downloadWithFallback = async () => { throw Object.assign(Error("missing"), { status: 404 }); };
+      await assert.rejects(source.getFileContentCache("data.txt", "missing-parent", () => ({ sha: sha(body), size: body.length })),
+        error => error.message === "repo_not_found" && error.httpStatus === 404);
+      assert.equal(classified, 1);
+    } finally { misses.classifyContentMiss = original; }
+  });
+  it("classifies unavailable submodule streams on cache bypass", async () => {
+    sharedBlobCache().options.budget = 1;
+    const original = GitHubStream.prototype.downloadWithFallback;
+    GitHubStream.prototype.downloadWithFallback = async () => Readable.from((async function* () {
+      yield Buffer.alloc(0);
+      throw Object.assign(Error("missing"), { response: { statusCode: 404 } });
+    })());
+    try {
+      const source = new GitHubStream({ repoId: "errors", organization: "org", repoName: "repo", commit: "commit", getToken: async () => "" });
+      await assert.rejects(async () => consume(await source.getFileContentCache("vendor/data.txt", "errors", () => ({ sha: sha(body), size: body.length,
+        upstream: { organization: "public", repoName: "dependency", commit: "a".repeat(40), path: "data.txt" } }))),
+      error => error.message === "submodule_not_available" && error.httpStatus === 404);
+    } finally { GitHubStream.prototype.downloadWithFallback = original; }
+  });
+  it("reuses unchanged decisions, separates changed rules, and caches small changed output", async () => {
+    const first = await request(port, "/a/plain", "gzip", headers);
+    const second = await request(port, "/a/plain", "gzip", headers);
+    assert.deepEqual(second.body, first.body);
+    const changed = await request(port, "/a/changed", "gzip", headers);
+    assert.equal(gunzipSync(changed.body).toString(), body.toString().replaceAll("Public", "Private"));
+    const repeated = await request(port, "/a/changed", "identity", headers);
+    assert.equal(repeated.body.toString(), body.toString().replaceAll("Public", "Private"));
+    assert.equal(resultBlobCache().stats().blobs, 1);
+    const report = cacheMetricsReport(60); assert.equal(report.available, true);
+    const count = name => report.rows.filter(r => r.name === name).reduce((n, r) => n + r.count, 0);
+    assert.equal(count("rewrite_unchanged"), 1); assert.equal(count("rewrite_changed"), 1); assert.equal(count("decision_hit"), 2);
+  });
+  it("respects gzip q=0, rejects all forbidden encodings, and gzip-encodes small identity blobs when required", async () => {
+    let r = await request(port, "/a/plain", "gzip;q=0, identity;q=1", headers);
+    assert.equal(r.headers["content-encoding"], undefined); assert.deepEqual(r.body, body);
+    r = await request(port, "/a/plain", "gzip;q=0, identity;q=0", headers); assert.equal(r.status, 406);
+    body = Buffer.from("tiny text");
+    r = await request(port, "/a/plain", "gzip, identity;q=0", headers);
+    assert.equal(r.headers["content-encoding"], "gzip"); assert.deepEqual(gunzipSync(r.body), body);
+    assert.ok(r.headers.vary.includes("Accept-Encoding"));
+  });
+  it("verifies the Git LFS pointer and resolved payload before sharing the payload", async () => {
+    const payload = Buffer.from("large file payload");
+    const pointer = Buffer.from(`version https://git-lfs.github.com/spec/v1\noid sha256:${createHash("sha256").update(payload).digest("hex")}\nsize ${payload.length}\n`);
+    const source = new GitHubStream({ repoId: "lfs", organization: "org", repoName: "repo", commit: "commit", getToken: async () => "" });
+    source.downloadWithFallback = async () => Readable.from(payload);
+    source.downloadFile = () => Readable.from(pointer);
+    assert.deepEqual(await consume(await source.getFileContentCache("data.bin", "lfs", () => ({ sha: sha(pointer), size: pointer.length }))), payload);
+    const badPointer = Buffer.from(pointer.toString().replace(`size ${payload.length}`, "size 999"));
+    source.downloadFile = () => Readable.from(badPointer);
+    await assert.rejects(source.getFileContentCache("bad.bin", "lfs", () => ({ sha: sha(badPointer), size: badPointer.length })), /invalid_blob_hash/);
+    assert.equal(sharedBlobCache().stats().blobs, 1);
+  });
+  it("uses bounded persistent metrics with separate monitoring traffic and no file identifiers", async () => {
+    await request(port, "/a/plain", "identity", { ...headers, "user-agent": "Uptime-Kuma/1" });
+    observeCache("compress", 100, 1, "zip");
+    const report = cacheMetricsReport(60);
+    assert.ok(report.rows.some(r => r.traffic === "monitor")); assert.ok(report.rows.some(r => r.traffic === "zip"));
+    assert.ok(!JSON.stringify(report).includes("README")); assert.ok(!JSON.stringify(report).includes(sha(body)));
+    config.CACHE_MONITORING_ENABLED = false; assert.equal(cacheMetricsReport().available, false);
+  });
+});

test/vue-ui.test.js

@@ -976,6 +976,37 @@ describe("Vue 3 UI", function () {
     expect(ui.errors).to.deep.equal([]);
   });
 
+  it("shows cache budgets and filters monitoring traffic without confusing unavailable data with zero", async function () {
+    ui = await browser("/", { "/api/user": { username: "admin", isAdmin: true },
+      "/api/admin/overview": { history: [] },
+      "/api/admin/cache": { available: true, enabled: true, filesystemSupported: true, sampledAt: Date.now(),
+        source: { allocatedBytes: 4096, budgetBytes: 65536, reservedBytes: 0, originalBytes: 19000, blobs: 1, activeReaders: 0 },
+        latencyBucketsMs: [0.1, 1, 10, 100, 1000, 10000],
+        rows: [{ name: "source_hit", traffic: "file", count: 4, bytes: 400, ms: 0 },
+          { name: "source_miss", traffic: "file", count: 1, bytes: 100, ms: 0 },
+          { name: "source_hit", traffic: "monitor", count: 10, bytes: 1000, ms: 0 }] } });
+    await ui.go("/admin/");
+    const panel = ui.window.document.querySelector(".ov-cache");
+    expect(panel.textContent).to.include("4.0 KB / 64.0 KB");
+    await ui.input("#cache-traffic", "file", "change");
+    expect(panel.textContent).to.include("80.0%");
+    await ui.input("#cache-traffic", "zip", "change");
+    expect(panel.textContent).to.include("No cache events");
+    await ui.input("#cache-window", "1440", "change");
+    expect(ui.requests.filter(r => r.url.pathname === "/api/admin/cache").at(-1).url.searchParams.get("minutes")).to.equal("1440");
+    expect(ui.errors).to.deep.equal([]);
+  });
+
+  it("shows unavailable cache telemetry with a retry action", async function () {
+    ui = await browser("/", { "/api/user": { username: "admin", isAdmin: true },
+      "/api/admin/overview": { history: [] }, "/api/admin/cache": { available: false, reason: "Cache index unavailable" } });
+    await ui.go("/admin/");
+    const panel = ui.window.document.querySelector(".ov-cache");
+    expect(panel.querySelector('[role="alert"]').textContent).to.include("Cache index unavailable");
+    expect(panel.querySelector("button").textContent).to.equal("Retry");
+    expect(panel.textContent).not.to.include("Source request hit rate"); expect(ui.errors).to.deep.equal([]);
+  });
+
   it("shows latency bounds, interruptions and resource samples on the admin overview", async function () {
     const sample = { instance: "streamer:fixture", service: "streamer", sampledAt: Date.now(), cpuPercent: 2,
       memory: { rss: 16 * 1024 * 1024 }, memoryLimitBytes: 128 * 1024 * 1024,