tricu

io-session.test.mjs (5955B)

---

 import test from 'node:test';
 import assert from 'node:assert/strict';
 import {Graph} from '../dist/graph.js';
 import {readText,readNatural,readList} from '../dist/codecs.js';
 import {constructors} from '../dist/io-private.js';
 import {Wire} from '../dist/io-core.js';
 import {Session} from '../dist/io-session.js';
 function fixture(budget={}){
  const g=new Graph(budget),t=constructors({leaf:g.leaf,observe:g.observe.bind(g),stem:g.stem.bind(g),fork:g.fork.bind(g),app:g.app.bind(g)}),w=new Wire(t,1n);
  const n=x=>t.natural(BigInt(x)),a=(tag,p=t.leaf)=>t.fork(n(tag),p),request=(group,op,p=t.leaf)=>a(10,t.fork(t.text(group),t.fork(n(1),t.fork(t.text(op),p))));
  const ref=key=>w.marker('io.v2.resource',key),read=request('stream','read',t.fork(ref(1),n(4))),write=bytes=>request('stream','write',t.fork(ref(2),w.list(bytes.map(n))));
  const seq=(left,right)=>a(1,t.fork(left,t.fork(t.leaf,right))),session=action=>new Session(t,action,'1',roots=>g.collect([...roots]));
  return {g,t,w,n,a,request,ref,read,write,seq,session};
 }
 async function poll(s,predicate){for(let i=0;i<100;i++){const r=await s.turn({run:'1',kind:'poll'});if(predicate(r))return r;}throw Error('poll did not progress');}
 const value=(f,s)=>f.w.pair(f.w.pair(s.core.result.value)[1])[0];
 test('stream read uses the independent native data-pair encoding, not a nested tuple',async()=>{
  const f=fixture(),s=f.session(f.read),r=await poll(s,r=>r.events.length);const event=r.events[0];
  await s.turn({run:'1',kind:'reply',reply:{...event,bytes:[65,0,255],eof:false}});if(!s.core.result)await poll(s,r=>r.done);
  const [tag,bytes]=f.w.pair(value(f,s));assert.equal(readText(f.g,tag),'data');assert.deepEqual(readList(f.g,bytes,3).map(h=>readNatural(f.g,h)),[65n,0n,255n]);
 });
 test('EOF is sticky and standard handles are borrowed, not physically closed',async()=>{
  const f=fixture(),s=f.session(f.seq(f.read,f.read)),r=await poll(s,r=>r.events.length);
  let out=await s.turn({run:'1',kind:'reply',reply:{...r.events[0],bytes:[],eof:true}});
  if(!out.done)out=await poll(s,r=>{assert.equal(r.events.length,0);return r.done;});
  assert.equal(readText(f.g,readList(f.g,value(f,s),1)[0]),'eof');
  const other=fixture(),close=other.session(other.request('resource','close',other.ref(2)));await poll(close,r=>r.done);
  assert.equal(readText(other.g,readList(other.g,value(other,close),4)[0]),'borrowed-resource');
 });
 test('structured session stop waits for provider acknowledgement and refuses reuse',async()=>{
  const f=fixture(),s=f.session(f.read),r=await poll(s,r=>r.events.length),event=r.events[0];
  await assert.rejects(s.turn({run:'2',kind:'stop'}),/stale/);
  const stopped=await s.turn({run:'1',kind:'stop'});assert(!stopped.done);assert.equal(stopped.events[0].kind,'cancel');
  let done=await s.turn({run:'1',kind:'reply',reply:{...event,kind:'cancelled'}});if(!done.done)done=await poll(s,r=>r.done);
  assert.equal(done.outcome.kind,'cancelled');assert(Object.values(done.counts).every(c=>Object.values(c).every(n=>n===0)));
  await assert.rejects(s.turn({run:'1',kind:'poll'}),/stale/);
 });
 test('failed output poisons the slot before demanding any later byte payload',async()=>{
  const f=fixture(),poison=f.g.app(f.g.leaf,f.g.leaf),observe=f.t.observe;
  f.t.observe=h=>{if(h===poison)throw Error('forced poisoned write data');return observe(h);};
  const later=f.request('stream','write',f.t.fork(f.ref(2),poison)),s=f.session(f.seq(f.write([65]),later)),r=await poll(s,r=>r.events.length);
  await s.turn({run:'1',kind:'reply',reply:{...r.events[0],kind:'error'}});if(!s.core.result)await poll(s,r=>r.done);
  assert.equal(s.core.hasFault,false);const error=readList(f.g,value(f,s),4);assert.equal(readText(f.g,error[0]),'invalid-state');assert.equal(readText(f.g,error[3]),'unknown');
 });
 test('attempted output has an aggregate bound and never silently retries partial writes',async()=>{
  const f=fixture({fuel:20000000});let action=f.write([1]);for(let i=0;i<4;i++)action=f.seq(f.write(Array(4096).fill(65)),action);
  const s=f.session(action);let events=0;
  for(let i=0;i<100&&!s.core.result;i++){
   const r=await s.turn({run:'1',kind:'poll'});
   for(const event of r.events){events++;await s.turn({run:'1',kind:'reply',reply:{...event,kind:'write',count:1}});}
  }
  assert(s.core.result);assert.equal(events,4);assert.equal(readText(f.g,readList(f.g,value(f,s),4)[0]),'invalid-bound');
 });
 test('named UTF8 facilities preserve nonfinal carry and generic BOM text',async()=>{
  const f=fixture(),bytes=[0xef,0xbb,0xbf,65,0xf0,0x9f],s=f.session(f.request('utf8','decode',f.w.list([f.n(0),f.w.list(bytes.map(f.n))])));
  await poll(s,r=>r.done);const [text,carry]=readList(f.g,value(f,s),2);
  assert.equal(readText(f.g,text),'\ufeffA');assert.deepEqual(readList(f.g,carry,2).map(h=>readNatural(f.g,h)),[240n,159n]);
  const other=fixture(),encoded=other.session(other.request('utf8','encode',other.t.text('λ😀')));await poll(encoded,r=>r.done);
  assert.deepEqual(readList(other.g,value(other,encoded),6).map(h=>Number(readNatural(other.g,h))),[206,187,240,159,152,128]);
 });
 test('named UTF8 rejects excess heads/tails without demand and malformed scalar values as ordinary errors',async()=>{
  for(const [op,limit]of [['encode',1024],['decode',4096]]){
   const f=fixture({fuel:20000000}),poison=f.g.app(f.g.leaf,f.g.leaf),observe=f.t.observe;
   f.t.observe=h=>{if(h===poison)throw Error('forced excess payload');return observe(h);};
   let list=f.t.fork(poison,poison);for(let i=0;i<limit;i++)list=f.t.fork(f.n(65),list);
   const s=f.session(f.request('utf8',op,op==='encode'?list:f.w.list([f.n(0),list])));await poll(s,r=>r.done);
   assert(!s.core.hasFault);assert.equal(readText(f.g,readList(f.g,value(f,s),4)[0]),'invalid-argument');
  }
  for(const scalar of [0xd800,0x110000]){
   const f=fixture(),s=f.session(f.request('utf8','encode',f.w.list([f.n(scalar)])));await poll(s,r=>r.done);
   assert.equal(readText(f.g,readList(f.g,value(f,s),4)[0]),'invalid-argument');
  }
 });