perf: share verified blobs and expose bounded cache metrics

scripts/cache-index-performance.js

@@ -0,0 +1,58 @@
+/* Synthetic metadata cardinality and bounded eviction probes. No production access. */
+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 { performance } = require('node:perf_hooks');
+require('ts-node/register/transpile-only');
+const config = require('../src/config').default;
+const { BlobCache } = require('../src/core/blob-cache');
+const percentile=(values,p)=>[...values].sort((a,b)=>a-b)[Math.min(values.length-1,Math.ceil(values.length*p)-1)];
+const consume=async stream=>{let size=0;for await(const chunk of stream)size+=chunk.length;return size;};
+async function main(){
+  const root=await fs.mkdtemp(join(tmpdir(),'cache-index-performance-'));config.CACHE_MONITORING_ENABLED=false;
+  const results=[];let cache;
+  try{
+    for(const cardinality of [0,10000,100000]){
+      cache=new BlobCache({root:join(root,'index-'+cardinality),budget:1024**3,maxObject:4096,gzip:false});
+      const real=await cache.getOrFill('real',async()=>Readable.from('verified real content'));const digest=real.cachedBlob.record.digest;await consume(real);
+      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 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.
+        const key='sha256-'+i.toString(16).padStart(64,'0');blob.run(key,'dead.raw',4096,4096,'identity',now,4096);
+        alias.run('seed-'+i,digest);alias.run('alias-'+i,digest);decision.run('rule-'+i,digest,0,now,now+86400000);history.run('history-'+i,now);
+      }
+      cache.db.exec('COMMIT');
+      const stats=[],hits=[],fills=[],decisions=[];
+      for(let i=0;i<30;i++){
+        let start=performance.now();cache.stats();stats.push(performance.now()-start);
+        start=performance.now();assert.equal(await consume(await cache.getOrFill(digest,async()=>{throw Error('unexpected miss');})),21);hits.push(performance.now()-start);
+        start=performance.now();cache.saveDecision('measured-'+i,digest,false);decisions.push(performance.now()-start);
+        start=performance.now();await consume(await cache.getOrFill('new-'+i,async()=>Readable.from('verified real content')));fills.push(performance.now()-start);
+      }
+      const summarize=x=>({p50Ms:percentile(x,.5),p95Ms:percentile(x,.95),maxMs:Math.max(...x)});
+      results.push({seedBlobs:cardinality,seedAliases:cardinality*2,seedDecisions:cardinality,seedHistory:cardinality,
+        adminStats:summarize(stats),warmSourceHit:summarize(hits),decisionWrite:summarize(decisions),smallFill:summarize(fills)});
+      cache.close();cache=undefined;
+    }
+    cache=new BlobCache({root:join(root,'eviction'),budget:4*1024**2,maxObject:65536,gzip:false});
+    const times=[];let upstream=0,peakAllocated=0;const started=performance.now();
+    for(let i=0;i<300;i++){
+      // 100 distinct objects, three sequential traversals: deliberately larger than the pool.
+      const id=i%100,data=Buffer.alloc(65536,id);const begin=performance.now();
+      await consume(await cache.getOrFill('object-'+id,async()=>{upstream++;return Readable.from(data);}));times.push(performance.now()-begin);
+      const stats=cache.stats();peakAllocated=Math.max(peakAllocated,stats.allocatedBytes);assert.ok(stats.allocatedBytes+stats.reservedBytes<=stats.budgetBytes);
+    }
+    const report={measuredAt:new Date().toISOString(),scope:'Synthetic metadata-only scale probe. Dummy blob rows have no payload files; excludes directory scanning and disk reads for those rows. Real reads use one verified blob. Eviction probe uses real 64 KiB files.',
+      node:process.version,cardinality:results,eviction:{requests:300,distinctObjects:100,upstreamFills:upstream,hitRate:1-upstream/300,
+        elapsedMs:performance.now()-started,p50Ms:percentile(times,.5),p95Ms:percentile(times,.95),peakAllocated,final:cache.stats()},assertions:'passed'};
+    await fs.writeFile(process.argv[2]||'docs/cache-index-performance-results.json',JSON.stringify(report,null,2)+'\n');
+    process.stdout.write(JSON.stringify(report,null,2)+'\n');
+  }finally{cache?.close();await fs.rm(root,{recursive:true,force:true});}
+}
+main().catch(error=>{console.error(error);process.exitCode=1;});

scripts/cache-performance.js

