37 lines
3.3 KiB
JavaScript
37 lines
3.3 KiB
JavaScript
/**
|
|
* 运行任务轮询器的暂停与终止测试。hold 只阻止下一次轮询，允许已发出的响应更新并保存游标。
|
|
* 可控 Promise 与假时钟分别控制响应和后续调度；这里不连接网络或真实存储。
|
|
*/
|
|
import { test } from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { createRunTracker } from '../../src/lib/canvas-runtime/runs.ts';
|
|
|
|
// 推进有限轮微任务，让轮询链执行；不会推进假时钟或主动触发未来的定时器。
|
|
const tick = async () => { for (let index=0;index<30;index++)await Promise.resolve(); };
|
|
// 由测试决定首个轮询何时返回，用来把 hold 放在请求已发出但响应尚未完成的位置。
|
|
const deferred = () => { let resolve;const promise=new Promise(yes=>{resolve=yes;});return{promise,resolve}; };
|
|
// 搭建一个运行中的任务：首轮响应可挂起，后续响应直接完成，保存记录用于核对落盘游标。
|
|
function fixture() {
|
|
let now=0,next=0;const timers=new Map(),runs=new Map([['run',{id:'run',status:'running',cursor:0,hasMore:true,attempt:0}]]),calls=[],saved=[],response=deferred();
|
|
const tracker=createRunTracker({clock:{now:()=>now,setTimer(callback,delay){const id=++next;timers.set(id,{callback,due:now+delay});return id;},clearTimer:id=>timers.delete(id)},getRuns:()=>runs,async persist(){saved.push(structuredClone([...runs]));},async poll(run){calls.push(run);return calls.length===1?response.promise:{status:'succeeded',events:[],next:2,hasMore:false,outputs:[],steps:[]};},accept(run,page){runs.set(run.id,{...runs.get(run.id),status:page.status,cursor:page.next,hasMore:page.hasMore});},update(id,reduce){runs.set(id,reduce(runs.get(id)));},onError(error){throw error;}});
|
|
return{tracker,runs,calls,saved,response,timers,async advance(ms){now+=ms;for(const[id,timer]of[...timers])if(timer.due<=now){timers.delete(id);timer.callback();}await tick();}};
|
|
}
|
|
|
|
// 两个 hold 共同阻止新轮询；在途响应仍保存 cursor=1，重复释放同一 hold 无效，最后释放后从该游标继续。
|
|
test('run holds preserve a completed cursor and notifications without launching another poll',async()=>{
|
|
const f=fixture();f.tracker.notify(['run']);await f.advance(0);assert.equal(f.calls.length,1);
|
|
const releaseA=f.tracker.hold(),releaseB=f.tracker.hold();
|
|
f.response.resolve({status:'running',events:[],next:1,hasMore:true,outputs:[],steps:[]});await f.tracker.settle();
|
|
assert.equal(f.runs.get('run').cursor,1);assert.equal(f.saved.at(-1)[0][1].cursor,1);assert.equal(f.timers.size,0);
|
|
await f.advance(10000);await f.tracker.retry();assert.equal(f.calls.length,1);assert.equal(f.timers.size,0);
|
|
releaseA();releaseA();await f.advance(0);assert.equal(f.calls.length,1);
|
|
releaseB();await f.advance(0);await f.tracker.settle();assert.equal(f.calls.length,2);assert.equal(f.calls[1].cursor,1);assert.equal(f.runs.get('run').cursor,2);
|
|
f.tracker.stop();
|
|
});
|
|
|
|
// stop 是终态：即使先前有待处理通知，随后释放 hold 也不能恢复定时器或请求，关闭后不再允许申请 hold。
|
|
test('releasing a run hold after stop cannot restart a poll',async()=>{
|
|
const f=fixture(),release=f.tracker.hold();f.tracker.notify(['run']);f.tracker.stop();release();await f.advance(10000);
|
|
assert.equal(f.calls.length,0);assert.equal(f.timers.size,0);assert.throws(()=>f.tracker.hold(),/run_tracker_closed/);
|
|
});
|