fix: coordinate cache disk reservations and tolerate reader renewal contention

docs/shared-cache.md

@@ -31,6 +31,11 @@ Restart processes together when changing pool limits. Reserve headroom for SQLit
 filesystem directories, existing caches and other application temporary files.
 The budget accounts for blob allocation with a conservative 4 KiB minimum plus
 in-flight reservations of three times the maximum object size and a small overhead.
+A shared SQLite ledger reserves physical headroom across source and result pools
+before producers write bytes. Its check uses a fresh free-space sample under a
+cross-process lock and includes all outstanding reservations. Reservations remain
+until cleanup or crash recovery completes. This is conservative while fills have
+already written some bytes, and cannot reserve space against unrelated applications.
 Keep each pool budget larger than this reservation. Admission can bypass the cache
 while other fills/readers occupy the budget. Oversized source objects are streamed
 without admission, still subject to the application's file-size limit.
@@ -53,6 +58,9 @@ other durable source material are untouched. Repository removal invalidates acce
 and project mappings, while shared blobs remain subject to global eviction.
 
 Transient SQLite write contention bypasses caching instead of failing file requests.
+Reader heartbeats retry contention within their last confirmed lease window and
+terminate only if renewal cannot succeed before expiry or the lease is removed.
+Admission-bypass streams retain the minimum expected-size check.
 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.

src/core/blob-cache.ts

@@ -1,6 +1,6 @@
 import { DatabaseSync } from "node:sqlite";
 import { createHash, randomUUID } from "crypto";
-import { createReadStream, createWriteStream, mkdirSync, unlinkSync, rmdirSync, promises as fs } from "fs";
+import { createReadStream, createWriteStream, mkdirSync, unlinkSync, rmdirSync, statfsSync, promises as fs } from "fs";
 import { join } from "path";
 import { Readable, Transform } from "stream";
 import { pipeline } from "stream/promises";
@@ -20,7 +20,7 @@ function cacheError(error: unknown): never {
 }
 export interface BlobRecord { digest: string; file: string; size: number; stored: number; allocated: number; encoding: "gzip" | "identity"; accessed: number }
 export interface BlobProof { gitSha: string; sha256: string; size: number }
-export interface BlobOptions { root: string; budget: number; maxObject: number; gzip?: boolean; saving?: number; freeBytes?: number; lowWater?: number; name?: string }
+export interface BlobOptions { root: string; budget: number; maxObject: number; gzip?: boolean; saving?: number; freeBytes?: number; lowWater?: number; name?: string; reservationRoot?: string }
 export type BlobStream = Readable & { cachedBlob?: { store: BlobCache; record: BlobRecord } };
 const validDigest = (s: string) => /^(git-[a-f0-9]{40}|sha256-[a-f0-9]{64})$/.test(s);
 
