diff --git a/headless/README.md b/headless/README.md index a3c3169..ef10de8 100644 --- a/headless/README.md +++ b/headless/README.md @@ -355,3 +355,11 @@ same indexed bytes. Document jobs use recovery v15 (readers accept 1–15, rejec report `import_resource_authorization_required`; other records can import with an explicit partial result. This is not complete asset round-trip support. Preparation still scans the envelope before progressive payload decoding and insertion. + +ImportRunner hosts may provide async `beforeCommit(units, job, signal)` for read-only +preparation such as resource permission checks. It returns Availability; a denial +uses normal independent-group failure isolation. A successful result still must +pass cancellation/hold/generation/status checks before the synchronous commit. +No document, history or committed source progress advances while it waits. This +hook alone does not authorize or adopt exported image references; that integration +is still pending. diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index b8b610c..3a10578 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -1,6 +1,6 @@ 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 } from '../canvas-core/types.ts'; +import type { Change, Id, Json, Result, Target, Point, Availability } from '../canvas-core/types.ts'; import { selectImportGroups } from './import-groups.ts'; import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts'; @@ -104,6 +104,8 @@ export type ImportRunnerPorts={ prepare(line:ImportLine,job:ImportJob,reserve:(sourceKey:string)=>Id):ImportUnit; /** Publish all document changes AND next job in the same StateRoot, or publish nothing. */ commit(units:readonly ImportUnit[],next:ImportJob,expectedGeneration:number):Result; + /** Optional async, read-only preparation; commit remains synchronous and generation-fenced. */ + beforeCommit?(units:readonly ImportUnit[],job:ImportJob,signal:AbortSignal):Promise; persist():Promise;onError(error:unknown):void; /** Host must durably gate every network write on the same document/job cut. */ optimisticLocal?:boolean; @@ -144,6 +146,13 @@ export function createImportRunner(ports:ImportRunnerPorts){ function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const start=now(),result=ports.commit(units,value,job.generation),cost=Math.max(.01,(now()-start)/units.length);costPerUnit=costPerUnit===undefined?cost:Math.max(cost,costPerUnit*.8+cost*.2);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:job))throw fault('invalid_import_commit');return result;} async function pump(signal:AbortSignal){ let frame=now(),didWork=false; + async function attempt(units:readonly ImportUnit[],job:ImportJob,proposed:ImportJob):Promise{ + let availability:Availability={enabled:true}; + if(ports.beforeCommit)try{availability=await ports.beforeCommit(units,job,signal);}catch(error){availability={enabled:false,code:(error as {code?:string}).code??'import_preparation_failed',targets:[]};} + if(closed||signal.aborted||holds.size||current().generation!==job.generation||current().status!=='running')return; + if(!availability.enabled)return{status:'rejected',code:availability.code??'import_preparation_failed',targets:availability.targets??[]}; + return commit(units,job,proposed); + } type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown}; let reading:{cursor:number;result:Promise}|undefined; const decoded=new Map();let decodedBytes=0; @@ -181,7 +190,7 @@ export function createImportRunner(ports:ImportRunnerPorts){ units.push(value.unit); } let proposed=completed(job,units,job.pending.filter(unit=>!chosen.includes(unit))); - let result=commit(units,job,proposed); + let result=await attempt(units,job,proposed);if(!result)return; if(result.status==='rejected'){ // Reject a reference cycle as a whole; isolate only independent SCCs. const byKey=new Map(units.map(unit=>[unit.key,unit])); @@ -189,7 +198,7 @@ export function createImportRunner(ports:ImportRunnerPorts){ job=current();const internal=new Set(group.map(row=>row.key)),members=group.map(row=>byKey.get(row.key)!); if(!members.every(unit=>unit.dependencies.every(key=>internal.has(key)||job.committedSourceKeys.includes(key))))continue; proposed=completed(job,members,job.pending.filter(value=>!internal.has(value.key))); - result=commit(members,job,proposed); + result=await attempt(members,job,proposed);if(!result)return; if(result.status==='rejected'){const code=result.code;publish(job,{pending:job.pending.filter(value=>!internal.has(value.key)),failures:[...job.failures,...group.map(row=>({key:row.key,cursor:row.cursor,code}))]});} await checkpoint();if(closed||signal.aborted||current().status!=='running')return; if(now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();} diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs index 0557d1d..0732bbe 100644 --- a/tests/headless/imports.test.mjs +++ b/tests/headless/imports.test.mjs @@ -137,3 +137,7 @@ test('resuming a failed partial-status write does not duplicate unresolved depen test('a ready parser publication observes host backpressure before admitting its content root',async()=>{ const gate=deferred(),entered=deferred();let blocked=true;const f=fixture([{id:'one'},{id:'two'}],{ports:{optimisticLocal:true,maxBatch:1,awaitDurabilityBudget(stage){if(blocked){assert.equal(stage,'parsed');entered.resolve();return gate.promise;}}}});const running=f.runner.start();await entered.promise;assert.equal(f.root.job.pending.length,1);assert.equal(f.root.job.committedSourceKeys.length,0);assert.equal(f.root.document.elements.size,0);await ticks();assert.equal(f.root.document.elements.size,0,'ready content bypassed host budget');blocked=false;gate.resolve();await running;assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,2);assert.equal(f.errors.length,0);await f.runner.close(); }); + +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();});