Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Synchronize certified canvas scopes while preserving the original loading snapshot

Browse Source
This commit is contained in:
Codex Bot
2026-11-09 04:02:22 +08:00
parent cabf2ec667
commit 6c0a421f59
19 changed files with 199 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/api.ts
+1
View File
@@ -65,6 +65,7 @@ export const errorText = (error: unknown) => {
model_stream_budget: "画布数据量超过当前加载上限，请联系支持。",
load_snapshot_changed: "加载期间画布快照发生变化，请重新加载。",
invalid_model_coverage: "画布的加载范围信息不一致，请重新加载。",
loading_waiting_for_receipts: "正在确认已发送的修改，请稍后重试。",
model_loading: "画布尚在加载，请等待加载完成后编辑。",
identity_changed: "账户已变化，当前画布已停止同步，请重新打开。",
engine_closing: "画布正在关闭，本地修改会保留用于恢复。",
src/lib/canvas-runtime/contracts.ts
+2 -1
View File
@@ -8,6 +8,7 @@ import type { ResourceRecord } from './resources.ts';
import type { RecoveryAudit } from './conflicts.ts';
export type RuntimeState = {
readonly loadingRead?:CanonicalProposal;
readonly loading?:import('./loading.ts').LoadingState;
readonly orphanFiles?:ReadonlyMap<Id,OrphanFile>;
readonly imports?: ReadonlyMap<Id, ImportJob>;
@@ -25,7 +26,7 @@ export type EngineSnapshot = Snapshot<RuntimeState>;
export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap<Id, number> };
/** Receiving records are read deltas, not a validated Document or an ACK base. */
export type CanonicalReadUpdate = CanonicalReadResult | { readonly status:'receiving'; readonly canvasElementCounts?:import('./scope-counts.ts').CanvasElementCounts; readonly projectId:Id; readonly canvases:readonly CanvasRecord[]; readonly elements:readonly ElementRecord[]; readonly tombstones:ReadonlyMap<Id,number>; readonly versions:ReadonlyMap<Id,number>; readonly receivedElements:number; readonly totalElements:number };
export type CanonicalReadResult = ({ readonly status: 'ready' } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap<Id, number> };
export type CanonicalReadResult = ({ readonly status: 'ready';readonly sourceToken?:string } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap<Id, number> };
/** A validated historical execution receipt; its old model is never published over a fresh read. */
export type AppliedModelProof = { readonly versions:ReadonlyMap<Id,number>; assertCanonical(proposal:CanonicalProposal):void };
export type PersistencePort = {
src/lib/canvas-runtime/engine.ts
+23 -18
View File
@@ -1,4 +1,4 @@
import {loadingQueryCoverage,assertLoadingChanges,loadingCommand,validateLoadingRuntime,validatePrefixExtension,validateLoadingCompletion,type LoadingState} from './loading.ts';
import {loadingCanSend,mergeLoadingRead,loadingQueryCoverage,assertLoadingChanges,loadingCommand,validateLoadingRuntime,validatePrefixExtension,validateLoadingCompletion,type LoadingState} from './loading.ts';
import {validImageDimensions,type ImageDecoderPort} from './image-decoder.ts';
import {unitScale} from '../canvas-core/units.ts';
import { validateCreationSizeEffect } from '../canvas-core/creation-size-plan.ts';
@@ -98,14 +98,14 @@ export async function openProjectEngine(options: EngineOptions) {
let recovery: { controller: AbortController; release(): void; cancel():Promise<void> } | undefined;
const admissionPins=(snapshot:EngineSnapshot)=>{const pins=new Map<Id,readonly Id[]>();if(copyPreparations.size)pins.set('copy-admission',[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt===undefined).map(resource=>resource.id));if(imageTasks.size)pins.set('image-decode',[...imageTasks.keys()]);return pins;};
const assertWritable = (complete=true) => { if(options.isCurrentIdentity&&!options.isCurrentIdentity())throw fault('identity_changed');if(complete&&editor.getSnapshot().runtime.loading)throw fault('model_loading');if(closing||closed)throw fault('engine_closing');if(recovery)throw fault('recovery_in_progress'); };
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ imports:readMap(new Map()), recoveries:readMap(new Map()), runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: openingVersions, ...(openingLoading?{loading:openingLoading}:{}), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ imports:readMap(new Map()), recoveries:readMap(new Map()), runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: openingVersions, ...(openingLoading?{loading:openingLoading,...(openingLoading.version===3?{loadingRead:Object.freeze({document:createDocument(options.document.projectId,[...options.document.canvases.values()],[...options.document.elements.values()],options.document.tombstones),versions:openingVersions})}:{})}:{}), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
validateLoadingRuntime(initial,options.registry);
const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit,queryBudget:options.queryBudget,queryCoverage:loadingQueryCoverage,
policy(command, document) { if(options.isCurrentIdentity&&!options.isCurrentIdentity())return{enabled:false,code:'identity_changed',targets:[]};if (closing || closed || recovery) return { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] }; if(editor.getSnapshot().runtime.loading){const allowed=loadingCommand(command,document,options.registry,editor.getSnapshot().runtime.loading,editor.getSnapshot().runtime.acknowledged);if(!allowed.enabled)return allowed;}return options.policy?.(command, document) ?? { enabled: true }; },
traversalPolicy(entry, direction) { if(options.isCurrentIdentity&&!options.isCurrentIdentity())return{enabled:false,code:'identity_changed',targets:[]};if(closing||closed||recovery)return{enabled:false,code:closing||closed?'engine_closing':'recovery_in_progress',targets:[]};if(editor.getSnapshot().runtime.loading){try{assertLoadingChanges(entry.changes,editor.getSnapshot().runtime.acknowledged,options.registry,editor.getSnapshot().runtime.loading);}catch(error){return{enabled:false,code:(error as Error).message,targets:[]};}}return options.traversalPolicy?.(entry,direction)??{enabled:true}; },
onListenerError: report,
prepareRuntime:(previous,next)=>{
if(!acceptingLoading&&!recovery&&(previous.runtime.loading||next.runtime.loading)&&(!equal(previous.runtime.loading,next.runtime.loading)||previous.runtime.acknowledged!==next.runtime.acknowledged||!equal(Object.fromEntries(previous.runtime.versions),Object.fromEntries(next.runtime.versions))))throw fault('loading_state_owned');
if(!acceptingLoading&&!recovery&&(previous.runtime.loading||next.runtime.loading||previous.runtime.loadingRead||next.runtime.loadingRead)&&(!equal(previous.runtime.loading,next.runtime.loading)||previous.runtime.loadingRead!==next.runtime.loadingRead||previous.runtime.acknowledged!==next.runtime.acknowledged||!equal(Object.fromEntries(previous.runtime.versions),Object.fromEntries(next.runtime.versions))))throw fault('loading_state_owned');
validateLoadingRuntime(next.runtime,options.registry);
const sizing=prepareDefaultImageSizing(previous,next);
const runtime=prepareResourceRetention({...next,runtime:sizing},options.registry,displayPins,collectionReservations,admissionPins(next));
@@ -171,7 +171,7 @@ export async function openProjectEngine(options: EngineOptions) {
}
persistenceFailed = false;
})();
writing = task.catch(error => { persistenceFailed = true; throw error; }).finally(() => { writing = undefined;if(!persistenceFailed&&loadingRelease&&loadingCompletionRevision!==undefined&&durableRootRevision>=loadingCompletionRevision&&!editor.getSnapshot().runtime.loading){const release=loadingRelease;loadingRelease=undefined;loadingCompletionRevision=undefined;release();} });
writing = task.catch(error => { persistenceFailed = true; throw error; }).finally(() => { writing = undefined;if(!persistenceFailed){if(loadingCompletionRevision!==undefined&&durableRootRevision>=loadingCompletionRevision&&!editor.getSnapshot().runtime.loading){const release=loadingRelease;loadingRelease=undefined;loadingCompletionRevision=undefined;release?.();}if(!closing&&!closed)scheduler.refreshAdmission();} });
return writing;
}
function collectResources(settings:{retentionMs:number;maxFiles?:number}) {
@@ -383,7 +383,7 @@ export async function openProjectEngine(options: EngineOptions) {
}
if(action){const run=runs.get(action.runId);if(!run)throw fault('run_identity_mismatch');runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,appliedOutputIds:[...new Set([...(run.appliedOutputIds??[]),action.outputId!])],hasMore:true,blocked:false,error:undefined,pendingPage:undefined,nextPollAt:runClock.now()}));}
const runtime: RuntimeState = Object.freeze({ ...previous, runs: readMap(runs), ledger: ledger.getSnapshot(), acknowledged, versions: readMap(proposal.versions), receipts: readMap(receipts), conflicts: readMap(conflicts) });
const result = editor.reconcile(document, runtime, root.rootRevision);
const oldAccepting=acceptingLoading;acceptingLoading=oldAccepting||previous.loading?.version===3;let result:Result;try{result=editor.reconcile(document, runtime, root.rootRevision);}finally{acceptingLoading=oldAccepting;}
if (result.status === 'rejected') throw fault(result.code, result.targets);
if(action)queueMicrotask(()=>runTracker.notify([action.runId]));
} catch (error) {
@@ -395,7 +395,7 @@ export async function openProjectEngine(options: EngineOptions) {
report(error); return { blocked: true };
}
}
const scheduler = createScheduler({ getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist,
const scheduler = createScheduler({ admit:intent=>{const root=editor.getSnapshot();if(root.runtime.loading&&root.runtime.loading.version!==3)return true;if(closing||closed||loadingCompletionRevision!==undefined&&durableRootRevision<loadingCompletionRevision)return false;return loadingCanSend(root.runtime,operationCanvases(intent,root.document))&&(!root.runtime.loading||durableRootRevision>=root.rootRevision);},getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist,
createId: options.createId ?? (() => crypto.randomUUID()),
participants: intent => { if(intent.historyEffect)validateCreationSizeEffect(intent.historyEffect);return operationCanvases(intent, editor.getSnapshot().document); },
compile(intent, id) {
@@ -424,14 +424,16 @@ export async function openProjectEngine(options: EngineOptions) {
query: (request, signal) => network.run(signal, () => ['run.start','run.cancel','run.apply'].includes(request.kind??'') ? options.runs?.query(request,signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' }) : request.kind === 'asset.upload' ? options.assets?.query(request, signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' }) : options.persistence.query(request, signal)),
supportsRequest: request => ['run.start','run.cancel','run.apply'].includes(request.kind??'') ? !!options.runs && request.protocolVersion === 2 : request.kind === 'asset.upload' ? !!options.assets && request.protocolVersion === 2 : options.persistence.supportsRequest?.(request) ?? request.protocolVersion === 1, complete, onError: report, clock: options.clock,
});
if(initial.loading)loadingRelease=scheduler.hold();
if(initial.loading&&initial.loading.version!==3)loadingRelease=scheduler.hold();
function completeLoading(document:Document,versions:ReadonlyMap<Id,number>,settings:{fresh?:boolean;sourceToken?:string}={}):Result{
try{assertWritable(false);const root=editor.getSnapshot(),previous=root.runtime;if(!previous.loading)throw fault('loading_not_active');
validateLoadingCompletion(previous.acknowledged,previous.versions,previous.loading,document,versions,options.registry,settings.fresh,settings.sourceToken);
const acknowledged=createDocument(document.projectId,[...document.canvases.values()],[...document.elements.values()],document.tombstones),working=replayPending(acknowledged,previous.ledger,options.registry),conflicts=new Map(previous.conflicts);conflicts.delete('loading');conflicts.delete('refresh');
acceptingLoading=true;const result=editor.reconcile(working,Object.freeze({...previous,loading:undefined,acknowledged,versions:readMap(versions),conflicts:readMap(conflicts)}),root.rootRevision);
if(settings.fresh&&previous.loading.version===3&&[...previous.ledger.requests.values()].some(request=>request.phase!=='resolved'))throw fault('loading_waiting_for_receipts');
const source=!settings.fresh&&previous.loading.version===3?previous.loadingRead:undefined;validateLoadingCompletion(source?.document??previous.acknowledged,source?.versions??previous.versions,previous.loading,document,versions,options.registry,settings.fresh,settings.sourceToken);
const proposal=!settings.fresh?mergeLoadingRead(previous,{document,versions},options.registry):{document,versions};
const acknowledged=createDocument(document.projectId,[...proposal.document.canvases.values()],[...proposal.document.elements.values()],proposal.document.tombstones),working=replayPending(acknowledged,previous.ledger,options.registry),conflicts=new Map(previous.conflicts);conflicts.delete('loading');conflicts.delete('refresh');
acceptingLoading=true;const result=editor.reconcile(working,Object.freeze({...previous,loading:undefined,loadingRead:undefined,acknowledged,versions:readMap(proposal.versions),conflicts:readMap(conflicts)}),root.rootRevision);
if(result.status==='applied')loadingCompletionRevision=editor.getSnapshot().rootRevision;return result;
}catch(error){const result=reject(error);if(editor.getSnapshot().runtime.loading&&!['engine_closing','identity_changed','recovery_in_progress','loading_not_active'].includes(result.code))editor.updateRuntime(previous=>{const conflicts=new Map(previous.conflicts);conflicts.set('loading',result.code);return Object.freeze({...previous,conflicts:readMap(conflicts)});});return result;}finally{acceptingLoading=false;}
}catch(error){const result=reject(error);if(editor.getSnapshot().runtime.loading&&!['engine_closing','identity_changed','recovery_in_progress','loading_not_active','loading_waiting_for_receipts'].includes(result.code))editor.updateRuntime(previous=>{const conflicts=new Map(previous.conflicts);conflicts.set('loading',result.code);return Object.freeze({...previous,conflicts:readMap(conflicts)});});return result;}finally{acceptingLoading=false;}
}
function updateRun(id: Id, reduce: (run: RunRecord) => RunRecord) {
const result = editor.updateRuntime(previous => { const current = previous.runs.get(id);if(!current)throw fault('run_missing');const runs = new Map(previous.runs);runs.set(id,freeze(reduce(current)));return Object.freeze({...previous,runs:readMap(runs)}); });
@@ -707,12 +709,15 @@ export async function openProjectEngine(options: EngineOptions) {
loading:Object.freeze({getState:()=>editor.getSnapshot().runtime.loading,
resume():Promise<Result>{
if(loadingRead)return loadingRead;
const task=(async()=>{assertWritable(false);if(!editor.getSnapshot().runtime.loading)return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:[]};if(!options.persistence.read)throw fault('canonical_reader_unavailable');const controller=new AbortController();loadingReadController=controller;
try{const read=await network.run(controller.signal,()=>options.persistence.read!(controller.signal));assertWritable(false);controller.signal.throwIfAborted();if(!editor.getSnapshot().runtime.loading)return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:[]};const result=completeLoading(read.document,read.versions,{fresh:true});if(result.status==='applied')await persist();return result;}finally{if(loadingReadController===controller)loadingReadController=undefined;}
const task=(async()=>{assertWritable(false);if(!editor.getSnapshot().runtime.loading)return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:[]};if(!options.persistence.read)throw fault('canonical_reader_unavailable');const controller=new AbortController();loadingReadController=controller;let releaseRead:(()=>void)|undefined;
try{if(editor.getSnapshot().runtime.loading?.version===3)await new Promise<void>((resolve,reject)=>{
let unsubscribe=()=>{};const finish=(error?:unknown)=>{unsubscribe();controller.signal.removeEventListener('abort',abort);error===undefined?resolve():reject(error);},abort=()=>finish(controller.signal.reason),check=()=>{if(![...editor.getSnapshot().runtime.ledger.requests.values()].some(request=>request.phase!=='resolved')){releaseRead=scheduler.hold();finish();}};
unsubscribe=editor.subscribe(check);controller.signal.addEventListener('abort',abort,{once:true});if(controller.signal.aborted)abort();else check();
});const read=await network.run(controller.signal,()=>options.persistence.read!(controller.signal));assertWritable(false);controller.signal.throwIfAborted();if(!editor.getSnapshot().runtime.loading)return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:[]};const result=completeLoading(read.document,read.versions,{fresh:true});if(result.status==='applied')await persist();return result;}finally{releaseRead?.();if(loadingReadController===controller)loadingReadController=undefined;}
})();loadingRead=task.finally(()=>{if(loadingRead===wrapped)loadingRead=undefined;});const wrapped=loadingRead;return wrapped;
},
accept(document:Document,versions:ReadonlyMap<Id,number>,state:LoadingState):Result{
try{assertWritable(false);const root=editor.getSnapshot(),previous=root.runtime;if(!previous.loading)throw fault('loading_not_active');validatePrefixExtension(previous.acknowledged,previous.versions,previous.loading,document,versions,state,options.registry);const acknowledged=createDocument(document.projectId,[...document.canvases.values()],[...document.elements.values()],document.tombstones),working=replayPending(acknowledged,previous.ledger,options.registry);acceptingLoading=true;return editor.reconcile(working,Object.freeze({...previous,loading:freeze(clone(state)),acknowledged}),root.rootRevision);}catch(error){return reject(error);}finally{acceptingLoading=false;}
try{assertWritable(false);const root=editor.getSnapshot(),previous=root.runtime;if(!previous.loading)throw fault('loading_not_active');const source=previous.loadingRead;validatePrefixExtension(source?.document??previous.acknowledged,source?.versions??previous.versions,previous.loading,document,versions,state,options.registry);const nextRead=Object.freeze({document:createDocument(document.projectId,[...document.canvases.values()],[...document.elements.values()],document.tombstones),versions:readMap(versions)}),proposal=mergeLoadingRead(previous,nextRead,options.registry),acknowledged=proposal.document,working=replayPending(acknowledged,previous.ledger,options.registry);acceptingLoading=true;return editor.reconcile(working,Object.freeze({...previous,loading:freeze(clone(state)),...(state.version===3?{loadingRead:nextRead}:{}),versions:readMap(proposal.versions),acknowledged}),root.rootRevision);}catch(error){return reject(error);}finally{acceptingLoading=false;}
},
async complete(document:Document,versions:ReadonlyMap<Id,number>,settings:{fresh?:boolean;sourceToken?:string}={}):Promise<Result>{const result=completeLoading(document,versions,settings);if(result.status==='applied')await persist();return result;},
}),
@@ -743,7 +748,7 @@ export async function openProjectEngine(options: EngineOptions) {
try{working=replayPending(acknowledged,previous.ledger,options.registry);}catch(error){const rejected=reject(error);conflict=freeze({code:rejected.code,targets:rejected.targets});}
let committed:ConflictPlan|'refresh'|undefined,committedIds:readonly Id[]=[];
const plans=new WeakSet<ConflictPlan>();
const finish=async()=>{if(loadingRelease&&!editor.getSnapshot().runtime.loading)loadingCompletionRevision=editor.getSnapshot().rootRevision;await persist();assertCurrent();await scheduler.resume();assertCurrent();scheduler.notify(committedIds);runTracker.notify(staged.runIds);active.release();return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:committedIds};};
const finish=async()=>{if(previous.loading&&!editor.getSnapshot().runtime.loading)loadingCompletionRevision=editor.getSnapshot().rootRevision;await persist();assertCurrent();await scheduler.resume();assertCurrent();scheduler.notify(committedIds);runTracker.notify(staged.runIds);active.release();return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:committedIds};};
active.cancel=async()=>{if(recovery!==active)return;if(committed){await finish();return;}active.release();};
return Object.freeze({baseRootRevision:root.rootRevision,canonical,acknowledgedRequestIds:staged.requestIds,
pendingOperationIds:Object.freeze([...previous.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)),
@@ -760,7 +765,7 @@ export async function openProjectEngine(options: EngineOptions) {
if(intent.operationType==='run.apply'){const run=previous.runs.get(intent.action.runId);if(!run||run.blocked||run.status!=='conflict'||!runContainsTarget(run,intent.action.target))return{status:'rejected',code:'output_not_applicable',targets:[intent.action.target.id]};}}
const recoveries=new Map(previous.recoveries),resourcePins=new Map(previous.resourcePins),conflicts=new Map(previous.conflicts);recoveries.set(plan.id,plan.audit);resourcePins.set(`recovery:${plan.id}`,plan.resourceIds);conflicts.delete('refresh');
for(const decision of plan.audit.decisions){const requestId=previous.ledger.operations.get(decision.operationId)?.requestId;if(requestId)conflicts.delete(requestId);}
const result=editor.reconcile(plan.document,Object.freeze({...previous,ledger:plan.ledger,loading:undefined,acknowledged,versions:canonical.versions,recoveries:readMap(recoveries),resourcePins:readMap(resourcePins),conflicts:readMap(conflicts)}),root.rootRevision,plan);
const result=editor.reconcile(plan.document,Object.freeze({...previous,ledger:plan.ledger,loading:undefined,loadingRead:undefined,acknowledged,versions:canonical.versions,recoveries:readMap(recoveries),resourcePins:readMap(resourcePins),conflicts:readMap(conflicts)}),root.rootRevision,plan);
if(result.status==='rejected')return result;committed=plan;committedIds=plan.operationIds;
}
return finish();
@@ -771,7 +776,7 @@ export async function openProjectEngine(options: EngineOptions) {
if(conflict)return{status:'rejected',...conflict};
if(!committed){
const conflicts=new Map(previous.conflicts);conflicts.delete('refresh');
const result=editor.reconcile(working!,Object.freeze({...previous,loading:undefined,acknowledged,versions:canonical.versions,conflicts:readMap(conflicts)}),root.rootRevision);
const result=editor.reconcile(working!,Object.freeze({...previous,loading:undefined,loadingRead:undefined,acknowledged,versions:canonical.versions,conflicts:readMap(conflicts)}),root.rootRevision);
if(result.status==='rejected')return result;
committed='refresh';
}
@@ -917,7 +922,7 @@ export async function openProjectEngine(options: EngineOptions) {
cancelRecovery:async()=>{await recovery?.cancel();},
getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }),
retry: async () => { if(recovery)await recovery.cancel();assertWritable(false);await persist();retryDefaultSizes();const runtime=editor.getSnapshot().runtime,ids=[...runtime.resources.values()].filter(resource=>!resource.digest&&runtime.ledger.operations.get(resource.operationId)?.blockers.some(code=>digestFailures.has(code))).map(resource=>resource.operationId);if(ids.length){mutate(ledger=>{for(const id of ids)ledger.block(id,['waiting_digest']);});prepareDigests();}if(options.images){editor.updateRuntime(previous=>{const resources=new Map(previous.resources);let changed=false;for(const [id,resource]of resources)if(resource.image?.version===1&&resource.image.status==='failed'){resources.set(id,freeze({...resource,image:{version:1 as const,status:'pending' as const}}));changed=true;}return changed?Object.freeze({...previous,resources:readMap(resources)}):previous;});prepareImages();}await scheduler.resume();await runTracker.retry(); },
async settle() { await Promise.resolve();await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; await scheduler.settle(); await runTracker.settle(); await persist(); },
async settle() { await Promise.resolve();await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; await scheduler.settle(); await runTracker.settle(); await persist();await scheduler.settle();await persist(); },
async close() {
if (closed) return;
closing = true;loadingReadController?.abort();await loadingRead?.catch(()=>{});unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));digestLifecycle.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop();
src/lib/canvas-runtime/host.ts
+2 -2
View File
@@ -37,11 +37,11 @@ export function createRuntimeHost(options: { registry: Registry; workspace(accou
const preview=settings.onPreview?createModelPreview(options.registry):undefined;let seen=0,previewFailed=false;let declaredCanvasIds:readonly Id[]=[];
const onUpdate=(state:import('./loader.ts').LoadSnapshot<CanonicalReadUpdate>)=>{
if(!preview||previewFailed||disposed||accountId!==account||authEpoch!==readEpoch||settings.signal?.aborted)return;
try{for(;seen<state.chunks.length;seen++){const chunk=state.chunks[seen],value=preview.update(chunk.value);if(chunk.value.status==='receiving'&&value){if(chunk.sequence===0)declaredCanvasIds=Object.freeze(chunk.value.canvases.map(canvas=>canvas.id));settings.onPreview?.(Object.freeze({...value,versions:readMap(new Map(Object.entries(chunk.serverVersions))),loading:Object.freeze({...(value.canvasElementCounts?{version:2 as const,canvasElementCounts:value.canvasElementCounts}:{version:1 as const}),sourceToken:chunk.snapshotToken,declaredCanvasIds,receivedElements:value.receivedElements,totalElements:value.totalElements})}));}}}
try{for(;seen<state.chunks.length;seen++){const chunk=state.chunks[seen],value=preview.update(chunk.value);if(chunk.value.status==='receiving'&&value){if(chunk.sequence===0)declaredCanvasIds=Object.freeze(chunk.value.canvases.map(canvas=>canvas.id));settings.onPreview?.(Object.freeze({...value,versions:readMap(new Map(Object.entries(chunk.serverVersions))),loading:Object.freeze({...(value.canvasElementCounts?{version:3 as const,canvasElementCounts:value.canvasElementCounts}:{version:1 as const}),sourceToken:chunk.snapshotToken,declaredCanvasIds,receivedElements:value.receivedElements,totalElements:value.totalElements})}));}}}
catch(error){previewFailed=true;try{options.onObserverError?.(error);}catch{}}
};
const handle=reader.acquire({accountId:account,workspaceId:workspace,projectId:null,canvasId:canvas,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry',authEpoch,navigationGeneration:settings.navigationGeneration??0},{signal:settings.signal,onUpdate});
try{const snapshot=await handle.ready;if(disposed||accountId!==account||authEpoch!==readEpoch)throw fault('identity_changed');const last=snapshot.chunks.at(-1);if(!last||!last.complete||last.value.status==='receiving')throw fault('incomplete_load');return last.value;}finally{handle.release();preview?.close();}
try{const snapshot=await handle.ready;if(disposed||accountId!==account||authEpoch!==readEpoch)throw fault('identity_changed');const last=snapshot.chunks.at(-1);if(!last||!last.complete||last.value.status==='receiving')throw fault('incomplete_load');return last.value.status==='ready'?Object.freeze({...last.value,sourceToken:last.snapshotToken}):last.value;}finally{handle.release();preview?.close();}
},
workspace(account: Id, workspace: Id) {
activate(account);
src/lib/canvas-runtime/loading.ts
+51 -16
View File
@@ -1,19 +1,19 @@
import {validateCanvasElementCounts,type CanvasElementCounts} from './scope-counts.ts';
import type {Availability,Change,Command,Document,Id,Registry} from '../canvas-core/types.ts';
import {validateDocument} from '../canvas-core/document.ts';
import {createDocument,validateDocument} from '../canvas-core/document.ts';
import {equal,fault} from '../canvas-core/values.ts';
import type {RuntimeState} from './contracts.ts';
import type {CanonicalProposal,RuntimeState} from './contracts.ts';
/** Durable marker for a verified read prefix. No network write is admitted while
* this exists. The token binds prefix extensions, not a resumable HTTP cursor. */
/** Durable marker for a verified read prefix. Versions 1/2 hold all writes;
* version 3 admits certified scopes with a retained read baseline. No HTTP cursor. */
type LoadingBase={readonly sourceToken:string;readonly declaredCanvasIds:readonly Id[];readonly receivedElements:number;readonly totalElements:number};
export type LoadingState=LoadingBase&({readonly version:1;readonly canvasElementCounts?:never}|{readonly version:2;readonly canvasElementCounts:CanvasElementCounts});
export type LoadingState=LoadingBase&({readonly version:1;readonly canvasElementCounts?:never}|{readonly version:2|3;readonly canvasElementCounts:CanvasElementCounts});
export function validateLoadingState(value:LoadingState,document:Document,versions:ReadonlyMap<Id,number>):void{
if(!value||![1,2].includes(value.version))throw fault('loading_interpreter_unavailable');
if(!value||![1,2,3].includes(value.version))throw fault('loading_interpreter_unavailable');
for(const [id,version]of versions)if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_canonical_version',[id]);
if(typeof value.sourceToken!=='string'||!value.sourceToken||value.sourceToken.length>256||!Array.isArray(value.declaredCanvasIds)||!value.declaredCanvasIds.length||value.declaredCanvasIds.some(id=>typeof id!=='string'||!id||!versions.has(id))||new Set(value.declaredCanvasIds).size!==value.declaredCanvasIds.length||!value.declaredCanvasIds.includes(document.projectId)||[...document.canvases.keys()].some(id=>!value.declaredCanvasIds.includes(id))||![value.receivedElements,value.totalElements].every(value=>Number.isSafeInteger(value)&&value>=0)||value.receivedElements>value.totalElements||document.elements.size>value.receivedElements)throw fault('invalid_loading_state');
if(value.version===1&&value.canvasElementCounts!==undefined)throw fault('invalid_loading_state');
if(value.version===2){validateCanvasElementCounts(value.canvasElementCounts,value.declaredCanvasIds,value.totalElements);const counts=new Map<Id,number>();for(const element of document.elements.values()){const count=(counts.get(element.ownerCanvasId)??0)+1;if(count>value.canvasElementCounts[element.ownerCanvasId])throw fault('invalid_model_coverage');counts.set(element.ownerCanvasId,count);}}
if(value.version!==1){validateCanvasElementCounts(value.canvasElementCounts,value.declaredCanvasIds,value.totalElements);const counts=new Map<Id,number>();for(const element of document.elements.values()){const count=(counts.get(element.ownerCanvasId)??0)+1;if(count>value.canvasElementCounts[element.ownerCanvasId])throw fault('invalid_model_coverage');counts.set(element.ownerCanvasId,count);}}
}
const denied:Availability=Object.freeze({enabled:false,code:'model_loading',targets:[]});
function dataKeyAllowed(componentId:Id,key:string,registry:Registry):boolean{
@@ -21,7 +21,7 @@ function dataKeyAllowed(componentId:Id,key:string,registry:Registry):boolean{
return !!definition?.copy&&!['src','assetId','presentation','ports','ai','taskId','runId'].includes(key)&&!definition.resourceFields?.includes(key)&&!definition.copy.references.some(ref=>ref.path[0]===key||ref.path[0]==='*');
}
function pathAllowed(id:Id,document:Document,registry:Registry,loading?:LoadingState,base=document):boolean{
const element=document.elements.get(id);if(loading?.version!==2||element?.structure!=='connection'||registry.resolve(element)!=='known')return false;
const element=document.elements.get(id);if((!loading||loading.version===1)||element?.structure!=='connection'||registry.resolve(element)!=='known')return false;
return (directCounts(base).get(element.ownerCanvasId)??0)===loading.canvasElementCounts[element.ownerCanvasId];
}
export function loadingCommand(command:Command,document:Document,registry:Registry,loading?:LoadingState,base=document):Availability{
@@ -41,12 +41,15 @@ export function assertLoadingChanges(changes:readonly Change[],document:Document
}
}
export function validateLoadingRuntime(runtime:RuntimeState,registry:Registry):void{
if(!runtime.loading)return;
validateLoadingState(runtime.loading,runtime.acknowledged,runtime.versions);
if(runtime.ledger.requests.size||runtime.resources.size||runtime.runs.size||runtime.imports?.size||runtime.orphanFiles?.size||runtime.recoveries.size)throw fault('invalid_loading_state');
if(!runtime.loading){if(runtime.loadingRead)throw fault('invalid_loading_state');return;}
const source=runtime.loading.version===3?runtime.loadingRead:undefined;
if(runtime.loading.version===3&&!source||runtime.loading.version!==3&&runtime.loadingRead)throw fault('invalid_loading_state');
validateLoadingState(runtime.loading,source?.document??runtime.acknowledged,source?.versions??runtime.versions);
if(source)validateLoadingRead(runtime,registry);
if(runtime.loading.version!==3&&runtime.ledger.requests.size||runtime.resources.size||runtime.runs.size||runtime.imports?.size||runtime.orphanFiles?.size||runtime.recoveries.size)throw fault('invalid_loading_state');
for(const record of runtime.ledger.operations.values()){
if(record.intent.operationType!=='document.change'||record.requestId||record.resolution||!['pending','blocked'].includes(record.phase))throw fault('invalid_loading_state');
assertLoadingChanges(record.intent.changes,runtime.acknowledged,registry,runtime.loading);
if(record.intent.operationType!=='document.change'||runtime.loading.version!==3&&(record.requestId||record.resolution||!['pending','blocked'].includes(record.phase)))throw fault('invalid_loading_state');
assertLoadingChanges(record.intent.changes,source?.document??runtime.acknowledged,registry,runtime.loading);
}
}
export function validatePrefixExtension(previous:Document,previousVersions:ReadonlyMap<Id,number>,state:LoadingState,next:Document,versions:ReadonlyMap<Id,number>,nextState:LoadingState,registry:Registry):void{
@@ -63,7 +66,7 @@ export function validateLoadingCompletion(previous:Document,previousVersions:Rea
for(const [id,record]of previous.elements)if(previousVersions.get(record.ownerCanvasId)===versions.get(record.ownerCanvasId)&&!equal(record,next.elements.get(id)))throw fault('canonical_version_mismatch',[record.ownerCanvasId]);
for(const [id,dead]of previous.tombstones)if((next.tombstones.get(id)??0)<dead)throw fault('stale_canonical_lifecycle',[id]);
const same=equal(Object.fromEntries(previousVersions),Object.fromEntries(versions));
if(state.version===2&&(!fresh||same)){const counts=new Map<Id,number>();for(const element of next.elements.values())counts.set(element.ownerCanvasId,(counts.get(element.ownerCanvasId)??0)+1);if(state.declaredCanvasIds.some(id=>(counts.get(id)??0)!==state.canvasElementCounts[id]))throw fault('invalid_model_coverage');}
if(state.version!==1&&(!fresh||same)){const counts=new Map<Id,number>();for(const element of next.elements.values())counts.set(element.ownerCanvasId,(counts.get(element.ownerCanvasId)??0)+1);if(state.declaredCanvasIds.some(id=>(counts.get(id)??0)!==state.canvasElementCounts[id]))throw fault('invalid_model_coverage');}
if(!fresh&&(!same||sourceToken!==state.sourceToken))throw fault('loading_source_mismatch');
if(same&&(next.elements.size!==state.totalElements||next.canvases.size!==state.declaredCanvasIds.length||state.declaredCanvasIds.some(id=>!next.canvases.has(id))||!equal(Object.fromEntries(previous.tombstones),Object.fromEntries(next.tombstones))))throw fault('incomplete_loading_model');
}
@@ -72,8 +75,40 @@ export function validateLoadingCompletion(previous:Document,previousVersions:Rea
const membershipCounts=new WeakMap<Document,ReadonlyMap<Id,number>>();
function directCounts(document:Document):ReadonlyMap<Id,number>{let counts=membershipCounts.get(document);if(!counts){const next=new Map<Id,number>();for(const element of document.elements.values())next.set(element.ownerCanvasId,(next.get(element.ownerCanvasId)??0)+1);counts=next;membershipCounts.set(document,counts);}return counts;}
export function loadingQueryCoverage(snapshot:import('./contracts.ts').EngineSnapshot):import('../canvas-core/queries.ts').QueryCoverage {
const loading=snapshot.runtime.loading,base=snapshot.runtime.acknowledged;
const loading=snapshot.runtime.loading,base=snapshot.runtime.loadingRead?.document??snapshot.runtime.acknowledged;
if(!loading)return{projectComplete:true,topologyComplete:true,completeCanvasIds:[...snapshot.document.canvases.keys()]};
const counts=directCounts(base);
return{projectComplete:false,topologyComplete:base.canvases.size===loading.declaredCanvasIds.length,completeCanvasIds:loading.version===2?[...base.canvases.keys()].filter(id=>(counts.get(id)??0)===loading.canvasElementCounts[id]):[]};
return{projectComplete:false,topologyComplete:base.canvases.size===loading.declaredCanvasIds.length,completeCanvasIds:loading.version!==1?[...base.canvases.keys()].filter(id=>(counts.get(id)??0)===loading.canvasElementCounts[id]):[]};
}
/** Validate the retained source separately from the moving ACK baseline. Only
* field edits in certified scopes may account for differences during loading. */
function validateLoadingRead(runtime:RuntimeState,registry:Registry):void {
const source=runtime.loadingRead!,loading=runtime.loading!,base=source.document,ack=runtime.acknowledged;
validateDocument(base,registry);
if(base.projectId!==ack.projectId||!equal(Object.fromEntries(base.canvases),Object.fromEntries(ack.canvases))||!equal(Object.fromEntries(base.tombstones),Object.fromEntries(ack.tombstones))||base.elements.size!==ack.elements.size||source.versions.size!==runtime.versions.size)throw fault('invalid_loading_read');
const ready=new Set(loadingQueryCoverage({runtime,document:ack} as import('./contracts.ts').EngineSnapshot).completeCanvasIds);
for(const [id,version]of source.versions){const current=runtime.versions.get(id);if(current===undefined||!Number.isSafeInteger(current)||current<version||current!==version&&!ready.has(id))throw fault('invalid_loading_read',[id]);}
for(const [id,before]of base.elements){const after=ack.elements.get(id);if(!after)throw fault('invalid_loading_read',[id]);if(equal(before,after))continue;
if(runtime.versions.get(before.ownerCanvasId)===source.versions.get(before.ownerCanvasId)||!ready.has(before.ownerCanvasId))throw fault('invalid_loading_read',[id]);
const {data:oldData,geometry:oldGeometry,path:oldPath,...oldEnvelope}=before as typeof before&{geometry?:unknown;path?:unknown};
const {data:newData,geometry:newGeometry,path:newPath,...newEnvelope}=after as typeof after&{geometry?:unknown;path?:unknown};
if(!equal(oldEnvelope,newEnvelope)||!equal(oldGeometry,newGeometry)&&before.structure!=='node'||!equal(oldPath,newPath)&&!pathAllowed(id,base,registry,loading))throw fault('invalid_loading_read',[id]);
for(const key of new Set([...Object.keys(oldData),...Object.keys(newData)]))if(!equal(oldData[key],newData[key])&&!dataKeyAllowed(before.componentId,key,registry))throw fault('invalid_loading_read',[id]);
}
for(const request of runtime.ledger.requests.values())if(request.kind!==undefined&&request.kind!=='document.change'||!request.canvasIds.length||new Set(request.canvasIds).size!==request.canvasIds.length||Object.keys(request.expectedVersions).length!==request.canvasIds.length||request.canvasIds.some(id=>!ready.has(id)||!Number.isSafeInteger(request.expectedVersions[id])||request.expectedVersions[id]<(source.versions.get(id)??Infinity)||request.expectedVersions[id]>(runtime.versions.get(id)??-1)))throw fault('invalid_loading_request');
}
/** Overlay only scopes whose ACK version advanced since this immutable capture. */
export function mergeLoadingRead(runtime:RuntimeState,read:CanonicalProposal,registry:Registry):CanonicalProposal {
const source=runtime.loadingRead;if(runtime.loading?.version!==3||!source)return read;
const advanced=new Set([...runtime.versions].filter(([id,version])=>version>(source.versions.get(id)??Infinity)).map(([id])=>id));
const elements=[...read.document.elements.values()].filter(record=>!advanced.has(record.ownerCanvasId));
for(const element of runtime.acknowledged.elements.values())if(advanced.has(element.ownerCanvasId))elements.push(element);
const document=importDocument(read.document,elements),versions=new Map(read.versions);for(const id of advanced)versions.set(id,runtime.versions.get(id)!);
validateDocument(document,registry);return{document,versions};
}
function importDocument(document:Document,elements:readonly import('../canvas-core/types.ts').ElementRecord[]):Document{return createDocument(document.projectId,[...document.canvases.values()],elements,document.tombstones);}
export function loadingCanSend(runtime:RuntimeState,canvasIds:readonly Id[]):boolean {
if(!runtime.loading)return true;if(runtime.loading.version!==3)return false;
const ready=new Set(loadingQueryCoverage({runtime,document:runtime.acknowledged} as import('./contracts.ts').EngineSnapshot).completeCanvasIds);return canvasIds.length>0&&canvasIds.every(id=>ready.has(id));
}
src/lib/canvas-runtime/recovery-encoding.ts
+2 -2
View File
@@ -52,8 +52,8 @@ export function encodeRecoveryValue(input:unknown):Json {
}
/** Only codec-owned, validated immutable images qualify for the storage copy fast path. */
export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
if(![1,2,3,4,5,6,7,8].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8|9,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
if(![1,2,3,4,5,6,7,8,9].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
const image=Object.freeze({formatVersion,rootRevision,localRevision,payload:encodeRecoveryValue(payload)});sealed.add(image);return image;
}
export function isSealedRecoveryImage(image:RecoveryImage):boolean{return sealed.has(image);}
src/lib/canvas-runtime/recovery.ts
+5 -5
View File
@@ -26,9 +26,9 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
document: documentValue(snapshot.document),
history: { undo: snapshot.history.undo, redo: snapshot.history.redo, entries: [...snapshot.history.entries] },
operations: [...snapshot.operations],
runtime: { ...(snapshot.runtime.loading?{loading:snapshot.runtime.loading}:{}), ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:[...snapshot.runtime.orphanFiles]}:{}), imports:[...(snapshot.runtime.imports??[])], recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
runtime: { ...(snapshot.runtime.loadingRead?{loadingRead:{document:documentValue(snapshot.runtime.loadingRead.document),versions:[...snapshot.runtime.loadingRead.versions]}}:{}), ...(snapshot.runtime.loading?{loading:snapshot.runtime.loading}:{}), ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:[...snapshot.runtime.orphanFiles]}:{}), imports:[...(snapshot.runtime.imports??[])], recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
};
return sealRecoveryImage(snapshot.runtime.loading?(snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
return sealRecoveryImage(snapshot.runtime.loading?(snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
}
type ObjectValue = Record<string, unknown>;
function object(value: unknown): ObjectValue { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('corrupt_recovery_image'); return value as ObjectValue; }
@@ -92,12 +92,12 @@ function readHistory(history:ObjectValue,operations:ReadonlyMap<Id,OperationInte
return Object.freeze({undo:freeze(undo),redo:freeze(redo),entries});
}
export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSnapshot & { readonly rootRevision: number; readonly runtime: RuntimeState } {
if (![1,2,3,4,5,6,7,8].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (![1,2,3,4,5,6,7,8,9].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
integer(image.rootRevision); integer(image.localRevision);
if (image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
// Own the parsed records; mutating a caller's RecoveryImage must not mutate live state.
const raw = object(JSON.parse(JSON.stringify(image.payload))), document = readDocument(raw.document, registry), history = object(raw.history), runtime = object(raw.runtime), ledger = object(runtime.ledger);
if(runtime.loading!==undefined&&image.formatVersion!==(object(runtime.loading).version===2?8:7))throw fault('unsupported_recovery_version');
if(runtime.loading!==undefined&&image.formatVersion!==(object(runtime.loading).version===3?9:object(runtime.loading).version===2?8:7))throw fault('unsupported_recovery_version');
if(image.formatVersion>=7&&runtime.loading===undefined)throw fault('invalid_loading_state');
const operations = table(raw.operations, intent),restoredHistory=readHistory(history,operations);
if(image.formatVersion<6&&[...operations.values()].some(intent=>intent.historyEffect))throw fault('unsupported_recovery_version');
@@ -243,7 +243,7 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
if(orphanFiles.size&&image.formatVersion<5)throw fault('unsupported_recovery_version');for(const id of orphanFiles.keys())if(resources.has(id)||importFiles(imports).has(id))throw fault('local_file_identity_conflict',[id]);
const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests });
const snapshot=Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history:restoredHistory, operations,
runtime: Object.freeze({ ...(runtime.loading!==undefined?{loading:freeze(runtime.loading as LoadingState)}:{}), orphanFiles, imports, recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
runtime: Object.freeze({ ...(runtime.loadingRead!==undefined?{loadingRead:Object.freeze({document:readDocument(object(runtime.loadingRead).document,registry),versions:table(object(runtime.loadingRead).versions,(value)=>integer(value,1))})}:{}), ...(runtime.loading!==undefined?{loading:freeze(runtime.loading as LoadingState)}:{}), orphanFiles, imports, recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
});
if([...resources.values()].some(resource=>resource.retirement?.purgedAt!==undefined&&!resource.stored)){
const pins=resourceRetentionPins(snapshot,registry);for(const [id,resource]of resources)if(resource.retirement?.purgedAt!==undefined&&!resource.stored&&pins.pins.has(id))throw fault('corrupt_recovery_image');
src/lib/canvas-runtime/scheduler.ts
+8 -2
View File
@@ -9,6 +9,7 @@ export type Clock = { now(): number; setTimer(callback: () => void, delay: numbe
const browserClock: Clock = { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
type Ledger = ReturnType<typeof createLedger>;
export type SchedulerOptions = {
admit?: (intent:OperationIntent)=>boolean;
getLedger(): LedgerSnapshot;
mutate(reduce: (ledger: Ledger) => void): void;
persist(): Promise<void>;
@@ -33,6 +34,7 @@ export type SchedulerOptions = {
export function createScheduler(options: SchedulerOptions) {
const clock = options.clock ?? browserClock, concurrency = options.concurrency ?? 2;
if (!Number.isSafeInteger(concurrency) || concurrency < 1) throw fault('invalid_concurrency');
const admissionWaiters=new Set<Id>();
const ready = createDeque<Id>(), queued = new Set<Id>(), claims = new Map<Id, Id>();
const waiters = new Map<Id, Set<Id>>(), dependents = new Map<Id, Set<Id>>(), indexed = new Set<Id>();
const retry = createMinHeap<{ id: Id; dueAt: number; sequence: number; token: number }>((a, b) => a.dueAt - b.dueAt || a.sequence - b.sequence);
@@ -50,8 +52,9 @@ export function createScheduler(options: SchedulerOptions) {
function enqueue(id: Id) {
if (queued.has(id)) return;
const record = options.getLedger().operations.get(id);
if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) return;
queued.add(id); ready.push(id);
if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) {admissionWaiters.delete(id);return;}
if(options.admit&&!options.admit(record.intent)){admissionWaiters.add(id);return;}
admissionWaiters.delete(id);queued.add(id); ready.push(id);
}
function index(id: Id) {
const record = options.getLedger().operations.get(id);
@@ -78,6 +81,7 @@ export function createScheduler(options: SchedulerOptions) {
const record = ledger.operations.get(ordered[i].id);
if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.operationType !== 'document.change' || record.intent.operationSchemaVersion !== 1) break;
if (record.intent.dependencies.some(id => !ids.has(id) && ledger.operations.get(id)?.phase !== 'resolved')) break;
if(options.admit&&!options.admit(record.intent))break;
const scope = options.participants(record.intent);
// Independent canvas lanes keep independent failure and retry boundaries.
if (!scope.length || !scope.some(id => canvases.has(id)) || scope.some(id => claims.has(id)) || changeCount + record.intent.changes.length > 10_000) break;
@@ -177,6 +181,7 @@ export function createScheduler(options: SchedulerOptions) {
const record = options.getLedger().operations.get(id);
if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue;
try {
if(options.admit&&!options.admit(record.intent)){admissionWaiters.add(id);continue;}
if (!canRun(record.intent.operationType)) {
const set = laneWaiters.get(record.intent.operationType) ?? new Set<Id>(); set.add(id); laneWaiters.set(record.intent.operationType, set); continue;
}
@@ -225,6 +230,7 @@ export function createScheduler(options: SchedulerOptions) {
}
pump();
},
refreshAdmission(){for(const id of [...admissionWaiters])enqueue(id);if(ready.size)requestPump();},
notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); requestPump(); },
/** Acquire synchronously, then settle(): already admitted work completes with its original identity. */
hold() {
src/lib/canvas-runtime/storage-port.ts
+1 -1
View File
@@ -2,7 +2,7 @@ import type { Json } from '../canvas-core/types.ts';
/** Portable durable-store contract. Browser database types belong to the adapter. */
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
export type DurableBatch = { readonly image: RecoveryImage; readonly deleteFiles?:readonly string[]; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] };
export type StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise<void> };
export type StoredFileInfo={readonly id:string;readonly byteLength:number;readonly mimeType:string};
src/lib/canvas-runtime/storage.ts
+1 -1
View File
@@ -30,7 +30,7 @@ function purgedFileIds(image:RecoveryImage):Set<string>{
return result;
}
function validateImage(image: RecoveryImage) {
if (![1,2,3,4,5,6,7,8].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (![1,2,3,4,5,6,7,8,9].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
}
const requestValue = <T>(request: IDBRequest<T>) => new Promise<T>((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); });
src/lib/canvas-runtime/workspace.ts
+2 -2
View File
@@ -31,8 +31,8 @@ export function createWorkspaceRuntime(options: WorkspaceOptions) {
accountId: options.accountId, workspaceId: options.workspaceId, authEpoch: options.authEpoch, registry: options.registry,
isCurrentIdentity: current,
getProjects: () => readMap(projects),
open(document: Document, versions: ReadonlyMap<Id, number>) {
return acquire(document,versions).then(engine=>{if(!current())throw fault('identity_changed');const previous=engine.getSnapshot().runtime.versions;if(![...versions].some(([id,value])=>(previous.get(id)??0)>value))engine.refresh(document,versions);return engine;});
open(document: Document, versions: ReadonlyMap<Id, number>,settings:{sourceToken?:string}={}) {
return acquire(document,versions).then(async engine=>{if(!current())throw fault('identity_changed');const loading=engine.loading.getState();if(loading&&settings.sourceToken===loading.sourceToken){const result=await engine.loading.complete(document,versions,{sourceToken:settings.sourceToken});if(result.status==='rejected')throw fault(result.code,result.targets);}else{const previous=engine.getSnapshot().runtime.versions;if(![...versions].some(([id,value])=>(previous.get(id)??0)>value)){const result=engine.refresh(document,versions);if(result.status==='rejected'&&loading)throw fault(result.code,result.targets);}}return engine;});
},
openPartial(document:Document,versions:ReadonlyMap<Id,number>,loading:LoadingState){
return acquire(document,versions,loading).then(engine=>{if(!current())throw fault('identity_changed');const previous=engine.loading.getState();if(previous?.sourceToken===loading.sourceToken){const result=engine.loading.accept(document,versions,loading);if(result.status==='rejected')throw fault(result.code,result.targets);}return engine;});