@@ -0,0 +1,140 @@
+/* Local HTTP benchmark. Uses temporary files and deterministic synthetic content, never production. */
+const { strict: assert } = require('node:assert');
+const { promises: fs, createReadStream } = require('node:fs');
+const { tmpdir, cpus, totalmem, loadavg } = require('node:os');
+const { join } = require('node:path');
+const { createHash } = require('node:crypto');
+const { Readable } = require('node:stream');
+const { pipeline } = require('node:stream/promises');
+const { createGzip, gunzipSync } = require('node:zlib');
+const { performance, monitorEventLoopDelay } = require('node:perf_hooks');
+const { setTimeout: delay } = require('node:timers/promises');
+const { setInterval, clearInterval } = require('node:timers');
+const { fork } = require('node:child_process');
+const http = require('node:http');
+const percentile = (values, p) => [...values].sort((a,b) => a-b)[Math.min(values.length-1, Math.ceil(values.length*p)-1)];
+function fixture(size, binary) {
+  const chunks = []; let length = 0;
+  for (let i=0; length<size; i++) {
+    const digest = createHash('sha256').update('cache-evaluation-' + i).digest();
+    const chunk = binary ? digest : Buffer.from(`${i}: ResearcherAlice record ${digest.toString('hex')} module/path/${i % 137}.ts\n`);
+    chunks.push(chunk); length += chunk.length;
+  }
+  return Buffer.concat(chunks).subarray(0,size);
+}
+async function worker(spec) {
+  let privateTmp;
+  if(spec.mode==='legacy-transform'){privateTmp=await fs.mkdtemp(join(tmpdir(),'cache-legacy-bench-'));process.env.TMPDIR=privateTmp;}
+  if(process.env.CACHE_BENCH_COMPILED !== '1') require('ts-node/register/transpile-only');
+  const code=process.env.CACHE_BENCH_COMPILED === '1' ? '../build' : '../src';
+  const express = require('express');
+  const config = require(code+'/config').default;
+  const { sharedBlobCache, resultBlobCache, closeBlobCaches } = require(code+'/core/blob-cache');
+  const { sendSharedContent } = require(code+'/core/cached-response');
+  const { AnonymizeTransformer } = require(code+'/core/anonymize-utils');
+  const { closeCacheMetrics, cacheMetricsReport } = require(code+'/core/cache-observability');
+  const root = await fs.mkdtemp(join(tmpdir(), 'cache-performance-'));
+  const body = fixture(spec.size, spec.binary);
+  const expected = spec.changed ? Buffer.from(body.toString().replaceAll('ResearcherAlice', 'ResearcherHidden')) : body;
+  const digest = 'git-' + createHash('sha1').update(`blob ${body.length}\0`).update(body).digest('hex');
+  const raw = join(root, 'source.raw'); let sourceReady = false, upstreamFills = 0, server;
+  Object.assign(config, { FOLDER:root, SHARED_CACHE_PATH:join(root,'pool'), SHARED_CACHE_ENABLED:spec.mode==='shared',
+    SHARED_CACHE_BYTES:64*1024**2, SHARED_CACHE_RESULT_BYTES:32*1024**2, SHARED_CACHE_MAX_FILE_BYTES:4*1024**2,
+    SHARED_CACHE_FREE_BYTES:0, CACHE_MONITORING_ENABLED:true, REDIS_HOSTNAME:'127.0.0.1', REDIS_PORT:1 });
+  if (spec.sqliteNormal) { sharedBlobCache().db.exec('PRAGMA synchronous=NORMAL'); resultBlobCache().db.exec('PRAGMA synchronous=NORMAL'); }
+  const options = { filePath:spec.binary?'fixture.bin':'fixture.txt', repoId:'performance-project',
+    terms:spec.changed?['ResearcherAlice=>ResearcherHidden']:['AbsentResearcher'], decisionManaged:spec.mode!=='legacy-transform' };
+  const obtain = async () => {
+    if (spec.mode === 'uncached') { upstreamFills++; await delay(20); return Readable.from(body); }
+    if (spec.mode === 'raw-source' || spec.mode === 'legacy-transform') {
+      if (!sourceReady) { upstreamFills++; await delay(20); await fs.writeFile(raw,body); sourceReady=true; }
+      return createReadStream(raw);
+    }
+    return sharedBlobCache().getOrFill(digest, async()=>{upstreamFills++;await delay(20);return Readable.from(body);},
+      async proof=>{assert.equal('git-'+proof.gitSha,digest);return digest;},body.length);
+  };
+  const app=express(); app.get('/file',async(_req,res)=>{
+    res.type(spec.binary?'application/octet-stream':'text/plain');
+    try {
+      const source=await obtain();
+      if (spec.mode==='shared' && await sendSharedContent(source,res,options)) return;
+      const stages=[source,new AnonymizeTransformer(options)];
+      if(spec.encoding==='gzip'){res.setHeader('Content-Encoding','gzip');stages.push(createGzip({level:6}));}
+      await pipeline(...stages,res);
+    }catch(error){res.destroy(error);}
+  });
+  const agent=new http.Agent({keepAlive:true,maxSockets:8});
+  let wireHash;
+  try {
+    await new Promise(resolve=>{server=app.listen(0,'127.0.0.1',resolve);});
+    const request=()=>new Promise((resolve,reject)=>{
+      const start=performance.now();
+      http.get({host:'127.0.0.1',port:server.address().port,path:'/file',agent,headers:{'accept-encoding':spec.encoding}},res=>{
+        const ttfb=performance.now()-start, chunks=[]; let bytes=0;const sha=createHash('sha256');
+        res.on('data',b=>{bytes+=b.length;sha.update(b);if(!wireHash)chunks.push(b);});
+        res.on('error',reject);res.on('end',()=>{
+          try {
+            assert.equal(res.statusCode,200); const digest=sha.digest('hex');
+            const elapsed=performance.now()-start;
+            if(!wireHash){const data=Buffer.concat(chunks);assert.deepEqual(res.headers['content-encoding']==='gzip'?gunzipSync(data):data,expected);wireHash=digest;}
+            else assert.equal(digest,wireHash);
+            resolve({ms:elapsed,ttfb,bytes});
+          }catch(error){reject(error);}
+        });
+      }).on('error',reject);
+    });
+    const coldCpu=process.cpuUsage(),coldStart=performance.now();const cold=await request();
+    const coldResources={...cold,wallMs:performance.now()-coldStart,cpuMs:Object.values(process.cpuUsage(coldCpu)).reduce((a,b)=>a+b,0)/1000};
+    // JIT and worker warmup is excluded from warm samples.
+    for(let i=0;i<3;i++) await request();
+    const batches=[];
+    for(const concurrency of [1,8]) for(let repeat=0;repeat<2;repeat++) {
+      const count=concurrency===1?16:32;let next=0;const rows=[];
+      let peakRss=process.memoryUsage().rss;
+      const timer=setInterval(()=>{peakRss=Math.max(peakRss,process.memoryUsage().rss);},5);
+      const eventLoop=monitorEventLoopDelay({resolution:10});eventLoop.enable();await delay(20);
+      const cpu=process.cpuUsage(), start=performance.now();
+      await Promise.all(Array.from({length:concurrency},async()=>{for(;;){const i=next++;if(i>=count)return;rows.push(await request());}}));
+      const elapsed=performance.now()-start,cpuUsed=process.cpuUsage(cpu);eventLoop.disable();clearInterval(timer);
+      batches.push({concurrency,repeat,requests:count,elapsedMs:elapsed,requestsPerSecond:count*1000/elapsed,
+        p50Ms:percentile(rows.map(r=>r.ms),.5),p95Ms:percentile(rows.map(r=>r.ms),.95),maxMs:Math.max(...rows.map(r=>r.ms)),
+        ttfbP95Ms:percentile(rows.map(r=>r.ttfb),.95),cpuMs:(cpuUsed.user+cpuUsed.system)/1000,
+        cpuMsPerRequest:(cpuUsed.user+cpuUsed.system)/1000/count,peakRssMiB:Math.max(peakRss,process.memoryUsage().rss)/1048576,
+        eventLoopP95Ms:eventLoop.count?eventLoop.percentile(95)/1e6:null,wireBytes:rows.reduce((n,r)=>n+r.bytes,0),latenciesMs:rows.map(r=>r.ms)});
+    }
+    const metrics=cacheMetricsReport(60);
+    const report={...spec,cold:coldResources,batches,upstreamFills,source:sharedBlobCache()?.stats()||null,results:resultBlobCache()?.stats()||null,
+      metrics:metrics.rows,wireBytesPerResponse:cold.bytes,uncompressedBytes:expected.length,verifiedResponses:100};
+    process.send({report});
+  } finally {agent.destroy();if(server)await new Promise(resolve=>server.close(resolve));closeBlobCaches();closeCacheMetrics();await fs.rm(root,{recursive:true,force:true});if(privateTmp)await fs.rm(privateTmp,{recursive:true,force:true});}
+}
+async function main() {
+  let scenarios=[];
+  for(const size of [16384,262144,2097152])for(const changed of [false,true])for(const mode of ['uncached','raw-source','shared'])scenarios.push({size,changed,mode,encoding:'gzip',binary:false});
+  for(const size of [262144,2097152])scenarios.push({size,changed:true,mode:'shared',encoding:'identity',binary:false});
+  for(const mode of ['raw-source','shared'])scenarios.push({size:262144,changed:false,mode,encoding:'gzip',binary:true});
+  if (process.argv.includes('--shared')) scenarios=scenarios.filter(s=>s.mode==='shared' && s.changed && s.encoding==='gzip');
+  if (process.argv.includes('--normal')) scenarios=[16384,262144,2097152].map(size=>({size,changed:true,mode:'shared',encoding:'gzip',binary:false,sqliteNormal:true}));
+  if (process.argv.includes('--legacy')) scenarios=[false,true].map(changed=>({size:2097152,changed,mode:'legacy-transform',encoding:'gzip',binary:false}));
+  const report={measuredAt:new Date().toISOString(),environment:{execution:process.env.CACHE_BENCH_COMPILED === '1' ? 'compiled JavaScript' : 'ts-node',node:process.version,platform:process.platform,arch:process.arch,cpu:cpus()[0]?.model,logicalCpus:cpus().length,totalMemoryGiB:totalmem()/1024**3,loadAverage:loadavg()},
+    methodology:{scope:'Local loopback HTTP, deterministic synthetic content, isolated process per scenario; no production traffic',
+      upstreamDelayMs:20,rawSourceBaseline:'Raw local file; anonymization and gzip repeated every request; no transformed-result cache. Not a complete legacy deployment comparison.',
+      measurements:'Cold request then 3 warmups, 2 batches each at concurrency 1 (16 requests) and 8 (32 requests). CPU and RSS include server, loopback client, hashing and Node runtime; worker-thread CPU included. First cold request also includes worker startup when applicable.',
+      limitations:'Warm OS page cache; no TLS, proxy, GitHub bandwidth, database auth or real-world RTT. Small sample p95 is descriptive, not a capacity SLA. Fresh metadata indexes; not a full-size index test.'},scenarios:[]};
+  for(const spec of scenarios){
+    process.stderr.write(`Measuring ${JSON.stringify(spec)}\n`);
+    report.scenarios.push(await new Promise((resolve,reject)=>{
+      const child=fork(__filename,['--worker',JSON.stringify(spec)],{env:{...process.env,NODE_ENV:'test'},stdio:['ignore','ignore','pipe','ipc']});
+      let result,stderr='';const timeout=setInterval(()=>{child.kill();reject(Error('Scenario exceeded 120 seconds'));},120000);
+      child.stderr.on('data',b=>{stderr=(stderr+b).slice(-4000);});child.on('message',m=>{result=m.report;});child.on('error',reject);
+      child.on('exit',code=>{clearInterval(timeout);if(code===0&&result)resolve(result);else reject(Error(`Scenario failed ${code}: ${stderr}`));});
+    }));
+  }
+  report.completedAt=new Date().toISOString();report.assertions=`All ${scenarios.length*100} response bodies verified against expected output or the validated wire digest`;
+  if (process.argv.includes('--normal')) report.methodology.experimentalOverride='PRAGMA synchronous=NORMAL in both temporary pool connections only. Application defaults are unchanged. WAL commits may be lost after power failure.';
+  if (process.argv.includes('--legacy')) report.methodology.legacyComparison='Existing large-file transformedFile cache enabled, isolated TMPDIR, local source file, no Redis coordination or database authorization.';
+  await fs.writeFile(process.argv[2]||'docs/cache-performance-results.json',JSON.stringify(report,null,2)+'\n');
+  process.stdout.write(JSON.stringify({scenarios:report.scenarios.length,assertions:report.assertions})+'\n');
+}
+if(process.argv[2]==='--worker')worker(JSON.parse(process.argv[3])).then(()=>process.exit(0),e=>{console.error(e);process.exit(1);});
+else main().catch(e=>{console.error(e);process.exitCode=1;});

