tricu

io-v2-host-work.test.mjs (6677B)

---

 import test from 'node:test';
 import assert from 'node:assert/strict';
 import {Tasks} from '../dist/io-v2-tasks.js';
 import {HostWork} from '../dist/io-v2-host-work.js';
 const ok=value=>({kind:'completed',value}),failed=value=>({kind:'failed',value});
 const finish=(tasks,scope)=>{tasks.bodyDone(scope,ok(null));assert(tasks.beginRelease(scope));return tasks.settle(scope);};
 const deferred=()=>{let resolve;const promise=new Promise(done=>{resolve=done;});return {promise,resolve};};
 const setup=()=>{const tasks=new Tasks(1n,'owner-stopped','cycle');return {tasks,work:new HostWork(tasks)};};
 
 test('pending read never blocks sibling work; cancellation waits for actual provider cleanup',async()=>{
  const {tasks,work}=setup(),read=tasks.spawn(0,true,true),writer=tasks.spawn(0,true,false),cleanup=deferred(),events=[];
  const opaque=new Proxy({},{get(){throw Error('late value forced');}});
  work.start(read.key,async signal=>{
   events.push('read started');
   await new Promise(resolve=>signal.addEventListener('abort',()=>{events.push('abort requested');resolve();},{once:true}));
   await cleanup.promise;events.push('read cleanup finished');return opaque;
  });
  work.start(writer.key,()=>{events.push('sibling wrote');return 7;});
  await work.wait();assert.equal(work.take().outcome.value,7);finish(tasks,writer.key);
  tasks.cancel(0,read,'stop');work.cancelStopped();work.cancelStopped();
  tasks.bodyDone(read.key,ok(null));assert.equal(tasks.beginRelease(read.key),false);
  await Promise.resolve();assert.deepEqual(events,['read started','sibling wrote','abort requested']);
  const settled=work.wait();cleanup.resolve();await settled;
  assert.equal(work.take(),null);assert.deepEqual(work.counts,{pending:0,replies:0});
  assert(tasks.beginRelease(read.key));tasks.settle(read.key);
  tasks.claimJoin(0,read);assert.equal(tasks.pollJoin(0).kind,'cancelled');finish(tasks,0);
  assert.equal(events.at(-1),'read cleanup finished');assert.deepEqual(tasks.counts,{scopes:0,tasks:0,hosts:0});
 });
 test('completion queued before cancellation is discarded and cannot revive the body',async()=>{
  const {tasks,work}=setup(),child=tasks.spawn(0,true,false),value={retained:'reply'};
  work.start(child.key,()=>value);await work.wait();assert.deepEqual([...work.values()],[value]);
  tasks.cancel(0,child,'stop');work.cancelStopped();assert.deepEqual([...work.values()],[]);assert.equal(work.take(),null);
  finish(tasks,child.key);finish(tasks,0);assert.deepEqual(work.counts,{pending:0,replies:0});
 });
 test('queued data for an already settled scope is dropped; no scope history required',async()=>{
  const {tasks,work}=setup(),child=tasks.spawn(0,true,false);
  work.start(child.key,()=>42);await work.wait();finish(tasks,child.key);
  work.cancelStopped();assert.equal(work.take(),null);assert.deepEqual(work.counts,{pending:0,replies:0});finish(tasks,0);
 });
 test('synchronous throws and async rejections settle ownership once without inspecting errors',async()=>{
  for(const asynchronous of [false,true]){
   const {tasks,work}=setup(),error=new Proxy({},{get(){throw Error('opaque error forced');}});
   work.start(0,()=>{if(asynchronous)return Promise.reject(error);throw error;});
   await work.wait();const reply=work.take();assert.equal(reply.outcome.ok,false);assert.equal(reply.outcome.error,error);
   assert.equal(work.take(),null);assert.equal(tasks.state(0).hostCalls,0);finish(tasks,0);
  }
 });
 test('read rejection after cancellation is acknowledgement, not a fresh task failure',async()=>{
  const {tasks,work}=setup(),child=tasks.spawn(0,true,false);
  work.start(child.key,signal=>new Promise((_resolve,reject)=>signal.addEventListener('abort',()=>reject(Error('abort')),{once:true})));
  tasks.stop(child.key,failed('original'));work.cancelStopped();tasks.bodyDone(child.key,ok(null));
  await work.wait();assert.equal(work.take(),null);assert(tasks.beginRelease(child.key));assert.equal(tasks.settle(child.key).value,'original');
  finish(tasks,0);
 });
 test('multiple waiters share notification; out-of-order replies have correct identities',async()=>{
  const {tasks,work}=setup(),a=deferred(),b=deferred();
  const first=work.start(0,()=>a.promise),second=work.start(0,()=>b.promise);
  const one=work.wait(),two=work.wait();assert.equal(one,two);b.resolve('second');await one;
  assert.deepEqual(work.take(),{token:second,outcome:{ok:true,value:'second'}});
  const next=work.wait();a.resolve('first');await next;
  assert.deepEqual(work.take(),{token:first,outcome:{ok:true,value:'first'}});finish(tasks,0);
 });
 test('sequential host operations retain neither callback history nor consumed reply data',async()=>{
  const {tasks,work}=setup();
  for(let i=0;i<1000;i++){
   const value={iteration:i};work.start(0,()=>value);await work.wait();assert.equal(work.take().outcome.value,value);
   assert.deepEqual(work.counts,{pending:0,replies:0});assert.equal(tasks.state(0).hostCalls,0);assert.deepEqual([...work.values()],[]);
  }
  finish(tasks,0);
 });
 test('async physical release survives new cancellation and reports cleanup failure',async()=>{
  const {tasks,work}=setup(),release=deferred();
  assert.throws(()=>work.release(0,()=>null),/invalid-transition/);
  tasks.bodyDone(0,ok('body'));assert(tasks.beginRelease(0));
  const token=work.release(0,async()=>{await release.promise;throw 'cleanup failed';});
  assert.equal(token.kind,'release');assert.throws(()=>tasks.settle(0),/invalid-transition/);
  tasks.stop(0,{kind:'cancelled',value:'primary'});work.cancelStopped();
  assert.deepEqual(tasks.takeCancellations(),[]);assert.throws(()=>tasks.settle(0),/invalid-transition/);
  const waiting=work.wait();release.resolve();await waiting;
  const reply=work.take();assert.equal(reply.outcome.ok,false);assert.equal(reply.outcome.error,'cleanup failed');
  const result=tasks.settle(0,{items:[reply.outcome.error],omitted:0});
  assert.equal(result.kind,'cancelled');assert.equal(result.value,'primary');assert.deepEqual(result.cleanup.items,['cleanup failed']);
  assert.deepEqual(work.counts,{pending:0,replies:0});
 });
 test('body callback identity exhaustion cannot prevent the reserved cleanup batch',async()=>{
  const tasks=new Tasks(1n,'stop','cycle',1),work=new HostWork(tasks);
  work.start(0,()=>1);await work.wait();work.take();
  assert.throws(()=>work.start(0,()=>2),/identity-exhausted/);
  tasks.bodyDone(0,ok(null));assert(tasks.beginRelease(0));
  const release=work.release(0,()=>3);await work.wait();assert.equal(work.take().outcome.value,3);
  assert.equal(release.kind,'release');assert.equal(tasks.endHost(release),'stale');
  assert.throws(()=>work.release(0,()=>4),/invalid-transition/);tasks.settle(0);
  assert.deepEqual(tasks.counts,{scopes:0,tasks:0,hosts:0});
 });