@@ -28,6 +28,7 @@ const validDigest = (s: string) => /^(git-[a-f0-9]{40}|sha256-[a-f0-9]{64})$/.te
 export class BlobCache {
   readonly db: DatabaseSync;
   readonly name: string;
+  private disk?: DatabaseSync;
   constructor(readonly options: BlobOptions) {
     if (!(options.budget > 0 && options.maxObject > 0)) throw new Error("invalid_cache_budget");
     this.name = options.name || "source";
@@ -80,8 +81,14 @@ export class BlobCache {
             BEGIN UPDATE summary SET ${values.map(([column, value]) => `${column}=${column}-OLD.${value}+NEW.${value}`).join(",")} WHERE id=1; END;`);
         }
       });
+      const reservationRoot = options.reservationRoot || options.root;
+      mkdirSync(reservationRoot, { recursive: true, mode: 0o700 });
+      this.disk = new DatabaseSync(join(reservationRoot, "disk-reservations.sqlite"));
+      this.disk.exec(`PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
+        CREATE TABLE IF NOT EXISTS reservations(token TEXT PRIMARY KEY,bytes INTEGER);
+        PRAGMA busy_timeout=50;`);
       this.db.exec("PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=50;");
-    } catch (error) { this.db.close(); throw error; }
+    } catch (error) { this.disk?.close(); this.db.close(); throw error; }
   }
   // Reader leases, history and decisions are disposable after a system failure.
   // Reservations and publication records use FULL before creating/renaming payloads,
@@ -98,6 +105,20 @@ export class BlobCache {
     const excess = Number(this.db.prepare(`SELECT ${table} n FROM summary WHERE id=1`).get()!.n) - limit;
     if (excess > 0) this.db.prepare(`DELETE FROM ${table} WHERE rowid IN (SELECT rowid FROM ${table} ORDER BY ${order} LIMIT ?)`).run(excess);
   }
+  private reserveDisk(token: string, bytes: number) {
+    const disk = this.disk!;
+    disk.exec("BEGIN IMMEDIATE");
+    try {
+      // Take a fresh filesystem sample while holding the shared admission lock.
+      const space = statfsSync(this.options.root);
+      const pending = Number(disk.prepare("SELECT COALESCE(SUM(bytes),0) bytes FROM reservations").get()!.bytes);
+      if (space.bavail * space.bsize < pending + bytes + (this.options.freeBytes || 0))
+        throw new CacheAdmissionError("disk_headroom");
+      disk.prepare("INSERT INTO reservations VALUES(?,?)").run(token, bytes);
+      disk.exec("COMMIT");
+    } catch (error) { disk.exec("ROLLBACK"); throw error; }
+  }
+  private releaseDisk(token: string) { this.disk!.prepare("DELETE FROM reservations WHERE token=?").run(token); }
   private metric(event: string, bytes = 0, ms = 0) { observeCache(`${this.name}_${event}`, bytes, ms); }
   private path(record: BlobRecord) {
     if (!validDigest(record.digest) || !/^[a-f0-9-]+\.(gz|raw)$/.test(record.file)) throw new Error("invalid_blob_record");
@@ -116,7 +137,11 @@ export class BlobCache {
         catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }
       }
       this.db.prepare("DELETE FROM publications WHERE token=?").run(row.token);
-      for (const suffix of ["raw", "gz"]) { try { unlinkSync(join(this.options.root, "tmp", `${row.token}.${suffix}`)); } catch { /* absent */ } }
+      for (const suffix of ["raw", "gz"]) {
+        try { unlinkSync(join(this.options.root, "tmp", `${row.token}.${suffix}`)); }
+        catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }
+      }
+      this.releaseDisk(String(row.token));
     }
     this.db.prepare("DELETE FROM fills WHERE expires<?").run(now);
     this.db.prepare("DELETE FROM readers WHERE expires<?").run(now);
@@ -170,10 +195,11 @@ export class BlobCache {
   private async openReader(digest: string, encoded = false): Promise<BlobStream | undefined> {
     if (!validDigest(digest)) return;
     const token = randomUUID();
+    let leaseUntil = Date.now() + 120000;
     const record = this.transaction(() => {
       const r = this.db.prepare("SELECT * FROM blobs WHERE digest=?").get(digest) as unknown as BlobRecord | undefined;
       if (!r) return;
-      this.db.prepare("INSERT INTO readers VALUES(?,?,?)").run(token, digest, Date.now() + 120000);
+      this.db.prepare("INSERT INTO readers VALUES(?,?,?)").run(token, digest, leaseUntil);
       if (Date.now() - r.accessed > 30000) this.db.prepare("UPDATE blobs SET accessed=? WHERE digest=?").run(Date.now(), digest);
       return r;
     });
@@ -194,10 +220,22 @@ export class BlobCache {
     const output = (needsDecode ? createGunzip() : source) as BlobStream;
     output.cachedBlob = { store: this, record };
     let start = performance.now();
+    let renewAt = Date.now() + 30000;
     const heartbeat = setInterval(() => {
-      try { if (!this.db.prepare("UPDATE readers SET expires=? WHERE token=?").run(Date.now() + 120000, token).changes) output.destroy(new Error("cache_reader_lease_lost")); }
-      catch { output.destroy(new Error("cache_reader_lease_lost")); }
-    }, 30000); heartbeat.unref();
+      const now = Date.now();
+      if (now < renewAt) return;
+      if (now >= leaseUntil) { output.destroy(new Error("cache_reader_lease_lost")); return; }
+      try {
+        const renewedUntil = now + 120000;
+        if (!this.db.prepare("UPDATE readers SET expires=? WHERE token=? AND expires>?").run(renewedUntil, token, now).changes) {
+          output.destroy(new Error("cache_reader_lease_lost")); return;
+        }
+        leaseUntil = renewedUntil; renewAt = now + 30000;
+      } catch (error) {
+        if (!isCacheContention(error)) output.destroy(new Error("cache_reader_lease_lost"));
+        // Retry on the next tick while the previously confirmed lease is valid.
+      }
+    }, 1000); heartbeat.unref();
     let released = false;
     const release = () => {
       if (released) return; released = true; clearInterval(heartbeat); source.destroy();
@@ -242,8 +280,6 @@ export class BlobCache {
     const token = randomUUID(), reservation = this.options.maxObject * 3 + 1024;
     const deadline = Date.now() + 120000;
     for (;;) {
-      const space = await fs.statfs(this.options.root);
-      if (space.bavail * space.bsize < reservation + (this.options.freeBytes || 0)) throw new CacheAdmissionError("disk_headroom");
       const owned = this.transaction(() => {
         this.expire();
         if (this.db.prepare("SELECT 1 FROM fills WHERE key=?").get(key)) return false;
@@ -268,6 +304,11 @@ export class BlobCache {
     }, 30000); heartbeat.unref();
     const timeout = setTimeout(() => input?.destroy(new Error("cache_fill_timeout")), 120000); timeout.unref();
     try {
+      // The pool reservation is already durable before acquiring physical headroom.
+      this.transaction(() => {
+        if (!this.db.prepare("SELECT 1 FROM fills WHERE token=? AND expires>?").get(token, Date.now())) throw new CacheAdmissionError("cache_fill_lease_lost");
+        this.reserveDisk(token, reservation);
+      });
       // Recheck after ownership; another producer may have published between the initial lookup and lease acquisition.
       hit = await this.lookup(key); if (hit) return hit;
       input = await produce();
@@ -312,13 +353,16 @@ export class BlobCache {
       const result = await this.open(digest); if (!result) throw new Error("published_blob_missing"); return result;
     } finally {
       clearTimeout(timeout); clearInterval(heartbeat); input?.destroy();
-      await Promise.all([raw, gz, ...(file ? [file] : [])].map(p => fs.unlink(p).catch(() => {})));
+      await Promise.all([raw, gz, ...(file ? [file] : [])].map(p => fs.unlink(p).catch(error => {
+        if (error.code !== "ENOENT") throw error;
+      })));
       this.releaseFill(key, token);
     }
   }
   private releaseFill(key: string, token: string) {
     try {
       this.transaction(() => {
+        this.releaseDisk(token);
         this.db.prepare("DELETE FROM readers WHERE token=?").run(token);
         this.db.prepare("DELETE FROM publications WHERE token=?").run(token);
         this.db.prepare("DELETE FROM fills WHERE key=? AND token=?").run(key, token);
@@ -341,19 +385,19 @@ export class BlobCache {
       });
     } catch (error) { if (!isCacheContention(error)) throw error; /* Optional decision can be recomputed. */ }
   }
-  close() { this.db.close(); }
+  close() { this.disk?.close(); this.db.close(); }
 }
 let source: BlobCache | undefined, results: BlobCache | undefined;
 export function sharedBlobCache() {
   if (!config.SHARED_CACHE_ENABLED || config.STORAGE !== "filesystem") return;
-  try { return source ||= new BlobCache({ root: join(config.SHARED_CACHE_PATH, "source"), budget: config.SHARED_CACHE_BYTES,
+  try { return source ||= new BlobCache({ reservationRoot: config.SHARED_CACHE_PATH, root: join(config.SHARED_CACHE_PATH, "source"), budget: config.SHARED_CACHE_BYTES,
     maxObject: config.SHARED_CACHE_MAX_FILE_BYTES, gzip: config.SHARED_CACHE_GZIP, saving: config.SHARED_CACHE_GZIP_MIN_SAVING,
     freeBytes: config.SHARED_CACHE_FREE_BYTES, lowWater: config.SHARED_CACHE_LOW_WATER });
   } catch (error) { if (!isCacheContention(error)) throw error; }
 }
 export function resultBlobCache() {
   if (!sharedBlobCache() || !config.SHARED_CACHE_DECISIONS) return;
-  try { return results ||= new BlobCache({ root: join(config.SHARED_CACHE_PATH, "results"), budget: config.SHARED_CACHE_RESULT_BYTES,
+  try { return results ||= new BlobCache({ reservationRoot: config.SHARED_CACHE_PATH, root: join(config.SHARED_CACHE_PATH, "results"), budget: config.SHARED_CACHE_RESULT_BYTES,
     maxObject: Math.min(config.SHARED_CACHE_MAX_FILE_BYTES, Math.floor(config.SHARED_CACHE_RESULT_BYTES / 8)), gzip: config.SHARED_CACHE_GZIP,
     freeBytes: config.SHARED_CACHE_FREE_BYTES, name: "result" });
   } catch (error) { if (!isCacheContention(error)) throw error; }

src/core/source/GitHubStream.ts

@@ -391,7 +391,7 @@ export default class GitHubStream extends GitHubBase {
       await assertActive();
       if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });
       let legacy = false, allowLegacy = true;
-      const download = async () => {
+      const download = async (checkMinimum = false) => {
         const parentToken = await this.data.getToken();
         const token = expected.upstream ? "" : parentToken;
         const sourceMiss = cachedContentMiss(this.data, parentToken);
@@ -425,6 +425,8 @@ export default class GitHubStream extends GitHubBase {
               if (bytes > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });
               yield chunk;
             }
+            if (checkMinimum && expected.size != null && expected.size > 0 && bytes < expected.size)
+              throw new AnonymousError("storage_write_size_mismatch", { httpStatus: 502 });
           } catch (error) { throw await normalize(error as Error); }
           finally { clearTimeout(deadline); input.destroy(); }
         })());
@@ -464,7 +466,7 @@ export default class GitHubStream extends GitHubBase {
         return content;
       } catch (error) {
         if (!(error instanceof CacheAdmissionError)) throw error;
-        observeCache("source_bypass"); await assertActive(); return download();
+        observeCache("source_bypass"); await assertActive(); return download(true);
       }
     }
     const hit = await measureStage("cache_lookup", cached);

test/fixtures/shared-cache-worker.js

@@ -12,12 +12,14 @@ if (process.env.BLOB_TEST_CRASH_AFTER_RENAME) {
     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");
-      await new Promise(resolve => setTimeout(resolve, 100));
+      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;

test/shared-blob-cache.test.js

@@ -143,6 +143,7 @@ describe("shared content-addressed blob pool", function () {
       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);
@@ -150,6 +151,7 @@ describe("shared content-addressed blob pool", function () {
       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(); }
@@ -182,6 +184,84 @@ describe("shared content-addressed blob pool", function () {
     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(); }
+  });
+  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("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

@@ -132,6 +132,13 @@ describe("shared cache HTTP and telemetry integration", function () {
         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;