xiaoshu-app/scripts/tests/request-scheduler.test.mjs

import {test} from 'node:test';
import assert from 'node:assert/strict';
import {readFileSync,existsSync} from 'node:fs';
import {createRequire} from 'node:module';
import vm from 'node:vm';
import {Subject,of,firstValueFrom} from 'rxjs';
import ts from 'typescript';
const require=createRequire(import.meta.url),file=new URL('../../projects/shared/request-scheduler.ts',import.meta.url);
function scheduler(options){assert.ok(existsSync(file),'shared scheduler must exist');const exports={};vm.runInNewContext(ts.transpileModule(readFileSync(file,'utf8'),{compilerOptions:{module:ts.ModuleKind.CommonJS,target:ts.ScriptTarget.ES2022}}).outputText,{exports,require,setTimeout,clearTimeout,Date,Error,Map,JSON});return new exports.RequestScheduler(options);}
const flush=()=>new Promise(r=>setImmediate(r));
test('caps reads at three, reserves a fourth slot for writes, releases queued work',async()=>{
 const s=scheduler(),sources=Array.from({length:6},()=>new Subject()),started=[];
 const subscriptions=sources.map((source,i)=>s.schedule(()=>{started.push(i);return source;},{read:i!==5}).subscribe());
 assert.deepEqual(started,[0,1,2,5]);sources[0].complete();assert.deepEqual(started,[0,1,2,5,3]);subscriptions.forEach(s=>s.unsubscribe());assert.equal(s.snapshot().active,0);
});
test('coalesces identical reads and aborts only after the last consumer leaves',()=>{
 const s=scheduler(),source=new Subject();let starts=0,stops=0;
 const run=()=>{starts++;return new (require('rxjs').Observable)(subscriber=>{const sub=source.subscribe(subscriber);return()=>{stops++;sub.unsubscribe();};});};
 const a=s.schedule(run,{read:true,key:'identity'}).subscribe(),b=s.schedule(run,{read:true,key:'identity'}).subscribe();
 assert.equal(starts,1);a.unsubscribe();assert.equal(stops,0);b.unsubscribe();assert.equal(stops,1);
});
test('caches only approved successes, invalidation fences in-flight results, never caches failures',async()=>{
 let now=0;const s=scheduler({now:()=>now}),p={read:true,key:'same',cacheMs:30_000,cacheable:v=>v.success===true};let starts=0;
 const run=()=>{starts++;return of({success:true});};await firstValueFrom(s.schedule(run,p));now=29_999;await firstValueFrom(s.schedule(run,p));assert.equal(starts,1);now=30_000;await firstValueFrom(s.schedule(run,p));assert.equal(starts,2);
 s.invalidate();const old=new Subject(),pending=firstValueFrom(s.schedule(()=>old,p));s.invalidate();old.next({success:true});old.complete();await pending;await firstValueFrom(s.schedule(run,p));assert.equal(starts,3);
 s.invalidate();await firstValueFrom(s.schedule(()=>of({success:false}),p));await firstValueFrom(s.schedule(run,p));assert.equal(starts,4);
});
test('drops stale queued sessions and enforces one heavy read at a time',async()=>{
 const s=scheduler(),hold=new Subject();let live=true,started=0;const one=s.schedule(()=>hold,{read:true,heavy:true}).subscribe();
 const result=firstValueFrom(s.schedule(()=>{started++;return of(1);},{read:true,heavy:true,valid:()=>live}));live=false;hold.complete();await assert.rejects(result,/会话/);assert.equal(started,0);one.unsubscribe();await flush();assert.equal(s.snapshot().queued,0);
});
test('bounds queue wait without starting timed-out work',async()=>{
 const s=scheduler({queueTimeoutMs:5}),hold=new Subject(),sub=s.schedule(()=>hold,{read:true,heavy:true}).subscribe();
 await assert.rejects(firstValueFrom(s.schedule(()=>of(1),{read:true,heavy:true})),/排队/);sub.unsubscribe();assert.equal(s.snapshot().queued,0);
});
test('fifty concurrent consumers remain bounded and a submission passes a read backlog',async()=>{
 const s=scheduler();let active=0,max=0;const order=[];
 const read=i=>s.schedule(()=>new (require('rxjs').Observable)(subscriber=>{active++;max=Math.max(max,active);const timer=setTimeout(()=>{order.push(i);subscriber.next(i);subscriber.complete();},2);return()=>{clearTimeout(timer);active--;};}),{read:true,key:'r'+i});
 const reads=Array.from({length:50},(_,i)=>firstValueFrom(read(i)));
 const write=firstValueFrom(s.schedule(()=>{order.push('write');return of('saved');},{read:false}));
 assert.equal(await write,'saved');assert.equal(order[0],'write');assert.deepEqual(await Promise.all(reads),Array.from({length:50},(_,i)=>i));assert.ok(max<=3);assert.equal(s.snapshot().active,0);assert.equal(s.snapshot().queued,0);
});