248 lines
27 KiB
TypeScript
248 lines
27 KiB
TypeScript
import { validateDuplicateCheckpoint, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts';
|
|
import type { Change, Id, Json, Result, Target, Point } from '../canvas-core/types.ts';
|
|
import { selectImportGroups } from './import-groups.ts';
|
|
import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
|
|
|
|
export type ImportLine = { readonly cursor: number; readonly nextCursor: number; readonly key: string; readonly value?: Json; readonly error?: string; readonly skip?: boolean };
|
|
export type ImportSource = { readonly fingerprint: string; readonly parserVersion: number; read(cursor: number, signal: AbortSignal): Promise<ImportLine | null> };
|
|
export type ImportUnit = { readonly key: string; readonly dependencies: readonly string[]; readonly changes: readonly Change[]; readonly guard?: Json };
|
|
export type ImportLocator = { readonly key: string; readonly cursor: number; readonly nextCursor: number; readonly dependencies: readonly string[]; readonly digest: string; readonly byteLength: number; readonly changeCount: number };
|
|
export type ImportFailure = { readonly key: string; readonly cursor: number; readonly code: string; readonly dependencies?: readonly string[] };
|
|
export type NativeImportSource = { readonly kind: 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot'; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
|
|
export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id};
|
|
export type ImportJob = {
|
|
readonly version: 1 | 2 | 3; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
|
|
readonly source?: NativeImportSource;
|
|
readonly preparation?: {readonly version:1|2;readonly planner?:DuplicateCheckpoint;readonly registrySignature:string;readonly selection:readonly Target[];readonly worldOffset:Point;readonly allocationIds:readonly Id[];readonly steps:number};
|
|
readonly generation: number; readonly status: 'preparing' | 'paused' | 'running' | 'completed' | 'partial' | 'cancelled'; readonly cursor: number; readonly eof: boolean;
|
|
readonly bindings?: Readonly<Record<Id,ImportBinding>>;
|
|
readonly stableIds: Readonly<Record<string, Id>>; readonly pending: readonly ImportLocator[]; readonly committedSourceKeys: readonly string[]; readonly failures: readonly ImportFailure[]; readonly error?: string;
|
|
};
|
|
export function createImportJob(input: Pick<ImportJob, 'id' | 'projectId' | 'sourceFingerprint' | 'parserVersion' | 'source'>): ImportJob {
|
|
if (![input.id,input.projectId,input.sourceFingerprint].every(value=>typeof value==='string'&&!!value) || !Number.isSafeInteger(input.parserVersion) || input.parserVersion<1) throw fault('invalid_import_job');
|
|
return freeze({ id: input.id, projectId: input.projectId, sourceFingerprint: input.sourceFingerprint, parserVersion: input.parserVersion, ...(input.source?{source:{...clone(input.source),integrityVersion:input.source.integrityVersion??1}}:{}), version: 3, generation: 0, status: 'paused', cursor: 0, eof: false, bindings: {}, stableIds: {}, pending: [], committedSourceKeys: [], failures: [] });
|
|
}
|
|
export function validateImportJob(job: ImportJob): void {
|
|
if(!job||typeof job!=='object'||!validJson(job)||!job.stableIds||typeof job.stableIds!=='object'||Array.isArray(job.stableIds)||![job.pending,job.committedSourceKeys,job.failures].every(Array.isArray)||job.error!==undefined&&typeof job.error!=='string')throw fault('invalid_import_job');
|
|
if (job.version !== 1 && job.version !== 2 && job.version !== 3) throw fault('import_interpreter_unavailable');
|
|
createImportJob(job);
|
|
if (!Number.isSafeInteger(job.generation) || job.generation<0 || !Number.isSafeInteger(job.cursor) || job.cursor<0 || typeof job.eof!=='boolean' || !['preparing','paused','running','completed','partial','cancelled'].includes(job.status)) throw fault('invalid_import_job');
|
|
if(job.source&&(!['native-ndjson','duplicate-ndjson','duplicate-snapshot'].includes(job.source.kind)||job.source.kind==='duplicate-ndjson'&&(!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0||!Array.isArray(job.source.copy.roots)||!job.source.copy.roots.length||job.source.copy.roots.some(id=>typeof id!=='string'||!id))||job.source.name!==undefined&&(typeof job.source.name!=='string'||job.source.name.length>255)||!job.source.fileId||!job.source.targetCanvasId||!Number.isSafeInteger(job.source.byteLength)||job.source.byteLength<0||job.source.targetEntry&&(!job.source.targetEntry.id||!Number.isSafeInteger(job.source.targetEntry.incarnation)||job.source.targetEntry.incarnation<1)))throw fault('invalid_import_source');
|
|
if(job.source){
|
|
if(job.source.integrityVersion!==undefined&&job.source.integrityVersion!==1)throw fault('import_source_integrity_interpreter_unavailable');
|
|
if(job.source.sha256!==undefined&&(typeof job.source.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(job.source.sha256))))throw fault('invalid_import_source');
|
|
if(job.source.integrityVersion===1&&!job.source.sha256&&(job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||(job.preparation?.allocationIds.length??0)>0||(job.preparation?.steps??0)>0))throw fault('import_source_integrity_missing');
|
|
}
|
|
if(job.source?.kind==='duplicate-snapshot'){
|
|
const p=job.preparation;
|
|
if(!p||![1,2].includes(p.version)||typeof p.registrySignature!=='string'||!p.registrySignature||!Array.isArray(p.selection)||!p.selection.length||p.selection.some(t=>!t||typeof t.id!=='string'||!t.id||!Number.isSafeInteger(t.incarnation)||t.incarnation<1)||!p.worldOffset||![p.worldOffset.x,p.worldOffset.y].every(Number.isFinite)||!Array.isArray(p.allocationIds)||p.allocationIds.some(id=>typeof id!=='string'||!id)||new Set(p.allocationIds).size!==p.allocationIds.length||!Number.isSafeInteger(p.steps)||p.steps<0||!['preparing','paused','cancelled'].includes(job.status)||job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0)throw fault('invalid_copy_preparation');
|
|
if(p.planner){validateDuplicateCheckpoint(p.planner);if(p.version!==2||p.planner.sourceToken!==job.source.sha256||p.planner.schemaToken!==p.registrySignature||p.planner.projectId!==job.projectId||!equal(p.planner.selection,p.selection)||!equal(p.planner.offset,p.worldOffset)||p.planner.steps!==p.steps||!equal([...p.planner.elementIds,...p.planner.canvasIds].map(pair=>pair[1]),p.allocationIds))throw fault('invalid_copy_preparation');}
|
|
else if(p.version===2&&(p.steps!==0||p.allocationIds.length))throw fault('invalid_copy_preparation');
|
|
}else if(job.status==='preparing'||job.preparation!==undefined)throw fault('invalid_copy_preparation');
|
|
const keys=new Set<string>(),ids=Object.values(job.stableIds),stableIdSet=new Set(ids);
|
|
if (stableIdSet.size!==ids.length || ids.some(id=>typeof id!=='string'||!id) || Object.keys(job.stableIds).some(key=>!key)) throw fault('invalid_import_job');
|
|
if(job.source?.kind==='duplicate-ndjson'&&(!job.source.copy||new Set(job.source.copy.roots).size!==job.source.copy.roots.length||job.source.copy.roots.some(id=>!stableIdSet.has(id))))throw fault('invalid_import_source');
|
|
for(const[id,binding]of Object.entries(job.bindings??{}))if(!id||!binding||!stableIdSet.has(id)||!['element','canvas'].includes(binding.kind)||binding.kind==='element'&&(!Number.isSafeInteger(binding.incarnation)||binding.incarnation<1)||binding.kind==='canvas'&&binding.entryElementId!==undefined&&typeof binding.entryElementId!=='string')throw fault('invalid_import_job');
|
|
for(const key of job.committedSourceKeys){if(!key||keys.has(key))throw fault('invalid_import_job');keys.add(key);}
|
|
if(job.source)for(const key of job.committedSourceKeys){const id=job.stableIds[key],binding=job.bindings?.[id];if(!Object.hasOwn(job.stableIds,key)||!binding||!key.startsWith(`${binding.kind}:`)||binding.kind==='canvas'&&(!binding.entryElementId||job.bindings?.[binding.entryElementId]?.kind!=='element'))throw fault('invalid_import_job');}
|
|
for(const unit of job.pending){if(!unit||typeof unit.key!=='string'||!unit.key||!Array.isArray(unit.dependencies)||keys.has(unit.key)||!Number.isSafeInteger(unit.cursor)||unit.cursor<0||!Number.isSafeInteger(unit.nextCursor)||unit.nextCursor<=unit.cursor||unit.nextCursor>job.cursor||!(/^[a-f0-9]{64}$/.test(unit.digest))||!Number.isSafeInteger(unit.byteLength)||unit.byteLength<1||!Number.isSafeInteger(unit.changeCount)||unit.changeCount<1||unit.changeCount>10000||new Set(unit.dependencies).size!==unit.dependencies.length||unit.dependencies.some(key=>typeof key!=='string'||!key))throw fault('invalid_import_job');keys.add(unit.key);}
|
|
if(job.status==='partial'&&!job.eof)throw fault('invalid_import_job');
|
|
if(job.status==='completed'&&(!job.eof||job.pending.length||job.failures.length))throw fault('invalid_import_job');
|
|
for(const failure of job.failures)if(!failure||typeof failure.key!=='string'||!failure.key||typeof failure.code!=='string'||!failure.code||!Number.isSafeInteger(failure.cursor)||failure.cursor<0||failure.cursor>job.cursor)throw fault('invalid_import_job');
|
|
}
|
|
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 {
|
|
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;
|
|
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;
|
|
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;
|
|
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'};
|
|
const bytes=new Uint8Array(total);let offset=0;for(const part of parts){bytes.set(part,offset);offset+=part.length;}
|
|
let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(bytes);}catch{return{...base,error:'invalid_utf8'};}
|
|
if(!text.trim())return{...base,skip:true};
|
|
try{const value=JSON.parse(text);return validJson(value)?{...base,value}:{...base,error:'invalid_json_value'};}catch{return{...base,error:'invalid_json'};}
|
|
},
|
|
});
|
|
}
|
|
|
|
export type ImportRunnerPorts={
|
|
source:ImportSource;getJob():ImportJob;createId():Id;
|
|
/** Synchronous compare-and-publish; never write a second private job store. */
|
|
publish(next:ImportJob,expectedGeneration:number):void;
|
|
/** Pure decoding; reserve() returns stable source-key identities. No upload/run effects here. */
|
|
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;
|
|
persist():Promise<void>;onError(error:unknown):void;
|
|
/** Host must durably gate every network write on the same document/job cut. */
|
|
optimisticLocal?:boolean;
|
|
/** Host backpressure for speculative local cuts. Network remains durably gated by persist(). */
|
|
awaitDurabilityBudget?():void|Promise<void>;
|
|
now?:()=>number;yieldTask?:()=>Promise<void>;frameBudgetMs?:number;maxPending?:number;maxPendingBytes?:number;maxBatch?:number;
|
|
};
|
|
|
|
/** Cooperative producer/consumer. Pending work stores seekable locators, not unbounded decoded payloads. */
|
|
export function createImportRunner(ports:ImportRunnerPorts){
|
|
const initial=ports.getJob(),identity={id:initial.id,projectId:initial.projectId,sourceFingerprint:initial.sourceFingerprint,parserVersion:initial.parserVersion},budget=ports.frameBudgetMs??6,maxPending=ports.maxPending??4096,maxBytes=ports.maxPendingBytes??1024*1024,maxBatch=ports.maxBatch??32,now=ports.now??(()=>performance.now());
|
|
validateImportJob(initial);
|
|
if(initial.source?.kind==='duplicate-snapshot')throw fault('import_interpreter_unavailable');
|
|
if(initial.source&&(!initial.source.sha256||initial.source.integrityVersion!==1))throw fault('import_source_integrity_unavailable');
|
|
if(!Number.isFinite(budget)||budget<=0||![maxPending,maxBytes,maxBatch].every(value=>Number.isSafeInteger(value)&&value>0))throw fault('invalid_import_budget');
|
|
let costPerUnit:number|undefined;
|
|
let active:Promise<void>|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set<symbol>();
|
|
const report=(error:unknown)=>{try{ports.onError(error);}catch{/* Reporting cannot change imported content. */}};
|
|
function current(){const job=ports.getJob();if(job.id!==identity.id||job.projectId!==identity.projectId||job.sourceFingerprint!==identity.sourceFingerprint||job.parserVersion!==identity.parserVersion)throw fault('import_identity_changed');return job;}
|
|
function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,version:3 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;return freeze(value);}
|
|
function publish(job:ImportJob,patch:Partial<ImportJob>){const value=next(job,patch);ports.publish(value,job.generation);if(!equal(current(),value))throw fault('invalid_import_publish');return value;}
|
|
function prepare(line:ImportLine,job:ImportJob){
|
|
const ids:Record<string,Id>=Object.assign(Object.create(null),job.stableIds),used=new Set(Object.values(ids));
|
|
const reserve=(key:string)=>{if(typeof key!=='string'||!key)throw fault('invalid_import_key');if(Object.hasOwn(ids,key))return ids[key];const id=ports.createId();if(typeof id!=='string'||!id||used.has(id))throw fault('duplicate_import_identity');ids[key]=id;used.add(id);return id;};
|
|
const unit=ports.prepare(line,job,reserve);
|
|
if(!validJson(unit)||typeof unit.key!=='string'||!unit.key||!Array.isArray(unit.changes)||!Array.isArray(unit.dependencies)||!unit.changes.length||unit.changes.length>10000||new Set(unit.dependencies).size!==unit.dependencies.length||unit.dependencies.some(key=>typeof key!=='string'||!key))throw fault('invalid_import_unit');
|
|
for(const change of unit.changes)if(change.kind!=='field'&&change.before===null&&change.after&&!used.has(change.id))throw fault('unstable_import_identity');
|
|
if(new TextEncoder().encode(JSON.stringify(unit)).byteLength>maxBytes)throw fault('import_unit_capacity');
|
|
return{unit:freeze(clone(unit)),ids:{...ids}};
|
|
}
|
|
const checkpoint=async()=>{if(ports.optimisticLocal&&ports.awaitDurabilityBudget)await ports.awaitDurabilityBudget();else await ports.persist();};
|
|
const digest=async(unit:ImportUnit)=>[...new Uint8Array(await crypto.subtle.digest('SHA-256',new TextEncoder().encode(JSON.stringify(unit))))].map(value=>value.toString(16).padStart(2,'0')).join('');
|
|
function completed(job:ImportJob,units:readonly ImportUnit[],pending:readonly ImportLocator[]){
|
|
const bindings:Record<Id,ImportBinding>=Object.assign(Object.create(null),job.bindings??{});
|
|
for(const unit of units)for(const change of unit.changes)if(change.kind==='element'&&change.before===null&&change.after)bindings[change.id]={kind:'element',incarnation:change.after.incarnation};else if(change.kind==='canvas'&&change.before===null&&change.after)bindings[change.id]={kind:'canvas',...(change.after.parent?{entryElementId:change.after.parent.entryElementId}:{})};
|
|
return next(job,{pending,committedSourceKeys:[...job.committedSourceKeys,...units.map(unit=>unit.key)],bindings:{...bindings}});
|
|
}
|
|
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;
|
|
type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown};
|
|
let reading:{cursor:number;result:Promise<ReadResult>}|undefined;
|
|
const decoded=new Map<string,{unit:ImportUnit;digest:string;bytes:number}>();let decodedBytes=0;
|
|
function cache(key:string,value:{unit:ImportUnit;digest:string;bytes:number}){const previous=decoded.get(key);if(previous)decodedBytes-=previous.bytes;decoded.delete(key);decoded.set(key,value);decodedBytes+=value.bytes;while(decoded.size>maxBatch||decodedBytes>maxBytes){const[key,value]=decoded.entries().next().value!;decoded.delete(key);decodedBytes-=value.bytes;}}
|
|
function forget(key:string){const value=decoded.get(key);if(value){decodedBytes-=value.bytes;decoded.delete(key);}}
|
|
async function read(cursor:number,flushReady:boolean):Promise<ReadResult|{kind:'flush'}>{
|
|
if(reading&&reading.cursor!==cursor)throw fault('invalid_import_cursor');
|
|
reading??={cursor,result:Promise.resolve().then(()=>ports.source.read(cursor,signal)).then(line=>({kind:'line' as const,line}),error=>({kind:'error' as const,error}))};
|
|
let timer:ReturnType<typeof setTimeout>|undefined;
|
|
const result=flushReady?await Promise.race([reading.result,new Promise<{kind:'flush'}>(resolve=>{timer=setTimeout(()=>resolve({kind:'flush'}),0);})]):await reading.result;
|
|
if(timer!==undefined)clearTimeout(timer);if(result.kind!=='flush')reading=undefined;return result;
|
|
}
|
|
if(!ports.optimisticLocal)await ports.persist();
|
|
while(!closed&&!signal.aborted&&!holds.size){
|
|
let job=current();if(job.status!=='running')return;
|
|
if(ports.source.fingerprint!==job.sourceFingerprint||ports.source.parserVersion!==job.parserVersion)throw fault('import_source_mismatch');
|
|
if(didWork&&now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();didWork=false;continue;}
|
|
didWork=true;
|
|
const committed=new Set(job.committedSourceKeys);
|
|
const batchLimit=Math.max(1,Math.min(maxBatch,costPerUnit===undefined?1:Math.floor(budget/(costPerUnit*1.5))));
|
|
const groups=selectImportGroups(job.pending,committed,{units:batchLimit,bytes:maxBytes,changes:10000}),chosen=groups.flat();
|
|
if(chosen.length){
|
|
const units:ImportUnit[]=[];
|
|
for(const locator of chosen){
|
|
const cached=decoded.get(locator.key);
|
|
if(cached&&cached.digest===locator.digest){units.push(cached.unit);continue;}
|
|
// A source need not support concurrent reads. Finish any speculative read before seeking.
|
|
if(reading)await reading.result;
|
|
const line=await ports.source.read(locator.cursor,signal);
|
|
if(closed||signal.aborted||current().generation!==job.generation)return;
|
|
if(!line||line.error||line.skip||line.nextCursor!==locator.nextCursor)throw fault('import_source_changed');
|
|
const value=prepare(line,job);
|
|
if(value.unit.key!==locator.key||!equal(value.unit.dependencies,locator.dependencies)||!equal(value.ids,job.stableIds)||value.unit.changes.length!==locator.changeCount||new TextEncoder().encode(JSON.stringify(value.unit)).byteLength!==locator.byteLength||await digest(value.unit)!==locator.digest)throw fault('import_source_changed');
|
|
if(closed||signal.aborted||current().generation!==job.generation)return;
|
|
units.push(value.unit);
|
|
}
|
|
let proposed=completed(job,units,job.pending.filter(unit=>!chosen.includes(unit)));
|
|
let result=commit(units,job,proposed);
|
|
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]));
|
|
for(const group of groups){
|
|
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);
|
|
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();}
|
|
}
|
|
}else await checkpoint();
|
|
for(const unit of units)forget(unit.key);
|
|
continue;
|
|
}
|
|
if(job.eof){
|
|
// Completion must never overtake the final document/source durability barrier.
|
|
await ports.persist();if(closed||signal.aborted||current().generation!==job.generation)return;
|
|
const unresolved=job.pending.filter(unit=>!job.failures.some(failure=>failure.key===unit.key&&failure.code==='unresolved_dependency')).map(unit=>({key:unit.key,cursor:unit.cursor,code:'unresolved_dependency',dependencies:unit.dependencies.filter(key=>!committed.has(key))}));
|
|
publish(job,{status:job.failures.length||unresolved.length?'partial':'completed',failures:[...job.failures,...unresolved]});await ports.persist();return;
|
|
}
|
|
if(job.pending.length>=maxPending||new TextEncoder().encode(JSON.stringify(job.pending)).byteLength>=maxBytes){publish(job,{status:'paused',error:'dependency_capacity'});await ports.persist();return;}
|
|
// Stage a bounded parser slice, then publish its cursor/IDs/locators once. This is
|
|
// unpublished transaction preparation, never another durable task state store.
|
|
const base=job;let staged=job,records=0;
|
|
while(!staged.eof&&records<maxBatch&&staged.pending.length<maxPending&&new TextEncoder().encode(JSON.stringify(staged.pending)).byteLength<maxBytes){
|
|
const ready=staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)));
|
|
const result=await read(staged.cursor,ready);
|
|
if(closed||signal.aborted||current().generation!==base.generation)return;
|
|
if(result.kind==='flush')break;
|
|
if(result.kind==='error'){
|
|
if(ready){reading={cursor:staged.cursor,result:Promise.resolve(result)};break;}
|
|
if(staged!==base){const{generation,...patch}=staged;publish(base,patch);}throw result.error;
|
|
}
|
|
const line=result.line;
|
|
if(!line){staged={...staged,eof:true};break;}
|
|
if(line.cursor!==staged.cursor||!Number.isSafeInteger(line.nextCursor)||line.nextCursor<=staged.cursor)throw fault('invalid_import_cursor');
|
|
let patch:Partial<ImportJob>;
|
|
if(line.skip)patch={cursor:line.nextCursor};
|
|
else if(line.error)patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:line.error}]};
|
|
else try{
|
|
const prepared=prepare(line,staged),key=prepared.unit.key;
|
|
if(staged.committedSourceKeys.includes(key)||staged.pending.some(unit=>unit.key===key)||staged.failures.some(failure=>failure.key===key))throw fault('duplicate_import_source_key');
|
|
const locator={key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:prepared.unit.dependencies,digest:await digest(prepared.unit),byteLength:new TextEncoder().encode(JSON.stringify(prepared.unit)).byteLength,changeCount:prepared.unit.changes.length};
|
|
cache(key,{unit:prepared.unit,digest:locator.digest,bytes:locator.byteLength});
|
|
patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:[...staged.pending,locator]};
|
|
}catch(error){patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:(error as {code?:string}).code??'import_decode_failed'}]};}
|
|
if(closed||signal.aborted||current().generation!==base.generation)return;
|
|
staged={...staged,...patch};records++;
|
|
// First ready record displays immediately; subsequent slices amortize publication.
|
|
if(costPerUnit===undefined&&staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)))||now()-frame>=budget)break;
|
|
}
|
|
if(staged!==base){const{generation,...patch}=staged;publish(base,patch);if(!ports.optimisticLocal||!staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key))))await checkpoint();}
|
|
|
|
}
|
|
}
|
|
function launch(){
|
|
if(closed||holds.size)return Promise.resolve();if(active){restart=true;return active;}
|
|
restart=false;controller=new AbortController();
|
|
active=pump(controller.signal).catch(async error=>{
|
|
if(closed||controller?.signal.aborted)return;
|
|
try{const job=current();if(['running','completed','partial'].includes(job.status)){publish(job,{status:'paused',error:(error as {code?:string}).code??'import_failed'});try{await ports.persist();}catch{/* Storage may still be unavailable. */}}}catch{/* A replaced identity cannot receive an old error. */}
|
|
report(error);
|
|
}).finally(()=>{active=undefined;controller=undefined;if(restart&&!closed&&!holds.size){try{if(current().status==='running')void launch();}catch{/* Replaced scope stays untouched. */}}});
|
|
return active;
|
|
}
|
|
return Object.freeze({
|
|
start(){if(closed)throw fault('import_runner_closed');const job=current();if(['completed','partial','cancelled'].includes(job.status))return Promise.resolve();if(job.status!=='running')publish(job,{status:'running',error:undefined});return launch();},
|
|
hold(){const token=Symbol();holds.add(token);controller?.abort();return()=>{if(!holds.delete(token)||holds.size||closed)return;if(current().status==='running')void launch();};},
|
|
async pause(){const job=current();if(job.status==='running'){controller?.abort();publish(job,{status:'paused'});await ports.persist();}},
|
|
async cancel(){const job=current();if(['running','paused'].includes(job.status)){controller?.abort();publish(job,{status:'cancelled',error:undefined});await ports.persist();}},
|
|
async settle(){while(active)await active;},
|
|
async close(){closed=true;controller?.abort();while(active)await active;},
|
|
});
|
|
}
|