diff --git a/src/core/blob-cache.ts b/src/core/blob-cache.ts index dfcb776..1ebdb46 100644 --- a/src/core/blob-cache.ts +++ b/src/core/blob-cache.ts @@ -279,6 +279,7 @@ export class BlobCache { if (sizeHint != null && sizeHint > this.options.maxObject) { this.metric("bypass", sizeHint); throw new CacheAdmissionError("object_too_large"); } const token = randomUUID(), reservation = this.options.maxObject * 3 + 1024; const deadline = Date.now() + 120000; + let leaseUntil = 0; for (;;) { const owned = this.transaction(() => { this.expire(); @@ -289,7 +290,8 @@ export class BlobCache { } // 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; + leaseUntil = Date.now() + 120000; + this.db.prepare("INSERT INTO fills VALUES(?,?,?,?)").run(key, token, leaseUntil, reservation); return true; }, true); if (owned === null) throw new CacheAdmissionError("cache_budget_busy"); if (owned) break; @@ -300,10 +302,23 @@ export class BlobCache { } const raw = join(this.options.root, "tmp", `${token}.raw`), gz = join(this.options.root, "tmp", `${token}.gz`); let input: Readable | undefined, file: string | undefined; + let renewAt = Date.now() + 30000, leaseLost = false; + const loseLease = () => { leaseLost = true; input?.destroy(new Error("cache_fill_lease_lost")); }; const heartbeat = setInterval(() => { - try { if (!this.db.prepare("UPDATE fills SET expires=? WHERE key=? AND token=?").run(Date.now() + 120000, key, token).changes) input?.destroy(new Error("cache_fill_lease_lost")); } - catch { input?.destroy(new Error("cache_fill_lease_lost")); } - }, 30000); heartbeat.unref(); + const now = Date.now(); + if (leaseLost || now < renewAt) return; + if (now >= leaseUntil) { loseLease(); return; } + try { + const renewedUntil = now + 120000; + if (!this.db.prepare("UPDATE fills SET expires=? WHERE key=? AND token=? AND expires>?").run(renewedUntil, key, token, now).changes) { + loseLease(); return; + } + leaseUntil = renewedUntil; renewAt = now + 30000; + } catch (error) { + if (!isCacheContention(error)) loseLease(); + // Retry contention on the next tick within the last confirmed lease. + } + }, 1000); 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. @@ -314,6 +329,7 @@ export class BlobCache { // 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(); + if (leaseLost || Date.now() >= leaseUntil) throw new Error("cache_fill_lease_lost"); let size = 0; const sha256 = createHash("sha256"); await pipeline(input, new Transform({ transform: (chunk, _encoding, done) => { size += chunk.length; sha256.update(chunk); diff --git a/test/shared-blob-cache.test.js b/test/shared-blob-cache.test.js index 1da7367..604df6b 100644 --- a/test/shared-blob-cache.test.js +++ b/test/shared-blob-cache.test.js @@ -212,6 +212,42 @@ describe("shared content-addressed blob pool", function () { 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 });