scripts/cache-workload.js

@@ -0,0 +1,70 @@
+/* Reproducible local acceptance workload. Never connects to production or downloads upstream files. */
+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 } = require("node:crypto");
+const http = require("node:http");
+const express = require("express");
+require("ts-node/register/transpile-only");
+const config = require("../src/config").default;
+const { BlobCache, sharedBlobCache, resultBlobCache, closeBlobCaches } = require("../src/core/blob-cache");
+const { sendSharedContent } = require("../src/core/cached-response");
+const { cacheMetricsReport, closeCacheMetrics } = require("../src/core/cache-observability");
+const { monitorRequests } = require("../src/core/request-monitoring");
+const consume = async s => { const chunks = []; for await (const b of s) chunks.push(Buffer.from(b)); return Buffer.concat(chunks); };
+async function main() {
+  const root = await fs.mkdtemp(join(tmpdir(), "cache-workload-")); let server, eviction;
+  const old = { ...config };
+  Object.assign(config, { SHARED_CACHE_ENABLED: true, SHARED_CACHE_PATH: root, SHARED_CACHE_BYTES: 1024 * 1024,
+    SHARED_CACHE_RESULT_BYTES: 1024 * 1024, SHARED_CACHE_MAX_FILE_BYTES: 32768, SHARED_CACHE_FREE_BYTES: 0, CACHE_MONITORING_ENABLED: true });
+  try {
+    const source = sharedBlobCache();
+    const content = Buffer.from("A public example with repeatable documentation.\n".repeat(400));
+    const key = "git-" + createHash("sha1").update(`blob ${content.length}\0`).update(content).digest("hex");
+    let fills = 0;
+    const obtain = () => source.getOrFill(key, async () => { fills++; return Readable.from(content); }, async p => {
+      assert.equal(key, "git-" + p.gitSha); return key;
+    }, content.length);
+    const app = express(); app.use(monitorRequests("api", undefined, { info() {}, warn() {} }));
+    app.get("/:project/:rule", async (req, res) => {
+      res.type("text/plain");
+      try { assert.equal(await sendSharedContent(await obtain(), res, { filePath: "README.md", repoId: req.params.project,
+        terms: req.params.rule === "changed" ? ["public=>private"] : ["Alice"] }), true); }
+      catch (e) { res.destroy(e); }
+    });
+    await new Promise(resolve => { server = app.listen(0, "127.0.0.1", resolve); });
+    const responses = [];
+    for (const path of ["/a/plain", "/b/plain", "/a/plain", "/a/changed", "/a/changed"]) {
+      responses.push(await new Promise((resolve, reject) => {
+        http.get({ host: "127.0.0.1", port: server.address().port, path, headers: { "accept-encoding": "gzip" } }, res => {
+          consume(res).then(b => resolve({ status: res.statusCode, encoding: res.headers["content-encoding"], bytes: b.length }), reject);
+        }).on("error", reject);
+      }));
+    }
+    assert.equal(fills, 1); assert.ok(responses.every(r => r.status === 200 && r.encoding === "gzip"));
+    eviction = new BlobCache({ root: join(root, "eviction"), budget: 65536, maxObject: 4096, gzip: false, name: "workload" });
+    for (let i = 0; i < 30; i++) {
+      const data = Buffer.alloc(4000, i);
+      await consume(await eviction.getOrFill(`item-${i}`, async () => Readable.from(data)));
+      const stats = eviction.stats(); assert.ok(stats.allocatedBytes + stats.reservedBytes <= stats.budgetBytes);
+    }
+    const stats = eviction.stats(); assert.ok(stats.blobs < 30);
+    const metrics = cacheMetricsReport(60);
+    const count = name => metrics.rows.filter(r => r.name === name).reduce((n, r) => n + r.count, 0);
+    assert.equal(count("decision_hit"), 2); assert.equal(count("rewrite_changed"), 1);
+    const result = { node: process.version, measuredAt: new Date().toISOString(), scope: "Synthetic local acceptance workload; not a production projection",
+      sourceRequests: 5, upstreamFills: fills, source: source.stats(), results: resultBlobCache().stats(), eviction: stats,
+      decisionsReused: count("decision_hit"), unchangedDecisionsComputed: count("rewrite_unchanged"), changedOutputsComputed: count("rewrite_changed"),
+      compressionSavingPercent: 100 * (1 - source.stats().storedBytes / source.stats().originalBytes), responses, metrics: metrics.rows,
+      assertions: "passed" };
+    const json = JSON.stringify(result, null, 2) + "\n";
+    if (process.argv[2]) await fs.writeFile(process.argv[2], json);
+    process.stdout.write(json);
+  } finally {
+    if (server) await new Promise(resolve => server.close(resolve));
+    eviction?.close(); closeBlobCaches(); closeCacheMetrics(); Object.assign(config, old); await fs.rm(root, { recursive: true, force: true });
+  }
+}
+main().catch(e => { console.error(e); process.exitCode = 1; });