import test from 'node:test'; import assert from 'node:assert/strict'; import express from 'express'; import { once } from 'node:events'; import { RpcBudget, RpcError, resourceConfig } from '../server/resources.ts'; test('global admission has bounded concurrency, write tokens, refill and idempotent release', () => { let time = 0; const config = { ...resourceConfig({}), requestsPerSecond: 3, writesPerSecond: 1, active: 2 }; const budget = new RpcBudget(config, () => time); const a = budget.acquire(true); assert.throws(() => budget.acquire(true), /admission/); const b = budget.acquire(false); assert.throws(() => budget.acquire(false), /admission/); a(); a(); b(); assert.equal(budget.status().active, 0); budget.acquire(false)(); assert.throws(() => budget.acquire(false), /admission/); time = 1000; budget.acquire(true)(); }); test('response bytes are bounded globally and refill without per-client storage', () => { let time = 0; const budget = new RpcBudget({ ...resourceConfig({}), responseBytes: 50, bytesPerSecond: 60 }, () => time); assert.throws(() => budget.encode('x'.repeat(50)), /response limit/); budget.encode('x'.repeat(28)); budget.encode('x'.repeat(28)); assert.throws(() => budget.encode('a'), /byte budget/); time = 1000; assert.equal(budget.encode('a'), '"a"'); for (const value of ['-1','0','Infinity','1.2','']) assert.throws(() => resourceConfig({ WIKIDECK_RPC_MAX_ACTIVE: value })); }); test('isolated HTTP burst sheds excess work, preserves peer/health lanes and releases slots', async t => { const app = express(), cfg = { ...resourceConfig({}), active: 4, requestsPerSecond: 1000 }; const publicBudget = new RpcBudget(cfg), peerBudget = new RpcBudget(cfg); let release!: () => void; const gate = new Promise(resolve => { release = resolve; }); app.get('/healthz', (_req,res) => { res.json({ status:'ok' }); }); app.use('/api',publicBudget.middleware()); app.use('/p2p',peerBudget.middleware()); app.get('/api/hold',async (_req,res)=>{ await gate; res.json({ok:true}); }); app.get('/p2p/status',(_req,res)=>{ res.json({ok:true}); }); const server=app.listen(0,'127.0.0.1'); await once(server,'listening'); const address=server.address(); assert.ok(address&&typeof address==='object'); const base=`http://127.0.0.1:${address.port}`; t.after(()=>{release();server.closeAllConnections();server.close();}); const requests=Array.from({length:40},()=>fetch(`${base}/api/hold`,{signal:AbortSignal.timeout(5000)})); // The first rejected response proves requests reached admission while accepted requests remain held. const first=await Promise.race(requests); assert.equal(first.status,503); assert.equal((await fetch(`${base}/p2p/status`)).status,200); assert.equal((await fetch(`${base}/healthz`)).status,200); release(); const responses=await Promise.all(requests); const counts=responses.reduce((out,r)=>{out[r.status]=(out[r.status]??0)+1;return out;},{} as Record); assert.equal(counts[200],4); assert.equal(counts[503],36); for(const r of responses)await r.text(); assert.equal(publicBudget.status().active,0); t.diagnostic(JSON.stringify({requests:40,accepted:counts[200],shed:counts[503],peerLane:'available',health:'available'})); }); test('oversized response errors restore JSON handling and do not recurse', async t => { const app=express(),budget=new RpcBudget({...resourceConfig({}),responseBytes:128}); app.use(budget.middleware()); app.get('/large',(_req,res)=>{res.json({data:'x'.repeat(1000)});}); app.use((error:Error,_req:express.Request,res:express.Response,_next:express.NextFunction)=>{ res.status(error instanceof RpcError?error.status:500).json({error:error.message}); }); const server=app.listen(0,'127.0.0.1');await once(server,'listening'); t.after(()=>{server.closeAllConnections();server.close();}); const a=server.address();assert.ok(a&&typeof a==='object'); const r=await fetch(`http://127.0.0.1:${a.port}/large`,{signal:AbortSignal.timeout(5000)}); assert.equal(r.status,413);assert.match((await r.json()).error,/response limit/); assert.equal(budget.status().active,0); });