Yield native import scans by elapsed time and bounded work

headless/README.md

@@ -473,3 +473,12 @@ runs in the bubble phase so target handlers can claim it first; it cancels only
 this view's gesture. Style toolbars dismiss their own modes on unclaimed Escape,
 while input controls retain priority. The helper needs no browser globals at
 module load or when called with non-DOM ports.
+
+
+Native import scanners share `ImportScheduling` (`now`, `yieldTask`,
+`frameBudgetMs`, default 6 ms). Array/envelope scanners check elapsed time every
+128 scan steps; NDJSON checks after each bounded chunk. A 16 KiB/step work
+threshold also yields when the clock does not advance. Cancellation is checked
+again after each real task yield. Byte/ordinal cursors, parser versions and source
+fingerprints are unchanged. This is a soft budget: a bounded JSON decode is
+indivisible, and document-envelope range indexing still precedes record display.

src/lib/canvas-runtime/import-budget.ts

@@ -0,0 +1,34 @@
+import {fault} from '../canvas-core/values.ts';
+
+export type ImportScheduling = {
+  now?: () => number;
+  yieldTask?: () => Promise<void>;
+  frameBudgetMs?: number;
+};
+
+/** A soft time slice with a work threshold, even when the clock is coarse.
+ * The caller bounds each indivisible parse/decode operation separately. */
+export function createImportBudget(signal: AbortSignal, options: ImportScheduling = {}) {
+  const budget = options.frameBudgetMs ?? 6;
+  if (!Number.isFinite(budget) || budget <= 0) throw fault('invalid_import_budget');
+  const now = options.now ?? (() => performance.now());
+  const yieldTask = options.yieldTask ?? (() => new Promise<void>(resolve => setTimeout(resolve, 0)));
+  let started = now(), work = 0, unchecked = 0;
+  return {
+    step(units = 1): Promise<void> | undefined {
+      work += units;
+      unchecked += units;
+      if (unchecked < 128 && work < 16384) return;
+      unchecked = 0;
+      if (signal.aborted) throw fault('import_aborted');
+      const elapsed = now() - started;
+      if (work < 16384 && elapsed >= 0 && elapsed < budget) return;
+      return (async () => {
+        await yieldTask();
+        if (signal.aborted) throw fault('import_aborted');
+        started = now();
+        work = 0;
+      })();
+    },
+  };
+}

src/lib/canvas-runtime/import-document.ts

@@ -1,15 +1,16 @@
+import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import type {ImportSource,ImportLine} from './imports.ts';
 import {fault,validJson} from '../canvas-core/values.ts';
 type Range={start:number;end:number};
 /** Index the envelope without decoding element payloads. Original bytes remain the source.
  * Preparation scans O(bytes) and retains O(records) ranges; this is not constant memory. */
-export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:string;signal:AbortSignal;maxRecordBytes?:number;yieldTask?:()=>Promise<void>}):Promise<ImportSource>{
+export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:string;signal:AbortSignal;maxRecordBytes?:number}&ImportScheduling):Promise<ImportSource>{
  const {bytes,signal}=options,max=options.maxRecordBytes??256*1024;
  if(!(bytes instanceof Blob)||bytes.size>64*1024*1024||!options.fingerprint||!Number.isSafeInteger(max)||max<1)throw fault('invalid_import_source');
- let cursor=0,cache={start:-1,bytes:new Uint8Array()},work=0;
- const yieldTask=options.yieldTask??(()=>new Promise<void>(resolve=>setTimeout(resolve,0))),space=(b:number)=>[9,10,13,32].includes(b);
+ let cursor=0,cache={start:-1,bytes:new Uint8Array()};const budget=createImportBudget(signal,options);
+ const space=(b:number)=>[9,10,13,32].includes(b);
  async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor<cache.start||cursor>=cache.start+cache.bytes.length){cache={start:cursor,bytes:new Uint8Array(await bytes.slice(cursor,cursor+16384).arrayBuffer())};if(signal.aborted)throw fault('import_aborted');}return cache.bytes[cursor-cache.start];}
