Negotiate and decode snapshot-bound canonical model streams
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {createRegistry,builtinComponents} from '../../src/lib/canvas-core/index.ts';
|
||||
import {decodeModelStream,modelStreamType} from '../../src/lib/canvas-service/model-stream.ts';
|
||||
import {createFoundryTransport} from '../../src/lib/canvas-service/foundry.ts';
|
||||
import {createFoundryModelLoader} from '../../src/lib/canvas-service/model-loader.ts';
|
||||
import {createScopedLoader} from '../../src/lib/canvas-runtime/loader.ts';
|
||||
const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000002',account='10000000-0000-4000-8000-000000000003';
|
||||
const registry=createRegistry(builtinComponents()),signal=()=>new AbortController().signal;
|
||||
const node={id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',data:{title:'中文',text:'🙂'},geometry:{x:0,y:0,width:100,height:80,rotation:0}};
|
||||
const records=()=>[{streamVersion:1,snapshotToken:'snapshot-a',sequence:0,kind:'header',documentFormatVersion:1,projectId:root,canvases:[{id:root,parent:null}],versions:{[root]:3},tombstones:{old:2},removedCanvasIds:[],elementCount:1},{streamVersion:1,snapshotToken:'snapshot-a',sequence:1,kind:'elements',elements:[structuredClone(node)]},{streamVersion:1,snapshotToken:'snapshot-a',sequence:2,kind:'complete',elementCount:1}];
|
||||
const encode=rows=>new TextEncoder().encode(rows.map(row=>JSON.stringify(row)).join('\n')+'\n');
|
||||
async function* source(rows){yield encode(rows);}
|
||||
const collect=async iterable=>{const result=[];for await(const item of iterable)result.push(item);return result;};
|
||||
const scope={accountId:account,workspaceId:workspace,projectId:null,canvasId:root,authEpoch:1,navigationGeneration:1,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry'};
|
||||
|
||||
test('stream decoder frames split UTF8/newlines, emits deltas and validates one final complete model',async()=>{
|
||||
const bytes=encode(records());async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);}
|
||||
const chunks=await collect(decodeModelStream(split(),root,registry,signal()));
|
||||
assert.deepEqual(chunks.map(chunk=>[chunk.sequence,chunk.complete,chunk.value.status]),[[0,false,'receiving'],[1,false,'receiving'],[2,true,'ready']]);
|
||||
assert.equal(chunks[1].value.elements[0].data.text,'🙂');assert.equal(chunks[2].value.document.elements.get('a').data.title,'中文');assert.equal(chunks[2].value.document.tombstones.get('old'),2);assert.equal(chunks.reduce((sum,chunk)=>sum+chunk.byteLength,0),bytes.length);assert.ok(Object.isFrozen(chunks[1].value.elements[0]));
|
||||
});
|
||||
test('snapshot switches, sequence gaps, duplicate IDs, wrong counts, late records and invalid topology cannot finish',async()=>{
|
||||
const cases=[rows=>rows[1].snapshotToken='other',rows=>rows[1].sequence=2,rows=>rows[0].elementCount=2,rows=>rows[2].elementCount=2,rows=>rows.push({...rows[2],sequence:3}),rows=>rows[1].elements.push(rows[1].elements[0]),rows=>rows[1].elements[0].ownerCanvasId=workspace,rows=>rows[1].elements[0].geometry.width=-1,rows=>rows[0].versions={},rows=>rows[0].canvases[0].parent={canvasId:root,entryElementId:'a'}];
|
||||
for(const mutate of cases){const rows=records();mutate(rows);await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));}
|
||||
for(const rows of [[],records().slice(0,1),records().slice(0,2)])await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));
|
||||
async function* truncated(){yield encode(records()).slice(0,-1);}await assert.rejects(()=>collect(decodeModelStream(truncated(),root,registry,signal())),{code:'incomplete_model_stream'});
|
||||
async function* huge(){yield new Uint8Array(8*1024*1024+1).fill(65);}await assert.rejects(()=>collect(decodeModelStream(huge(),root,registry,signal())),{code:'model_stream_budget'});
|
||||
});
|
||||
test('Foundry transport negotiates streaming without buffering; coordinator retains first delta before gated tail',async()=>{
|
||||
let release,opened;const gate=new Promise(resolve=>release=resolve),started=new Promise(resolve=>opened=resolve);let requested;
|
||||
const rows=records(),stream=new ReadableStream({async start(controller){controller.enqueue(encode(rows.slice(0,2)));opened();await gate;controller.enqueue(encode(rows.slice(2)));controller.close();}});
|
||||
const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>true,fetch:async(_url,input)=>{requested=input;return new Response(stream,{headers:{'Content-Type':modelStreamType}});}});
|
||||
const seen=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport}),current:()=>true});
|
||||
const handle=loader.acquire(scope,{onUpdate:snapshot=>seen.push(snapshot)});try{
|
||||
await started;while(!seen.some(snapshot=>snapshot.chunks.length===2))await new Promise(resolve=>setImmediate(resolve));
|
||||
assert.equal(requested.headers.Accept,modelStreamType);assert.equal(handle.getSnapshot().status,'partial');assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving');
|
||||
release();const done=await handle.ready;assert.equal(done.status,'complete');assert.equal(done.chunks.at(-1).value.document.elements.size,1);
|
||||
}finally{release();handle.release();loader.close();}
|
||||
});
|
||||
test('malformed tails retain read deltas without yielding an authoritative model',async()=>{
|
||||
async function* broken(){yield encode(records().slice(0,2));yield new TextEncoder().encode('{broken}\n');}
|
||||
const states=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:broken()})}),current:()=>true});const handle=loader.acquire(scope,{onUpdate:value=>states.push(value)});
|
||||
try{await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');assert.equal(handle.getSnapshot().chunks.length,2);assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving');}finally{handle.release();loader.close();}
|
||||
});
|
||||
test('terminal record still requires EOF; trailing garbage rejects rather than opening a baseline',async()=>{
|
||||
const gate={},rows=records();let release;gate.promise=new Promise(resolve=>release=resolve);
|
||||
async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');}
|
||||
const loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:trailing()})}),current:()=>true}),handle=loader.acquire(scope);let ready=false;handle.ready.then(()=>ready=true,()=>{});
|
||||
try{while(handle.getSnapshot().chunks.length<2)await new Promise(resolve=>setImmediate(resolve));assert.equal(ready,false);release();await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');}finally{release();handle.release();loader.close();}
|
||||
});
|
||||
test('stream transport fences changed identity and abort cancels a blocked body reader',async()=>{
|
||||
let current=true,cancelled=0;const abort=new AbortController();const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>current,fetch:async()=>new Response(new ReadableStream({cancel(){cancelled++;}}),{headers:{'Content-Type':modelStreamType}})});
|
||||
const reply=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1',signal:abort.signal});const iterator=reply.stream[Symbol.asyncIterator]();const pending=iterator.next();await Promise.resolve();abort.abort();await assert.rejects(pending);assert.equal(cancelled,1);
|
||||
const other=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1'});current=false;await assert.rejects(()=>other.stream[Symbol.asyncIterator]().next(),{code:'identity_changed'});assert.equal(cancelled,2);
|
||||
});
|
||||
@@ -10,9 +10,12 @@ import { setTimeout as pause } from 'node:timers/promises';
|
||||
const source=resolve(import.meta.dirname,'../..'),directory=await mkdtemp(join(tmpdir(),'endoleap-proxy-check-')),received=[];
|
||||
const account='10000000-0000-4000-8000-000000000004',key='90000000-0000-4000-8000-000000000002';
|
||||
const responseBytes=' { "protocol_version":2, "request_id":"'+key+'", "status":"not-applied", "terminal":true, "error":"fixture_rejection", "extension":{"retained":true} }\n';
|
||||
let streamRelease;const streamGate=new Promise(resolve=>streamRelease=resolve);
|
||||
const streamPrefix='{"streamVersion":1,"kind":"header"}\n',streamTail='{"streamVersion":1,"kind":"complete"}\n';
|
||||
const upstream=createServer(async(request,response)=>{
|
||||
const chunks=[];for await(const chunk of request)chunks.push(chunk);
|
||||
received.push({path:request.url,body:Buffer.concat(chunks),headers:request.headers});
|
||||
if(request.url.endsWith('/model')&&request.headers.accept==='application/vnd.foundry.canvas-model+ndjson;version=1'){response.writeHead(200,{'Content-Type':'application/vnd.foundry.canvas-model+ndjson;version=1'});response.write(streamPrefix);await streamGate;response.end(streamTail);return;}
|
||||
response.writeHead(request.url.endsWith('/model/operations')?409:200,{'Content-Type':'application/json'});
|
||||
response.end(request.url.endsWith('/model/operations')?responseBytes:'{"id":"fixture"}');
|
||||
});
|
||||
@@ -60,10 +63,15 @@ try{
|
||||
const aiEvents=await fetch(`http://127.0.0.1:${port}/api/foundry/${aiPath}/${key}?after_seq=41&limit=100`);await aiEvents.text();assert.equal(aiEvents.status,200);
|
||||
assert.ok(received.some(item=>item.path.endsWith('/ai/runs/'+key+'?limit=100&after_seq=41')));
|
||||
console.log('PASS frozen AI bytes, identity, version header, admission receipt and event cursor cross the actual Next proxy');
|
||||
const streamed=await fetch(`http://127.0.0.1:${port}/api/foundry/${path.replace('/operations','')}`,{headers:{Accept:'application/vnd.foundry.canvas-model+ndjson;version=1'}});
|
||||
assert.equal(streamed.headers.get('content-type'),'application/vnd.foundry.canvas-model+ndjson;version=1');
|
||||
const reader=streamed.body.getReader(),first=await reader.read();assert.equal(new TextDecoder().decode(first.value),streamPrefix);streamRelease();let rest='';while(true){const next=await reader.read();if(next.done)break;rest+=new TextDecoder().decode(next.value);}assert.equal(rest,streamTail);reader.releaseLock();
|
||||
console.log('PASS negotiated canonical NDJSON and first chunk cross the actual Next proxy before the gated tail');
|
||||
const before=received.length,invalid=await fetch(`http://127.0.0.1:${port}/api/foundry/${path}/unexpected/extra`,{signal:AbortSignal.timeout(10000)});await invalid.text();assert.equal(invalid.status,404);assert.equal(received.length,before);
|
||||
console.log('PASS actual Next proxy preserves frozen request bytes/key/account and raw terminal receipt above 1 MiB');
|
||||
console.log('PASS canonical route allowlist rejects unexpected paths without contacting the upstream');
|
||||
}finally{
|
||||
streamRelease();
|
||||
if(next?.pid){const exited=new Promise(resolve=>{if(next.exitCode!==null)resolve();else next.once('exit',resolve)});try{process.kill(-next.pid,'SIGTERM')}catch{}await Promise.race([exited,pause(3000)]);if(next.exitCode===null){try{process.kill(-next.pid,'SIGKILL')}catch{}}}
|
||||
upstream.closeAllConnections();await new Promise(resolve=>upstream.close(resolve));await rm(directory,{recursive:true,force:true,maxRetries:5});
|
||||
}
|
||||
|
||||
@@ -39,6 +39,14 @@ const upstream = createServer(async (request, response) => {
|
||||
if (path.endsWith('/materials')) return reply(200, { items: boards, next_offset: null });
|
||||
const board = boards.find(board => path.endsWith('/materials/' + board.id)); if (board) return reply(200, board);
|
||||
const read = path.match(/\/materials\/([^/]+)\/canvas\/model$/);
|
||||
if(read&&request.headers.accept==='application/vnd.foundry.canvas-model+ndjson;version=1'){
|
||||
const value=model(documents.get(read[1])),token='fixture-snapshot-'+calls.length;let sequence=0;
|
||||
response.writeHead(200,{'Content-Type':'application/vnd.foundry.canvas-model+ndjson;version=1'});
|
||||
const line=record=>response.write(JSON.stringify({streamVersion:1,snapshotToken:token,sequence:sequence++,...record})+'\n');
|
||||
line({kind:'header',documentFormatVersion:1,projectId:value.projectId,canvases:value.canvases,versions:value.versions,tombstones:value.tombstones,removedCanvasIds:value.removedCanvasIds,elementCount:value.elements.length});
|
||||
for(let offset=0;offset<value.elements.length;offset+=64)line({kind:'elements',elements:value.elements.slice(offset,offset+64)});
|
||||
line({kind:'complete',elementCount:value.elements.length});return response.end();
|
||||
}
|
||||
if (read) return reply(200, { model: model(documents.get(read[1])) });
|
||||
const write = path.match(/\/materials\/([^/]+)\/canvas\/model\/operations$/);
|
||||
if (write) {
|
||||
@@ -324,6 +332,8 @@ try {
|
||||
|
||||
await click('账户和工作空间');await until(`!!document.querySelector('#workspace-account:popover-open')`,'account menu did not open');await evaluate(`[...document.querySelectorAll('button')].find(button=>button.textContent.trim()==='退出登录').click()`);
|
||||
await until(`location.pathname==='/login'&&!document.body.innerText.includes('This page couldn’t load')`,'logout did not safely detach the view');
|
||||
assert.ok(calls.some(call=>call.path.endsWith('/canvas/model')&&call.headers.accept==='application/vnd.foundry.canvas-model+ndjson;version=1'));
|
||||
console.log('PASS actual GraphPage negotiates and assembles canonical NDJSON through the real Next proxy');
|
||||
console.log('PASS logout retires the runtime before dropping its view without a render error');
|
||||
}
|
||||
} finally {
|
||||
|
||||
Reference in New Issue
Block a user