Slice pending import encoding and reuse completed immutable plans

This commit is contained in:
2026-11-11 04:29:11 +08:00
parent 210bc14074
commit 5eabca5056
5 changed files with 48 additions and 5 deletions
+4 -4
View File
@@ -162,7 +162,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
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:dataArray(freeze([...job.committedSourceKeys,...units.map(unit=>unit.key)])),bindings:dataRecord(freeze({...bindings}))});
return next(job,{pending:dataArray(freeze(pending)),committedSourceKeys:dataArray(freeze([...job.committedSourceKeys,...units.map(unit=>unit.key)])),bindings:dataRecord(freeze({...bindings}))});
}
function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const base=current();if(base.generation!==job.generation)throw fault('stale_import_job');value=freeze(checkpointEvidence(value));const start=now(),result=ports.commit(units,value,job.generation),elapsed=Math.max(0,now()-start);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:base))throw fault('invalid_import_commit');if(result.status==='applied')batches.record(units.length,elapsed);return result;}
async function pump(signal:AbortSignal){
@@ -220,7 +220,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
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=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}))]});}
if(result.status==='rejected'){const code=result.code;publish(job,{pending:dataArray(freeze(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();}
}
@@ -256,9 +256,9 @@ export function createImportRunner(ports:ImportRunnerPorts){
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};
const locator=dataRecord(freeze({key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:dataArray(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]};
patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:dataArray([...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++;
@@ -118,6 +118,11 @@ export async function sealRecoveryImageAsync(formatVersion:ImageVersion,rootRevi
function* encode(plan:Prepared,depth=0,key=''):Generator<Promise<void>,Encoded,void>{
if(plan===null||typeof plan!=='object')return normalize(plan,depth,key);
if('height' in plan)return plan;
// Preparation must capture every unsafe occurrence before yielding. Once an
// earlier occurrence has assembled a proven immutable source, later plans
// can reuse that fact without rebuilding the same JSON containers.
const cached=immutable.get(plan.source);
if(cached){if(depth+cached.height>64)throw fault('invalid_recovery_payload');return cached;}
let height=0,bytes=2,reusable=true,properties=0;
const array=plan.keys===undefined,result:Json[]|Record<string,Json>=array?[]:{};
for(let index=0;index<plan.children.length;index++){