- async function step(){cursor++;if(++work>=16384){work=0;await yieldTask();if(signal.aborted)throw fault('import_aborted');}}
+ function step(){cursor++;return budget.step();}
  async function spaces(){while(space(await peek()))await step();}
  async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();}
  async function value():Promise<Range>{await spaces();const start=cursor,stack:number[]=[];let quoted=false,escaped=false;
@@ -30,8 +31,8 @@ export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:s
  const canvases=fields.get('canvases') as {id:string;parent:null|{canvasId:string;entryElementId:string}}[],ids=new Set<string>();
  for(const canvas of canvases){if(!canvas||typeof canvas.id!=='string'||!canvas.id||ids.has(canvas.id)||canvas.parent!==null&&(!canvas.parent||typeof canvas.parent.canvasId!=='string'||typeof canvas.parent.entryElementId!=='string'))throw fault('invalid_document_import');ids.add(canvas.id);}
  const roots=canvases.filter(canvas=>canvas.parent===null);if(roots.length!==1)throw fault('invalid_document_import');const rootId=roots[0].id;
- const byId=new Map(canvases.map(canvas=>[canvas.id,canvas]));let topologyWork=0;
- for(const canvas of canvases){const seen=new Set<string>();let current=canvas;while(current.parent){if(seen.has(current.id))throw fault('invalid_document_import');seen.add(current.id);if(++topologyWork%128===0){await yieldTask();if(signal.aborted)throw fault('import_aborted');}const parent=byId.get(current.parent.canvasId);if(!parent)throw fault('invalid_document_import');current=parent;}}
+ const byId=new Map(canvases.map(canvas=>[canvas.id,canvas]));
+ for(const canvas of canvases){const seen=new Set<string>();let current=canvas;while(current.parent){if(seen.has(current.id))throw fault('invalid_document_import');seen.add(current.id);await budget.step();const parent=byId.get(current.parent.canvasId);if(!parent)throw fault('invalid_document_import');current=parent;}}
  const tombstones=fields.get('tombstones');if(tombstones!==undefined&&(!tombstones||typeof tombstones!=='object'||Array.isArray(tombstones)||Object.values(tombstones).some(n=>!Number.isSafeInteger(n)||(n as number)<1)))throw fault('invalid_document_import');
  const children=canvases.filter(canvas=>canvas.id!==rootId),count=children.length+elements.length;
  return Object.freeze({fingerprint:options.fingerprint,parserVersion:4,async read(position:number,readSignal:AbortSignal):Promise<ImportLine|null>{

src/lib/canvas-runtime/import-json-array.ts

@@ -1,18 +1,19 @@
+import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import type {ImportSource,ImportLine} from './imports.ts';
 import {fault,validJson} from '../canvas-core/values.ts';
 /** Seekable JSON array reader. Cursors are byte offsets after '[' or a record comma.
  * Structure errors end the source; previously published records remain explicit. */
-export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number;yieldTask?:()=>Promise<void>}):ImportSource{
+export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource{
  const bytes=options.bytes,max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16384;
  if(!(bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source');
- let cache={offset:-1,value:new Uint8Array()};const whitespace=(b:number)=>[9,10,13,32].includes(b),yieldTask=options.yieldTask??(()=>new Promise<void>(resolve=>setTimeout(resolve,0)));
+ let cache={offset:-1,value:new Uint8Array()};const whitespace=(b:number)=>[9,10,13,32].includes(b);
  return Object.freeze({fingerprint:options.fingerprint,parserVersion:3,async read(start:number,signal:AbortSignal):Promise<ImportLine|null>{
   if(!Number.isSafeInteger(start)||start<0||start>bytes.size)throw fault('invalid_import_cursor');
   if(start>0&&start===bytes.size){if(signal.aborted)throw fault('import_aborted');return null;}
   if(bytes.size===0)throw fault('invalid_json_array');
-  let cursor=start,work=0;
+  let cursor=start;const budget=createImportBudget(signal,options);
   async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor<cache.offset||cursor>=cache.offset+cache.value.length){cache={offset:cursor,value:new Uint8Array(await bytes.slice(cursor,cursor+chunk).arrayBuffer())};if(signal.aborted)throw fault('import_aborted');}return cache.value[cursor-cache.offset];}
-  async function advance(){cursor++;if(++work>=16384){work=0;await yieldTask();if(signal.aborted)throw fault('import_aborted');}}
+  function advance(){cursor++;return budget.step();}
   async function spaces(){while(whitespace(await peek()))await advance();}
   const failure=(error:string):ImportLine=>({cursor:start,nextCursor:bytes.size,key:`byte:${start}`,error});
   await spaces();if(start===0){if(await peek()!==91)return failure('invalid_json_array');await advance();await spaces();}

src/lib/canvas-runtime/imports.ts

@@ -1,3 +1,4 @@
+import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import {validateImportFileRetention,type FileRetirement,type PreviousImportFile} from './import-file-retention.ts';
 import { validateDuplicateCheckpoint, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts';
 import type { Change, Id, Json, Result, Target, Point, Availability } from '../canvas-core/types.ts';
@@ -66,25 +67,24 @@ export function validateImportJob(job: ImportJob): void {
 const yieldTask=()=>new Promise<void>(resolve=>setTimeout(resolve,0));
 
 /** Seekable bounded NDJSON parser. Fingerprint must identify the host's immutable, durable byte source. */
-export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number;yieldTask?:()=>Promise<void>}):ImportSource {
+export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource {
   const max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16*1024;
   if(!(options.bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source');
-  const sourceBytes=options.bytes,sourceYield=options.yieldTask??yieldTask;
+  const sourceBytes=options.bytes;
   let cache={offset:-1,bytes:new Uint8Array()};
   return Object.freeze({fingerprint:options.fingerprint,parserVersion:1,
     async read(cursor:number,signal:AbortSignal):Promise<ImportLine|null>{
       if(!Number.isSafeInteger(cursor)||cursor<0||cursor>sourceBytes.size)throw fault('invalid_import_cursor');
       if(signal.aborted)throw fault('import_aborted');if(cursor===sourceBytes.size)return null;
-      const parts:Uint8Array[]=[],start=cursor;let total=0,oversized=false,scanned=0;
+      const parts:Uint8Array[]=[],start=cursor,budget=createImportBudget(signal,options);let total=0,oversized=false;
       while(cursor<sourceBytes.size){
         if(signal.aborted)throw fault('import_aborted');
         if(cursor<cache.offset||cursor>=cache.offset+cache.bytes.length)cache={offset:cursor,bytes:new Uint8Array(await sourceBytes.slice(cursor,cursor+chunk).arrayBuffer())};
         if(signal.aborted)throw fault('import_aborted');
         const begin=cursor-cache.offset,newline=cache.bytes.indexOf(10,begin),end=newline<0?cache.bytes.length:newline,part=cache.bytes.subarray(begin,end);total+=part.length;
         if(total>max){oversized=true;parts.length=0;}else if(!oversized)parts.push(part);
-        cursor=cache.offset+end+(newline<0?0:1);scanned+=end-begin;
+        cursor=cache.offset+end+(newline<0?0:1);await budget.step(end-begin);
         if(newline>=0||cursor===sourceBytes.size)break;
-        if(scanned>=64*1024){scanned=0;await sourceYield();}
       }
       const base={cursor:start,nextCursor:cursor,key:`byte:${start}`};
       if(oversized)return{...base,error:'record_too_large'};

src/lib/canvas-runtime/index.ts

@@ -33,3 +33,5 @@ export type {CanvasElementCounts,ElementOwners} from './scope-counts.ts';
 export {createAssetDepot} from './asset-depot.ts';
 export type {AssetDepot,AssetAuthority,SharedAsset,SharedAssetBinding,AssetDepotPort,DepotCut,DepotCollection} from './asset-depot.ts';
 export {createJsonArrayImportSource} from './import-json-array.ts';
+
+export type {ImportScheduling} from './import-budget.ts';

tests/headless/import-document.test.mjs

@@ -8,3 +8,4 @@ import {prepareNativeImport} from '../../src/lib/canvas-runtime/import-native.ts
 import {createImportJob} from '../../src/lib/canvas-runtime/imports.ts';
 import {createRegistry,createDocument} from '../../src/lib/canvas-core/index.ts';
 test('document root aliases also remap component-declared canvas references without reserving a duplicate root',()=>{const registry=createRegistry([{id:'ref',dataVersion:1,structure:'node',features:[],validate:()=>true,copy:{references:[{path:['canvasRef'],target:'canvas',external:'reject'}],omit:[]}}]),target=createDocument('p',[{id:'target',parent:null}],[]),job=createImportJob({id:'job',projectId:'p',parserVersion:4,sourceFingerprint:'file',source:{kind:'native-document',fileId:'file',byteLength:100,targetCanvasId:'target'}}),reserved=[],record={id:'a',incarnation:1,ownerCanvasId:'source',componentId:'ref',dataVersion:1,structure:'node',geometry:{x:0,y:0,width:10,height:10,rotation:0},data:{canvasRef:'source'}},unit=prepareNativeImport({cursor:0,nextCursor:1,key:'record:0',value:{kind:'element',key:'a',record,sourceRootId:'source'}},job,key=>{reserved.push(key);return'new-'+key;},target,registry);assert.equal(unit.changes[0].after.data.canvasRef,'target');assert.equal(unit.changes[0].after.ownerCanvasId,'target');assert.ok(!reserved.includes('canvas:source'));});
+test('document preparation yields by elapsed time even for files below the byte ceiling',async()=>{const controller=new AbortController();let clock=0,yields=0;await assert.rejects(openDocumentImportSource({bytes:source({...document,elements:[{id:'slow',text:'x'.repeat(2048)}]}),fingerprint:'slow',signal:controller.signal,now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}}),/import_aborted/);assert.equal(yields,1);});

tests/headless/import-json-array.test.mjs

@@ -5,4 +5,5 @@ async function all(reader){let cursor=0;const rows=[];for(let i=0;i<100;i++){con
 test('array reader handles UTF8, escaped strings and nested delimiters across one-byte chunks and resumes from the same byte cursor',async()=>{const text=' [ {"title":"图\\\"片,]","nested":[1,{"x":"]"}]},\n {"title":"second"} ] ',reader=source(text,{chunkBytes:1}),rows=await all(reader);assert.deepEqual(rows.map(r=>r.value),JSON.parse(text));assert.ok(rows.every(r=>!r.error));const reopened=source(text,{chunkBytes:3});assert.deepEqual(await reopened.read(rows[1].cursor,signal()),rows[1]);assert.equal(await reopened.read(rows[1].nextCursor,signal()),null);});
 test('structurally malformed tails report failure after the already accepted prefix',async()=>{for(const tail of [',]',',','','] garbage']){const rows=await all(source('[{"id":1},{"id":2}'+tail));assert.deepEqual(rows[0].value,{id:1});assert.ok(rows.at(-1).error,tail);}assert.deepEqual(await all(source('[]')),[]);await assert.rejects(all(source('')),/invalid_json_array/);assert.equal((await all(source('[1,]')))[0].error,'invalid_json_array');});
 test('oversized records skip bounded payload and allow a later valid record; bad UTF8 remains explicit',async()=>{const rows=await all(source(JSON.stringify(['x'.repeat(200),{ok:true}]),{maxRecordBytes:32,chunkBytes:8}));assert.equal(rows[0].error,'record_too_large');assert.deepEqual(rows[1].value,{ok:true});const reader=createJsonArrayImportSource({bytes:new Blob([new Uint8Array([91,34,255,34,93])]),fingerprint:'bad'});assert.equal((await all(reader))[0].error,'invalid_utf8');});
-test('long scans yield and honor abort without publishing a completed record',async()=>{const controller=new AbortController();let yields=0;const reader=source(JSON.stringify(['x'.repeat(40000)]),{yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(reader.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);await assert.rejects(reader.read(-1,signal()),/invalid_import_cursor/);});
+test('long scans yield and honor abort without publishing a completed record',async()=>{const controller=new AbortController();let yields=0;const reader=source(JSON.stringify(['x'.repeat(40000)]),{now:()=>0,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(reader.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);await assert.rejects(reader.read(-1,signal()),/invalid_import_cursor/);});
+test('slow array scans yield by elapsed time before the fixed byte ceiling and abort at that boundary',async()=>{const controller=new AbortController();let clock=0,yields=0;const reader=source(JSON.stringify(['x'.repeat(2048)]),{now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(reader.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);});

tests/headless/imports.test.mjs

@@ -141,3 +141,5 @@ test('a ready parser publication observes host backpressure before admitting its
 test('async commit preparation cannot publish while waiting and cancellation fences a late successful result',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'image'}],{ports:{async beforeCommit(units,job,signal){assert.equal(f.root.document.elements.size,0);assert.ok(job.pending.length);entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;assert.equal(f.commits.length,0);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.cancel();gate.resolve();await running;await f.runner.settle();assert.equal(f.root.job.status,'cancelled');assert.equal(f.commits.length,0);assert.equal(f.root.document.elements.size,0);await f.runner.close();});
 test('async preparation denial isolates independent groups without publishing the denied record',async()=>{const prepared=[],f=fixture([{id:'denied'},{id:'allowed'}],{ports:{async beforeCommit(units){prepared.push(units.map(unit=>unit.key));return units.some(unit=>unit.key==='denied')?{enabled:false,code:'asset_forbidden',targets:[]}:{enabled:true};}}});await f.runner.start();assert.equal(f.root.job.status,'partial');assert.ok(f.root.job.failures.some(row=>row.key==='denied'&&row.code==='asset_forbidden'));assert.equal(f.root.document.elements.size,1);assert.equal([...f.root.document.elements.values()][0].data.title,'allowed');assert.ok(prepared.length>=2);assert.equal(f.errors.length,0);await f.runner.close();});
 test('a generation change during async preparation prevents the old attempt from committing or changing progress',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'later'}],{ports:{async beforeCommit(){entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;f.root={...f.root,job:{...f.root.job,generation:f.root.job.generation+1}};const generation=f.root.job.generation;gate.resolve();await running;assert.equal(f.commits.length,0);assert.equal(f.root.job.generation,generation);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,1);await f.runner.close();});
+
+test('NDJSON checks elapsed time before returning a short scanned record and honors cancellation',async()=>{const controller=new AbortController();let clock=0,yields=0;const source=createNdjsonImportSource({bytes:new Blob([JSON.stringify({text:'x'.repeat(2048)})+'\n']),fingerprint:'timed',chunkBytes:256,now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(source.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);});