fix: preserve shared cache response semantics and recover from contention

docs/shared-cache.md

@@ -52,6 +52,11 @@ and retire legacy data only after validating rollout. Uploaded ZIP originals and
 other durable source material are untouched. Repository removal invalidates access
 and project mappings, while shared blobs remain subject to global eviction.
 
+Transient SQLite write contention bypasses caching instead of failing file requests.
+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.
+
 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

scripts/cache-index-performance.js

@@ -20,7 +20,7 @@ async function main(){
       const now=Date.now();cache.db.exec('BEGIN');
       const blob=cache.db.prepare('INSERT INTO blobs VALUES(?,?,?,?,?,?,?)');
       const alias=cache.db.prepare('INSERT INTO aliases VALUES(?,?)');
-      const decision=cache.db.prepare('INSERT INTO decisions VALUES(?,?,?,?,?)');
+      const decision=cache.db.prepare('INSERT INTO decisions(key,digest,changed,accessed,expires) VALUES(?,?,?,?,?)');
       const history=cache.db.prepare('INSERT INTO history VALUES(?,?,0)');
       for(let i=0;i<cardinality;i++){
         // Metadata-only dummy rows, never opened. Real lookups/fills use the separate verified content above.

src/core/blob-cache.ts

@@ -10,6 +10,14 @@ import config from "../config";
 import { observeCache } from "./cache-observability";
 
 export class CacheAdmissionError extends Error {}
+export function isCacheContention(error: unknown) {
+  const code = Number((error as { errcode?: number })?.errcode) & 255;
+  return code === 5 || code === 6;
+}
+function cacheError(error: unknown): never {
+  if (isCacheContention(error)) throw new CacheAdmissionError("cache_index_busy");
+  throw error;
+}
 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 }
@@ -25,50 +33,55 @@ export class BlobCache {
     this.name = options.name || "source";
     mkdirSync(join(options.root, "tmp"), { recursive: true, mode: 0o700 });
     this.db = new DatabaseSync(join(options.root, "index.sqlite"));
-    this.db.exec(`PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL;
-      CREATE TABLE IF NOT EXISTS blobs(digest TEXT PRIMARY KEY,file TEXT,size INTEGER,stored INTEGER,encoding TEXT,accessed INTEGER,allocated INTEGER);
-      CREATE INDEX IF NOT EXISTS blobs_lru ON blobs(accessed);
-      CREATE TABLE IF NOT EXISTS aliases(key TEXT PRIMARY KEY,digest TEXT REFERENCES blobs(digest) ON DELETE CASCADE);
-      CREATE INDEX IF NOT EXISTS aliases_digest ON aliases(digest);
-      CREATE TABLE IF NOT EXISTS fills(key TEXT PRIMARY KEY,token TEXT,expires INTEGER,reserved INTEGER);
-      CREATE TABLE IF NOT EXISTS publications(token TEXT PRIMARY KEY,digest TEXT,file TEXT);
-      CREATE TABLE IF NOT EXISTS readers(token TEXT PRIMARY KEY,digest TEXT,expires INTEGER);
-      CREATE INDEX IF NOT EXISTS readers_digest ON readers(digest);
-      CREATE INDEX IF NOT EXISTS readers_expiry ON readers(expires);
-      CREATE TABLE IF NOT EXISTS decisions(key TEXT PRIMARY KEY,digest TEXT,changed INTEGER,accessed INTEGER,expires INTEGER);
-      CREATE INDEX IF NOT EXISTS decisions_age ON decisions(accessed);
-      CREATE INDEX IF NOT EXISTS decisions_expiry ON decisions(expires);
-      CREATE TABLE IF NOT EXISTS maintenance(id INTEGER PRIMARY KEY,nextScan INTEGER);
-      CREATE TABLE IF NOT EXISTS history(key TEXT PRIMARY KEY,seen INTEGER,evicted INTEGER);
-      CREATE INDEX IF NOT EXISTS history_seen ON history(seen);
-      CREATE INDEX IF NOT EXISTS blobs_file ON blobs(file);
-      CREATE TABLE IF NOT EXISTS totals(id INTEGER PRIMARY KEY, bytes INTEGER);
-      CREATE TRIGGER IF NOT EXISTS blob_added AFTER INSERT ON blobs BEGIN UPDATE totals SET bytes=bytes+NEW.allocated WHERE id=1; END;
-      CREATE TRIGGER IF NOT EXISTS blob_removed AFTER DELETE ON blobs BEGIN UPDATE totals SET bytes=bytes-OLD.allocated WHERE id=1; END;
-      PRAGMA foreign_keys=ON; PRAGMA recursive_triggers=ON;`);
-    // Initialize once under a write lock; all connections see data and counters atomically.
-    this.transaction(() => {
-      if (!this.db.prepare("SELECT 1 FROM totals WHERE id=1").get()) this.db.exec("INSERT INTO totals SELECT 1,COALESCE(SUM(allocated),0) FROM blobs");
-      this.db.exec(`CREATE TABLE IF NOT EXISTS summary(id INTEGER PRIMARY KEY,blobs INTEGER,originalBytes INTEGER,storedBytes INTEGER,allocatedBytes INTEGER,activeReaders INTEGER,decisions INTEGER,aliases INTEGER,history INTEGER,reservedBytes INTEGER)`);
-      if (!this.db.prepare("SELECT 1 FROM summary WHERE id=1").get()) {
-        this.db.exec(`INSERT INTO summary SELECT 1,COUNT(*),COALESCE(SUM(size),0),COALESCE(SUM(stored),0),COALESCE(SUM(allocated),0),
-          (SELECT COUNT(*) FROM readers),(SELECT COUNT(*) FROM decisions),(SELECT COUNT(*) FROM aliases),(SELECT COUNT(*) FROM history),(SELECT COALESCE(SUM(reserved),0) FROM fills) FROM blobs`);
-      }
-      for (const [table, columns] of [
-        ["blobs", { blobs: "1", originalBytes: "size", storedBytes: "stored", allocatedBytes: "allocated" }],
-        ["readers", { activeReaders: "1" }], ["decisions", { decisions: "1" }],
-        ["aliases", { aliases: "1" }], ["history", { history: "1" }], ["fills", { reservedBytes: "reserved" }]
-      ] as [string, Record<string, string>][]) {
-        const changes = (prefix: string, sign: string) => Object.entries(columns).map(([column, value]) =>
-          `${column}=${column}${sign}${value === "1" ? "1" : prefix + "." + value}`).join(",");
-        this.db.exec(`CREATE TRIGGER IF NOT EXISTS summary_${table}_insert AFTER INSERT ON ${table} BEGIN UPDATE summary SET ${changes("NEW", "+")} WHERE id=1; END;
-          CREATE TRIGGER IF NOT EXISTS summary_${table}_delete AFTER DELETE ON ${table} BEGIN UPDATE summary SET ${changes("OLD", "-")} WHERE id=1; END;`);
-        const values = Object.entries(columns).filter(([, value]) => value !== "1");
-        if (values.length) this.db.exec(`CREATE TRIGGER IF NOT EXISTS summary_${table}_update AFTER UPDATE OF ${values.map(([, value]) => value).join(",")} ON ${table}
-          BEGIN UPDATE summary SET ${values.map(([column, value]) => `${column}=${column}-OLD.${value}+NEW.${value}`).join(",")} WHERE id=1; END;`);
-      }
-    });
-    this.db.exec("PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=50;");
+    try {
+      this.db.exec(`PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL;
+        CREATE TABLE IF NOT EXISTS blobs(digest TEXT PRIMARY KEY,file TEXT,size INTEGER,stored INTEGER,encoding TEXT,accessed INTEGER,allocated INTEGER);
+        CREATE INDEX IF NOT EXISTS blobs_lru ON blobs(accessed);
+        CREATE TABLE IF NOT EXISTS aliases(key TEXT PRIMARY KEY,digest TEXT REFERENCES blobs(digest) ON DELETE CASCADE);
+        CREATE INDEX IF NOT EXISTS aliases_digest ON aliases(digest);
+        CREATE TABLE IF NOT EXISTS fills(key TEXT PRIMARY KEY,token TEXT,expires INTEGER,reserved INTEGER);
+        CREATE TABLE IF NOT EXISTS publications(token TEXT PRIMARY KEY,digest TEXT,file TEXT);
+        CREATE TABLE IF NOT EXISTS readers(token TEXT PRIMARY KEY,digest TEXT,expires INTEGER);
+        CREATE INDEX IF NOT EXISTS readers_digest ON readers(digest);
+        CREATE INDEX IF NOT EXISTS readers_expiry ON readers(expires);
+        CREATE TABLE IF NOT EXISTS decisions(key TEXT PRIMARY KEY,digest TEXT,changed INTEGER,accessed INTEGER,expires INTEGER);
+        CREATE INDEX IF NOT EXISTS decisions_age ON decisions(accessed);
+        CREATE INDEX IF NOT EXISTS decisions_expiry ON decisions(expires);
+        CREATE TABLE IF NOT EXISTS maintenance(id INTEGER PRIMARY KEY,nextScan INTEGER);
+        CREATE TABLE IF NOT EXISTS history(key TEXT PRIMARY KEY,seen INTEGER,evicted INTEGER);
+        CREATE INDEX IF NOT EXISTS history_seen ON history(seen);
+        CREATE INDEX IF NOT EXISTS blobs_file ON blobs(file);
+        CREATE TABLE IF NOT EXISTS totals(id INTEGER PRIMARY KEY, bytes INTEGER);
+        CREATE TRIGGER IF NOT EXISTS blob_added AFTER INSERT ON blobs BEGIN UPDATE totals SET bytes=bytes+NEW.allocated WHERE id=1; END;
+        CREATE TRIGGER IF NOT EXISTS blob_removed AFTER DELETE ON blobs BEGIN UPDATE totals SET bytes=bytes-OLD.allocated WHERE id=1; END;
+        PRAGMA foreign_keys=ON; PRAGMA recursive_triggers=ON;`);
+      // Initialize once under a write lock; all connections see data and counters atomically.
+      this.transaction(() => {
+        // Classification is nullable for decisions written by older versions.
+        if (!this.db.prepare("PRAGMA table_info(decisions)").all().some(row => row.name === "isText"))
+          this.db.exec("ALTER TABLE decisions ADD COLUMN isText INTEGER");
+        if (!this.db.prepare("SELECT 1 FROM totals WHERE id=1").get()) this.db.exec("INSERT INTO totals SELECT 1,COALESCE(SUM(allocated),0) FROM blobs");
+        this.db.exec(`CREATE TABLE IF NOT EXISTS summary(id INTEGER PRIMARY KEY,blobs INTEGER,originalBytes INTEGER,storedBytes INTEGER,allocatedBytes INTEGER,activeReaders INTEGER,decisions INTEGER,aliases INTEGER,history INTEGER,reservedBytes INTEGER)`);
+        if (!this.db.prepare("SELECT 1 FROM summary WHERE id=1").get()) {
+          this.db.exec(`INSERT INTO summary SELECT 1,COUNT(*),COALESCE(SUM(size),0),COALESCE(SUM(stored),0),COALESCE(SUM(allocated),0),
+            (SELECT COUNT(*) FROM readers),(SELECT COUNT(*) FROM decisions),(SELECT COUNT(*) FROM aliases),(SELECT COUNT(*) FROM history),(SELECT COALESCE(SUM(reserved),0) FROM fills) FROM blobs`);
+        }
+        for (const [table, columns] of [
+          ["blobs", { blobs: "1", originalBytes: "size", storedBytes: "stored", allocatedBytes: "allocated" }],
+          ["readers", { activeReaders: "1" }], ["decisions", { decisions: "1" }],
+          ["aliases", { aliases: "1" }], ["history", { history: "1" }], ["fills", { reservedBytes: "reserved" }]
+        ] as [string, Record<string, string>][]) {
+          const changes = (prefix: string, sign: string) => Object.entries(columns).map(([column, value]) =>
+            `${column}=${column}${sign}${value === "1" ? "1" : prefix + "." + value}`).join(",");
+          this.db.exec(`CREATE TRIGGER IF NOT EXISTS summary_${table}_insert AFTER INSERT ON ${table} BEGIN UPDATE summary SET ${changes("NEW", "+")} WHERE id=1; END;
+            CREATE TRIGGER IF NOT EXISTS summary_${table}_delete AFTER DELETE ON ${table} BEGIN UPDATE summary SET ${changes("OLD", "-")} WHERE id=1; END;`);
+          const values = Object.entries(columns).filter(([, value]) => value !== "1");
+          if (values.length) this.db.exec(`CREATE TRIGGER IF NOT EXISTS summary_${table}_update AFTER UPDATE OF ${values.map(([, value]) => value).join(",")} ON ${table}
+            BEGIN UPDATE summary SET ${values.map(([column, value]) => `${column}=${column}-OLD.${value}+NEW.${value}`).join(",")} WHERE id=1; END;`);
+        }
+      });
+      this.db.exec("PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=50;");
+    } catch (error) { 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,
@@ -152,6 +165,9 @@ export class BlobCache {
       oldestAccess: this.db.prepare("SELECT accessed FROM blobs ORDER BY accessed LIMIT 1").get()?.accessed ?? null };
   }
   async open(digest: string, encoded = false): Promise<BlobStream | undefined> {
+    try { return await this.openReader(digest, encoded); } catch (error) { return cacheError(error); }
+  }
+  private async openReader(digest: string, encoded = false): Promise<BlobStream | undefined> {
     if (!validDigest(digest)) return;
     const token = randomUUID();
     const record = this.transaction(() => {
@@ -214,9 +230,13 @@ export class BlobCache {
     });
   }
   async getOrFill(key: string, produce: () => Promise<Readable>, verify?: (proof: BlobProof) => Promise<string>, sizeHint?: number): Promise<BlobStream> {
+    try { return await this.fill(key, produce, verify, sizeHint); } catch (error) { return cacheError(error); }
+  }
+  private async fill(key: string, produce: () => Promise<Readable>, verify?: (proof: BlobProof) => Promise<string>, sizeHint?: number): Promise<BlobStream> {
     if (!/^[a-z0-9-]{1,100}$/.test(key)) throw new Error("invalid_cache_key");
     let hit = await this.lookup(key);
-    this.accessed(key, !!hit, hit?.cachedBlob?.record.size || sizeHint || 0);
+    try { this.accessed(key, !!hit, hit?.cachedBlob?.record.size || sizeHint || 0); }
+    catch (error) { hit?.destroy(); throw error; }
     if (hit) return hit;
     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;
@@ -283,6 +303,8 @@ export class BlobCache {
         if (!duplicate) this.db.prepare("INSERT INTO blobs VALUES(?,?,?,?,?,?,?)").run(digest, record.file, size, stored, encoding, Date.now(), record.allocated);
         this.db.prepare("INSERT INTO aliases VALUES(?,?) ON CONFLICT(key) DO UPDATE SET digest=excluded.digest").run(key, digest);
         this.trim("aliases", 200000, "rowid");
+        // Bridge publication to open(), including asynchronous duplicate cleanup.
+        this.db.prepare("INSERT INTO readers VALUES(?,?,?)").run(token, digest, Date.now() + 120000);
       }, true);
       if (duplicate) { await fs.unlink(file); this.metric("deduplicated", stored); }
       file = undefined;
@@ -291,34 +313,49 @@ export class BlobCache {
     } finally {
       clearTimeout(timeout); clearInterval(heartbeat); input?.destroy();
       await Promise.all([raw, gz, ...(file ? [file] : [])].map(p => fs.unlink(p).catch(() => {})));
-      this.db.prepare("DELETE FROM publications WHERE token=?").run(token);
-      this.db.prepare("DELETE FROM fills WHERE key=? AND token=?").run(key, token);
+      this.releaseFill(key, token);
     }
   }
-  decision(key: string) {
-    const row = this.db.prepare("SELECT * FROM decisions WHERE key=? AND expires>?").get(key, Date.now());
-    if (row) this.db.prepare("UPDATE decisions SET accessed=? WHERE key=? AND accessed<?").run(Date.now(), key, Date.now() - 30000);
-    return row ? { digest: String(row.digest), changed: !!row.changed } : undefined;
+  private releaseFill(key: string, token: string) {
+    try {
+      this.transaction(() => {
+        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);
+      });
+    } catch (error) { if (!isCacheContention(error)) throw error; /* Leases expire conservatively. */ }
   }
-  saveDecision(key: string, digest: string, changed: boolean) {
-    this.transaction(() => {
-      this.db.prepare("INSERT INTO decisions VALUES(?,?,?,?,?) ON CONFLICT(key) DO UPDATE SET digest=excluded.digest,changed=excluded.changed,accessed=excluded.accessed,expires=excluded.expires").run(key, digest, +changed, Date.now(), Date.now() + 7 * 86400000);
-      this.trim("decisions", 100000, "accessed");
-    });
+  decision(key: string) {
+    try {
+      const row = this.db.prepare("SELECT * FROM decisions WHERE key=? AND expires>?").get(key, Date.now());
+      if (row) this.db.prepare("UPDATE decisions SET accessed=? WHERE key=? AND accessed<?").run(Date.now(), key, Date.now() - 30000);
+      return row ? { digest: String(row.digest), changed: !!row.changed, isText: row.isText == null ? undefined : !!row.isText } : undefined;
+    } catch (error) { return cacheError(error); }
+  }
+  saveDecision(key: string, digest: string, changed: boolean, isText?: boolean) {
+    try {
+      this.transaction(() => {
+        this.db.prepare("INSERT INTO decisions(key,digest,changed,accessed,expires,isText) VALUES(?,?,?,?,?,?) ON CONFLICT(key) DO UPDATE SET digest=excluded.digest,changed=excluded.changed,accessed=excluded.accessed,expires=excluded.expires,isText=excluded.isText")
+          .run(key, digest, +changed, Date.now(), Date.now() + 7 * 86400000, isText == null ? null : +isText);
+        this.trim("decisions", 100000, "accessed");
+      });
+    } catch (error) { if (!isCacheContention(error)) throw error; /* Optional decision can be recomputed. */ }
   }
   close() { this.db.close(); }
 }
 let source: BlobCache | undefined, results: BlobCache | undefined;
 export function sharedBlobCache() {
   if (!config.SHARED_CACHE_ENABLED || config.STORAGE !== "filesystem") return;
-  return source ||= new BlobCache({ root: join(config.SHARED_CACHE_PATH, "source"), budget: config.SHARED_CACHE_BYTES,
+  try { return source ||= new BlobCache({ 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;
-  return results ||= new BlobCache({ root: join(config.SHARED_CACHE_PATH, "results"), budget: config.SHARED_CACHE_RESULT_BYTES,
+  try { return results ||= new BlobCache({ 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; }
 }
 export function closeBlobCaches() { source?.close(); results?.close(); source = results = undefined; }

src/core/cached-response.ts

@@ -16,6 +16,11 @@ export function decisionKey(digest: string, options: unknown) {
 }
 /** Returns false when the caller should use the existing streaming pipeline. Authorization precedes this call. */
 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; }
+}
+async function sendCached(content: Readable, res: Response,
   options: ConstructorParameters<typeof AnonymizeTransformer>[0]): Promise<boolean> {
   const source = (content as BlobStream).cachedBlob;
   const results = resultBlobCache();
@@ -24,7 +29,8 @@ export async function sendSharedContent(content: Readable, res: Response,
   res.vary("Accept-Encoding");
   if (!encoding) { content.destroy(); res.status(406).end(); return true; }
   const key = decisionKey(source.record.digest, options);
-  const send = async (store: typeof results, digest: string, onHit?: () => void) => {
+  const namedType = classifyByName(options.filePath);
+  const send = async (store: typeof results, digest: string, isText: boolean, onHit?: () => void) => {
     const stream = await store.open(digest, encoding === "gzip");
     if (!stream) return false;
     const record = stream.cachedBlob!.record;
@@ -33,7 +39,9 @@ export async function sendSharedContent(content: Readable, res: Response,
     if (record.encoding === "gzip" && encoding === "gzip") {
       res.setHeader("Content-Encoding", "gzip"); res.setHeader("Content-Length", record.stored);
     } else { res.removeHeader("Content-Encoding"); res.setHeader("Content-Length", record.size); }
-    res.setHeader("Accept-Ranges", "none");
+    if (isText && !res.getHeader("Content-Type")) res.type("text/plain");
+    if (isText) res.setHeader("Accept-Ranges", "none");
+    else res.removeHeader("Accept-Ranges");
     if (record.encoding === "identity" && !res.req.acceptsEncodings("identity")) {
       res.setHeader("Content-Encoding", "gzip"); res.removeHeader("Content-Length");
       await pipeline(stream, createGzip(), res);
@@ -41,18 +49,19 @@ export async function sendSharedContent(content: Readable, res: Response,
     return true;
   };
   const decision = results.decision(key);
-  if (decision && await send(decision.changed ? results : source.store, decision.digest, () => {
+  if (decision && (decision.isText !== undefined || namedType !== null) && await send(decision.changed ? results : source.store, decision.digest, decision.isText ?? namedType!, () => {
     observeCache("decision_hit", source.record.size);
     observeCache(decision.changed ? "rewrite_changed_cached" : "rewrite_unchanged_cached", source.record.size);
   })) return true;
   observeCache("decision_miss", source.record.size);
   if (classifyByName(options.filePath) === false) {
-    results.saveDecision(key, source.record.digest, false);
+    results.saveDecision(key, source.record.digest, false, false);
     observeCache("rewrite_binary", source.record.size);
-    return send(source.store, source.record.digest);
+    return send(source.store, source.record.digest, false);
   }
   // Preserve the original source reader as a fallback if result admission fails.
   let processing: Readable | undefined;
+  let transformer: AnonymizeTransformer | undefined;
   const started = performance.now();
   const abort = () => { if (!res.writableFinished) processing?.destroy(new Error("response_cancelled")); };
   res.once("close", abort);
@@ -60,23 +69,23 @@ export async function sendSharedContent(content: Readable, res: Response,
     const output = await results.getOrFill(`render-${key}`, async () => {
       processing = await source.store.open(source.record.digest);
       if (!processing) throw new CacheAdmissionError("source_evicted");
-      const transform = new AnonymizeTransformer({ ...options, decisionManaged: true });
+      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 => {
       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); throw new Unchanged(); }
+      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);
+    results.saveDecision(key, digest, true, true);
     // Keep the result pinned until its HTTP reader has been opened.
-    try { if (await send(results, digest)) return true; } finally { output.destroy(); }
+    try { 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);
+    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(); }

src/core/source/GitHubStream.ts

@@ -394,16 +394,42 @@ export default class GitHubStream extends GitHubBase {
       const download = async () => {
         const parentToken = await this.data.getToken();
         const token = expected.upstream ? "" : parentToken;
-        const input = await origin.downloadWithFallback(token, expected.sha, sourcePath);
-        let bytes = 0;
-        const limited = new stream.Transform({ transform(chunk, _encoding, done) {
-          bytes += chunk.length;
-          done(bytes > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk);
-        } });
-        input.on("error", error => limited.destroy(error));
-        const deadline = setTimeout(() => limited.destroy(new Error("source_download_timeout")), 120000);
-        deadline.unref(); limited.once("close", () => { clearTimeout(deadline); input.destroy(); });
-        input.pipe(limited); return limited;
+        const sourceMiss = cachedContentMiss(this.data, parentToken);
+        if (sourceMiss && sourceMiss !== "file_not_found") throw new AnonymousError(sourceMiss, { httpStatus: 404 });
+        const missKey = createHash("sha256").update(JSON.stringify([
+          key, this.data.organization, this.data.repoName, token,
+        ])).digest("hex");
+        const missed = missingContent.get(missKey);
+        if (missed) throw new AnonymousError(missed, { httpStatus: 404 });
+        const normalize = async (error: Error) => {
+          const normalized = downloadError(error, filePath);
+          if (expected.upstream && ["file_not_found", "file_not_accessible"].includes(normalized.message))
+            return new AnonymousError("submodule_not_available", { httpStatus: 404, cause: normalized });
+          if (!expected.upstream && normalized.message === "file_not_found" && normalized.httpStatus === 404) {
+            const code = await classifyContentMiss(this.data, token);
+            missingContent.set(missKey, code, 30_000);
+            if (code !== "file_not_found") return new AnonymousError(code, { httpStatus: 404, cause: normalized });
+          }
+          return normalized;
+        };
+        const input = await origin.downloadWithFallback(token, expected.sha, sourcePath)
+          .catch(async error => { throw await normalize(error); });
+        input.on("error", () => {}); // The iterator below propagates normalized failures.
+        const deadline = setTimeout(() => input.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120000);
+        deadline.unref();
+        const limited = stream.Readable.from((async function* () {
+          let bytes = 0;
+          try {
+            for await (const chunk of input) {
+              bytes += chunk.length;
+              if (bytes > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });
+              yield chunk;
+            }
+          } catch (error) { throw await normalize(error as Error); }
+          finally { clearTimeout(deadline); input.destroy(); }
+        })());
+        stopSourceOnDestroy(limited, () => { clearTimeout(deadline); input.destroy(); });
+        return limited;
       };
       const produce = async () => {
         if (allowLegacy && await cached()) { legacy = true; observeCache("source_legacy_promotion"); return storage.read(repoId, cachePath); }

test/shared-blob-cache.test.js

@@ -57,6 +57,15 @@ describe("shared content-addressed blob pool", function () {
     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 = [];
@@ -145,6 +154,34 @@ describe("shared content-addressed blob pool", function () {
       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("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

@@ -39,12 +39,14 @@ describe("shared cache HTTP and telemetry integration", function () {
       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");
-      res.type("text/plain");
+      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("README.md", req.params.project, () => ({ sha: sha(body), size: body.length }));
-        const handled = await sendSharedContent(input, res, { filePath: "README.md", repoId: req.params.project,
+        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) { input.destroy(); res.status(500).end("unexpected bypass"); }
+        if (!handled) res.send(await consume(input));
       } catch (e) { if (!res.headersSent) res.status(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;
@@ -83,6 +85,79 @@ describe("shared cache HTTP and telemetry integration", function () {
       assert.equal(calls, 1);
     } finally { GitHubStream.prototype.downloadWithFallback = original; }
   });
+  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("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);