From 6c0a421f59eaa2bf5d845b0f980cc9a416c82288 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Mon, 9 Nov 2026 04:02:22 +0800 Subject: [PATCH] Synchronize certified canvas scopes while preserving the original loading snapshot --- headless/README.md | 30 +++++++-- .../canvas/headless-graph-canvas.tsx | 2 +- src/components/workspace/graph-page.tsx | 2 +- src/lib/api.ts | 1 + src/lib/canvas-runtime/contracts.ts | 3 +- src/lib/canvas-runtime/engine.ts | 41 +++++++----- src/lib/canvas-runtime/host.ts | 4 +- src/lib/canvas-runtime/loading.ts | 67 ++++++++++++++----- src/lib/canvas-runtime/recovery-encoding.ts | 4 +- src/lib/canvas-runtime/recovery.ts | 10 +-- src/lib/canvas-runtime/scheduler.ts | 10 ++- src/lib/canvas-runtime/storage-port.ts | 2 +- src/lib/canvas-runtime/storage.ts | 2 +- src/lib/canvas-runtime/workspace.ts | 4 +- tests/headless/browser-storage-cases.mjs | 23 +++++-- tests/headless/loading-engine.test.mjs | 43 ++++++++++++ tests/headless/package.integration.mjs | 1 + tests/headless/recovery-encoding.test.mjs | 2 +- tests/headless/renderer.integration.mjs | 12 +++- 19 files changed, 199 insertions(+), 64 deletions(-) diff --git a/headless/README.md b/headless/README.md index d46a85b..8001054 100644 --- a/headless/README.md +++ b/headless/README.md @@ -37,7 +37,8 @@ implied by this local package. Experimental partial-model runtime: pass a validated prefix plus `loading` in `EngineOptions`. Version 1 uses recovery format 7; version 2 carries snapshot-bound direct canvas -membership counts and uses format 8. Both keep the scheduler held. +membership counts and uses format 8. Both keep the scheduler held. Version 3 +uses format 9 and retains `runtime.loadingRead` separately from the moving ACK base. Loaded non-reference data fields and node geometry can be edited and undone; creation, deletions, ownership/reference changes, uploads, imports and AI admission wait. `engine.loading.accept(document, versions, state)` extends the captured prefix without @@ -55,8 +56,8 @@ restricted field/geometry edits above, coalesces incoming prefixes, and preserve textarea, history and viewport when the full model arrives. App navigation cancels the view's stream and asks the retained project owner to complete through a fresh read. Reload recovers the durable prefix when that project is opened again. This is not a -persistent HTTP cursor or a workspace-wide startup scan; partial-scope ACK/chunk -interleaving and broader structural permissions remain unfinished. +persistent HTTP cursor or a workspace-wide startup scan. Version-3 scoped ACK/chunk +coordination is described below; broader structural permissions remain unfinished. Scope-aware queries are obtained from `engine.editor.getQueries()` and share the same publication boundary as the root. `lookupElement(id)` returns `present`, @@ -73,6 +74,23 @@ coverage. Counts include nodes and connections; they must cover exactly the decl active canvases, sum to the total, and never be exceeded by received records. A scope is ready only after its full count has entered the dependency-closed published prefix. Version-2 loading permits `connection.path` edits and history in a certified owner -scope; endpoint, ownership, creation and deletion edits still wait. Neither scope -completion nor receiving the last element replaces terminal/EOF validation or releases -network scheduling. Partial-scope ACK/chunk interleaving remains pending. +scope; endpoint, ownership, creation and deletion edits still wait. Scope completion never replaces terminal/EOF validation for the whole project. +Versions 1/2 continue to hold network scheduling. In version 3, mutable operations +wait in the scheduler's derived admission set until every participant scope is +complete and its root is durable; unrelated certified scopes can send meanwhile. +The normal frozen-body/key, writer claims and pre-dispatch durability barrier apply. + +`runtime.loadingRead` retains the original source document and version vector. Incoming +prefixes and same-source completion validate against that capture; scopes with advanced +ACK versions retain their newer acknowledged records, then pending intents replay. +`Workspace.open(document, versions, {sourceToken})` completes that original stream, +even when its captured version is now older than a scoped ACK. An unrelated fresh +read waits for outstanding receipts and uses a scheduling hold while capturing and +persisting its base; edits during the read remain local and are replayed. + +Recovery format 9 encodes both baselines and original frozen requests. Unknown results +still query/retry the same body/key after reopening. Partial-source completion clears +the retained read baseline and forces a full checkpoint without older partial journal +dependencies. Creation, deletion, owner/endpoint/reference changes, uploads and AI +remain gated during partial loading; broader structural scope admission and real +Foundry deployment/integration remain unfinished. diff --git a/src/components/canvas/headless-graph-canvas.tsx b/src/components/canvas/headless-graph-canvas.tsx index ea1624d..5261b64 100644 --- a/src/components/canvas/headless-graph-canvas.tsx +++ b/src/components/canvas/headless-graph-canvas.tsx @@ -252,7 +252,7 @@ function CanvasView({ engine, board, viewId, renderers,fitLoading=true }: { engi }} /> flow.zoomOut()} /> flow.zoomIn()} /> flow.fitView({ padding: .2 })} /> { const snapshot = engine.getSnapshot(), url = URL.createObjectURL(new Blob([JSON.stringify({ documentFormatVersion: 1, projectId: snapshot.document.projectId, canvases: [...snapshot.document.canvases.values()], elements: [...snapshot.document.elements.values()], tombstones: Object.fromEntries(snapshot.document.tombstones) })], { type: 'application/json' })); const a = window.document.createElement('a'); a.href = url; a.download = board.title + '.json'; a.click(); setTimeout(() => URL.revokeObjectURL(url), 1000); }} /> -

{blocked ? '有修改需要处理' : loading ? `正在加载… ${loading.receivedElements}/${loading.totalElements}，修改已在本地排队` : pending.length ? '正在同步…' : '已同步'}{blocked && <>}

+

{blocked ? '有修改需要处理' : loading ? `正在加载… ${loading.receivedElements}/${loading.totalElements}，${loading.version===3?"按已就绪范围同步":"修改已在本地排队"}` : pending.length ? '正在同步…' : '已同步'}{blocked && <>}

engine.sessions.tool(viewId, 'select')} /> engine.sessions.tool(viewId, 'pan')} /> {if(selected&&selectedResource)report(engine.fitImage({id:selected.id,incarnation:selected.incarnation},selectedResource.id));}} /> diff --git a/src/components/workspace/graph-page.tsx b/src/components/workspace/graph-page.tsx index 54f4092..6fbe02f 100644 --- a/src/components/workspace/graph-page.tsx +++ b/src/components/workspace/graph-page.tsx @@ -60,7 +60,7 @@ export default function GraphPage({ workspaceId, boardId }: { workspaceId: strin if (result.mode === 'legacy') { setLoaded({ board: result.board, snapshot: result.snapshot, route });setPreview(null); return; } while(pumping)await pumping; if(!active||!currentIdentity?.())return; - const engine = await runtimeHost.workspace(result.profile.id, result.board.workspace_id).open(result.model.document, result.model.versions); + const engine = await runtimeHost.workspace(result.profile.id, result.board.workspace_id).open(result.model.document, result.model.versions,{sourceToken:result.model.sourceToken}); if (active&&(!currentIdentity||currentIdentity())){setLoaded({ board: result.board, engine, route });setPreview(null);} }) .catch((error) => { if (active) { setError(errorText(error)); if (error.message === "invalid_session") router.replace("/login"); } }); diff --git a/src/lib/api.ts b/src/lib/api.ts index 6e9bee4..fd4f3a5 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -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: "画布正在关闭，本地修改会保留用于恢复。", diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts index d74ea3a..723d736 100644 --- a/src/lib/canvas-runtime/contracts.ts +++ b/src/lib/canvas-runtime/contracts.ts @@ -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; readonly imports?: ReadonlyMap; @@ -25,7 +26,7 @@ export type EngineSnapshot = Snapshot; export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; /** 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; readonly versions:ReadonlyMap; readonly receivedElements:number; readonly totalElements:number }; -export type CanonicalReadResult = ({ readonly status: 'ready' } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap }; +export type CanonicalReadResult = ({ readonly status: 'ready';readonly sourceToken?:string } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap }; /** A validated historical execution receipt; its old model is never published over a fresh read. */ export type AppliedModelProof = { readonly versions:ReadonlyMap; assertCanonical(proposal:CanonicalProposal):void }; export type PersistencePort = { diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 4505238..94ef45e 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -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 } | undefined; const admissionPins=(snapshot:EngineSnapshot)=>{const pins=new Map();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=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,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{ 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((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,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,settings:{fresh?:boolean;sourceToken?:string}={}):Promise{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(); - 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(); diff --git a/src/lib/canvas-runtime/host.ts b/src/lib/canvas-runtime/host.ts index 1c5d7a2..77b8eb5 100644 --- a/src/lib/canvas-runtime/host.ts +++ b/src/lib/canvas-runtime/host.ts @@ -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)=>{ if(!preview||previewFailed||disposed||accountId!==account||authEpoch!==readEpoch||settings.signal?.aborted)return; - try{for(;seencanvas.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(;seencanvas.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); diff --git a/src/lib/canvas-runtime/loading.ts b/src/lib/canvas-runtime/loading.ts index 62f0f8a..78355c3 100644 --- a/src/lib/canvas-runtime/loading.ts +++ b/src/lib/canvas-runtime/loading.ts @@ -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):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();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();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,state:LoadingState,next:Document,versions:ReadonlyMap,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)();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();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>(); function directCounts(document:Document):ReadonlyMap{let counts=membershipCounts.get(document);if(!counts){const next=new Map();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!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)); } diff --git a/src/lib/canvas-runtime/recovery-encoding.ts b/src/lib/canvas-runtime/recovery-encoding.ts index 7d27b97..f17a86d 100644 --- a/src/lib/canvas-runtime/recovery-encoding.ts +++ b/src/lib/canvas-runtime/recovery-encoding.ts @@ -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);} diff --git a/src/lib/canvas-runtime/recovery.ts b/src/lib/canvas-runtime/recovery.ts index fbd8a67..2914a13 100644 --- a/src/lib/canvas-runtime/recovery.ts +++ b/src/lib/canvas-runtime/recovery.ts @@ -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; 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 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'); diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index 9532fda..225f481 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -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) }; type Ledger = ReturnType; export type SchedulerOptions = { + admit?: (intent:OperationIntent)=>boolean; getLedger(): LedgerSnapshot; mutate(reduce: (ledger: Ledger) => void): void; persist(): Promise; @@ -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(); const ready = createDeque(), queued = new Set(), claims = new Map(); const waiters = new Map>(), dependents = new Map>(), indexed = new Set(); 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(); 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() { diff --git a/src/lib/canvas-runtime/storage-port.ts b/src/lib/canvas-runtime/storage-port.ts index 808e3b6..a5cbf91 100644 --- a/src/lib/canvas-runtime/storage-port.ts +++ b/src/lib/canvas-runtime/storage-port.ts @@ -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 }; export type StoredFileInfo={readonly id:string;readonly byteLength:number;readonly mimeType:string}; diff --git a/src/lib/canvas-runtime/storage.ts b/src/lib/canvas-runtime/storage.ts index af926a3..4397b90 100644 --- a/src/lib/canvas-runtime/storage.ts +++ b/src/lib/canvas-runtime/storage.ts @@ -30,7 +30,7 @@ function purgedFileIds(image:RecoveryImage):Set{ 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 = (request: IDBRequest) => new Promise((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); }); diff --git a/src/lib/canvas-runtime/workspace.ts b/src/lib/canvas-runtime/workspace.ts index 8f3b397..80bceab 100644 --- a/src/lib/canvas-runtime/workspace.ts +++ b/src/lib/canvas-runtime/workspace.ts @@ -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) { - 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,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,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;}); diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index fc6cdd8..c943a30 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -679,9 +679,9 @@ async function engineRecoveryCase(name, assert) { } finally { await engine?.close(); storage.close(); indexedDB.deleteDatabase(`${name}-engine`); } } -async function loadingRecoveryCase(name,assert,coverage=false){ - const format=coverage?8:7;name+=coverage?"-coverage":"-legacy"; - const root=crypto.randomUUID(),scope={accountId:crypto.randomUUID(),workspaceId:crypto.randomUUID(),projectId:root},registry=createRegistry(builtinComponents()),node=id=>({id,incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',data:{title:id,text:'original'},geometry:{x:0,y:0,width:120,height:80,rotation:0}}),a=node('A'),b=node('B'),canvases=[{id:root,parent:null}],prefix=createDocument(root,canvases,[a]),full=createDocument(root,canvases,[a,b]),versions=new Map([[root,1]]),loading={...(coverage?{version:2,canvasElementCounts:{[root]:2}}:{version:1}),sourceToken:'browser-captured-read',declaredCanvasIds:[root],receivedElements:1,totalElements:2}; +async function loadingRecoveryCase(name,assert,coverage=false,scoped=false){ + const format=scoped?9:coverage?8:7;name+=scoped?"-scoped":coverage?"-coverage":"-legacy"; + const root=crypto.randomUUID(),scope={accountId:crypto.randomUUID(),workspaceId:crypto.randomUUID(),projectId:root},registry=createRegistry(builtinComponents()),node=id=>({id,incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',data:{title:id,text:'original'},geometry:{x:0,y:0,width:120,height:80,rotation:0}}),a=node('A'),b=node('B'),canvases=[{id:root,parent:null}],prefix=createDocument(root,canvases,[a]),full=createDocument(root,canvases,[a,b]),versions=new Map([[root,1]]),loading={...(coverage?{version:scoped?3:2,canvasElementCounts:{[root]:2}}:{version:1}),sourceToken:'browser-captured-read',declaredCanvasIds:[root],receivedElements:1,totalElements:2}; const adapter=createBrowserStorage({name:`${name}-loading`,checkpointEvery:3});let ownerForTest;const storage={...adapter,async claim(scope){ownerForTest=await adapter.claim(scope);return ownerForTest;}};const calls=[],receipts=new Map();let server=full,version=1,engine; const model=(complete=true)=>({documentFormatVersion:1,projectId:root,complete,canvases,elements:[...server.elements.values()],tombstones:{},versions:{[root]:version},removedCanvasIds:[]}); const transport=async input=>{if(input.method==='GET')return{status:200,body:{model:model()}};calls.push(input);const body=JSON.parse(input.body);assert(body.expected_versions[root]===version,'loading writer used a stale version');server=applyChanges(server,body.operations.flatMap(operation=>operation.changes));version++;const receipt={protocol_version:2,request_id:input.requestId,status:'applied',operation_ids:body.operations.map(operation=>operation.id),model:model(false)};receipts.set(input.requestId,receipt);return{status:200,body:receipt};}; @@ -699,6 +699,17 @@ assert(!engine.loading.getState()&&server.elements.get('A').data.text==='local b }finally{await engine?.close();storage.close();indexedDB.deleteDatabase(`${name}-loading`);} } +async function scopedAckRecoveryCase(name,assert){ + const root=crypto.randomUUID(),child=crypto.randomUUID(),scope={accountId:crypto.randomUUID(),workspaceId:crypto.randomUUID(),projectId:root},registry=createRegistry(builtinComponents()),node=(id,owner=root)=>({id,incarnation:1,ownerCanvasId:owner,componentId:'text',dataVersion:1,structure:'node',data:{title:id,text:'original'},geometry:{x:0,y:0,width:120,height:80,rotation:0}}),a=node('A'),b=node('B'),f={...node('F'),componentId:'frame'},c=node('C',child),edge={id:'E',incarnation:1,ownerCanvasId:root,componentId:'connection',dataVersion:1,structure:'connection',data:{},source:{elementId:'A'},target:{elementId:'C'},path:null},canvases=[{id:root,parent:null},{id:child,parent:{canvasId:root,entryElementId:'F'}}],prefix=createDocument(root,canvases,[a,f,c]),full=createDocument(root,canvases,[a,b,f,c,edge]),versions=new Map([[root,1],[child,1]]),loading={version:3,sourceToken:'scoped-browser-source',declaredCanvasIds:[root,child],receivedElements:3,totalElements:5,canvasElementCounts:{[root]:4,[child]:1}},storage=createBrowserStorage({name:name+'-scope-ack'}),receipts=new Map(),bodies=[]; + let server=full,currentVersions=new Map(versions),time=0,engine; + const model=(participants,complete=false)=>({documentFormatVersion:1,projectId:root,complete,canvases:canvases.filter(canvas=>participants.includes(canvas.id)),elements:[...server.elements.values()].filter(element=>participants.includes(element.ownerCanvasId)),tombstones:{},versions:Object.fromEntries(participants.map(id=>[id,currentVersions.get(id)])),removedCanvasIds:[]}); + const transport=async input=>{if(input.method==='GET')return{status:200,body:receipts.get(input.path.split('/').at(-1))};const body=JSON.parse(input.body),participants=Object.keys(body.expected_versions);bodies.push(input.body);for(const id of participants)assert(body.expected_versions[id]===currentVersions.get(id),'scoped request used a stale owner version');server=applyChanges(server,body.operations.flatMap(operation=>operation.changes));for(const id of participants)currentVersions.set(id,currentVersions.get(id)+1);const receipt={protocol_version:2,request_id:input.requestId,status:'applied',operation_ids:body.operations.map(operation=>operation.id),model:model(participants)};receipts.set(input.requestId,receipt);if(bodies.length===1)throw Error('lost scoped ACK');return{status:200,body:receipt};}; + const persistence=createFoundryPersistence({registry,workspaceId:scope.workspaceId,rootId:root,transport}),clock={now:()=>time,setTimer:()=>0,clearTimer(){}},open=()=>openProjectEngine({scope,registry,document:prefix,versions,loading,storage,persistence,clock}); + try{engine=await open();engine.editor.dispatch({type:'element.data',target:{id:'A',incarnation:1},patch:{text:'root waits'}});engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'child applied'}});await engine.settle();assert(bodies.length===1&&server.elements.get('C').data.text==='child applied'&&server.elements.get('A').data.text==='original','incomplete owner blocked another certified owner or sent prematurely');const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];await engine.close();time=1000;engine=await open();await engine.settle();assert(bodies.length===1&&engine.getSnapshot().runtime.ledger.requests.get(request.id).body===request.body,'reopen repeated the lost-ACK request');assert(engine.getSnapshot().runtime.versions.get(child)===2&&engine.getSnapshot().runtime.loadingRead.versions.get(child)===1,'read and ACK baselines did not recover independently'); + assert(engine.loading.accept(full,versions,{...loading,receivedElements:5}).status==='applied','old source tail did not reconcile');await engine.settle();assert(bodies.length===2&&server.elements.get('A').data.text==='root waits','newly complete owner did not wake');assert((await engine.loading.complete(full,versions,{sourceToken:loading.sourceToken})).status==='applied','original source could not complete after scoped ACKs');await engine.settle();assert(engine.getSnapshot().runtime.loadingRead===undefined&&engine.getSnapshot().document.elements.get('C').data.text==='child applied'&&engine.getSnapshot().document.elements.get('A').data.text==='root waits','old completion replaced ACKs or retained the source'); + }finally{await engine?.close();storage.close();indexedDB.deleteDatabase(name+'-scope-ack');} +} + export async function run() { const results = [], name = `headless-test-${crypto.randomUUID()}`; const scope = { accountId: 'test-account', workspaceId: 'test-workspace', projectId: 'test-project' }; @@ -720,7 +731,7 @@ export async function run() { await storage.commit(leader, { image: image(1) }); await rejects(() => storage.commit(leader, { image: { ...image(1), payload: null } }), 'storage_revision_conflict'); await rejects(() => storage.commit(leader, { image: image(3) }), 'storage_revision_gap'); - await rejects(() => storage.commit(leader, { image: { ...image(2), formatVersion: 9 } }), 'unsupported_recovery_version'); + await rejects(() => storage.commit(leader, { image: { ...image(2), formatVersion: 10 } }), 'unsupported_recovery_version'); results.push('same-cut retry is idempotent; conflicting, skipped and future-format cuts reject'); // The first put succeeds before a later duplicate resource aborts the whole transaction. @@ -773,6 +784,10 @@ export async function run() { results.push('recovery v7 preserves loaded-field edits/history in IndexedDB; an injected full-checkpoint failure retains the partial cut and zero writes, then retry removes v7 journal dependencies before releasing requests'); await loadingRecoveryCase(name,assert,true); results.push('recovery v8 preserves certified scope coverage and loading edits; full-checkpoint fault rolls back atomically, retry retires all partial journal dependencies before sending'); + await loadingRecoveryCase(name,assert,true,true); + results.push('recovery v9 retains the immutable read baseline and defers incomplete owners; full-checkpoint failure preserves the partial source/ledger and retry retires its journal before writes'); + await scopedAckRecoveryCase(name,assert); + results.push('scoped v9 requests survive lost ACK and IndexedDB reopening with exact body/key; querying confirms the original child write, old tail wakes the parent and completion preserves both advanced owners'); await scopedNetworkLoaderCase(assert); results.push('scoped loader shares one real streamed fetch, exposes the first chunk before tail release, keeps the other subscriber alive after cancellation and preserves partial data after a malformed tail'); await defaultImageSizeCase(name, assert); diff --git a/tests/headless/loading-engine.test.mjs b/tests/headless/loading-engine.test.mjs index 033fa63..958790a 100644 --- a/tests/headless/loading-engine.test.mjs +++ b/tests/headless/loading-engine.test.mjs @@ -153,3 +153,46 @@ test('v2 admits path edits in a certified connection owner while another scope i }finally{await engine.close();} const other=fixture(),incomplete=await other.open({document:full,loading:{...loading,canvasElementCounts:{[root]:5,[child]:1}}});try{assert.equal(incomplete.editor.can({type:'connection.path',target:{id:'E',incarnation:1},path}).enabled,false);}finally{await incomplete.close();} }); + +test('v3 sends a certified child before the parent tail, retains its ACK through old chunks and completes the original source',async()=>{ + const f=fixture(),document=createDocument(root,canvases,[a,frame,c]),loading={...state,version:3,receivedElements:3,canvasElementCounts:{[root]:4,[child]:1}},engine=await f.open({document,loading}); + try{ + assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'parent waits'}}).status,'applied');const edit=engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'child acknowledged before tail'}});assert.equal(edit.status,'applied');await engine.settle(); + assert.equal(f.calls.length,1,JSON.stringify(f.errors));assert.equal(f.server().elements.get('C').data.text,'child acknowledged before tail');assert.equal(engine.loading.getState().version,3);assert.equal(engine.getSnapshot().runtime.versions.get(child),2);assert.equal(engine.getSnapshot().runtime.loadingRead.versions.get(child),1);assert.equal(f.storage.image().formatVersion,9);decodeRecovery(f.storage.image(),registry); + assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'parent waits'}}).status,'applied');await engine.settle();assert.equal(f.calls.length,1,'incomplete owner froze a request'); + assert.equal(engine.loading.accept(full,versions,{...loading,receivedElements:5}).status,'applied');assert.equal(engine.getSnapshot().document.elements.get('C').data.text,'child acknowledged before tail');await engine.settle();assert.equal(f.calls.length,2);assert.equal(f.server().elements.get('A').data.title,'parent waits');assert.equal(engine.loading.getState().version,3); + assert.equal((await engine.loading.complete(full,versions,{sourceToken:loading.sourceToken})).status,'applied');await engine.settle();assert.equal(engine.getSnapshot().runtime.loadingRead,undefined);assert.equal(engine.getSnapshot().document.elements.get('C').data.text,'child acknowledged before tail');assert.equal(engine.getSnapshot().document.elements.get('A').data.title,'parent waits');assert.equal(engine.getSnapshot().runtime.versions.get(root),2);assert.equal(engine.getSnapshot().runtime.versions.get(child),2);assert.equal(engine.getSnapshot().history.undo.length,2);assert.equal(f.errors.length,0); + assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(f.server().elements.get('A').data.title,'parent waits');assert.equal(f.server().elements.get('C').data.text,'original');assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(f.server().elements.get('A').data.title,'A'); + }finally{await engine.close();} +}); +test('v3 unknown request reopens with original body/key, queries its receipt and rejects source tampering',async()=>{ + let time=0;const f=fixture(),document=createDocument(root,canvases,[a,frame,c]),loading={...state,version:3,receivedElements:3,canvasElementCounts:{[root]:4,[child]:1}},clock={now:()=>time,setTimer:()=>0,clearTimer(){}},persistence={...f.persistence,async submit(request,signal){await f.persistence.submit(request,signal);throw Error('lost ACK');}}; + let engine=await f.open({document,loading,persistence,clock});try{ + engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'applied unknown'}});await engine.settle();const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert.ok(request);assert.equal(request.phase,'waiting');assert.equal(f.server().elements.get('C').data.text,'applied unknown'); + const beforeFresh=engine.getSnapshot();assert.equal((await engine.loading.complete(f.server(),f.currentVersions(),{fresh:true})).code,'loading_waiting_for_receipts');assert.strictEqual(engine.getSnapshot(),beforeFresh);assert.ok(engine.loading.getState());await engine.persist();const saved=f.storage.image();assert.equal(saved.formatVersion,9);decodeRecovery(saved,registry); + for(const mutate of [image=>delete image.payload.runtime.loadingRead,image=>image.payload.runtime.loadingRead.versions.find(([id])=>id===child)[1]=99,image=>image.formatVersion=8,image=>delete image.payload.runtime.ledger.requests[0][1].expectedVersions[child]]){const broken=structuredClone(saved);mutate(broken);assert.throws(()=>decodeRecovery(broken,registry));} + await engine.close();time=1000;engine=await f.open({document,loading,clock});await engine.settle();const restored=engine.getSnapshot().runtime.ledger.requests.get(request.id);assert.equal(restored.body,request.body);assert.equal(restored.phase,'resolved');assert.equal(f.calls.length,1,'reopening submitted a new write instead of querying the old receipt');assert.equal(engine.getSnapshot().runtime.versions.get(child),2); + const before=engine.getSnapshot(),tampered=createDocument(root,canvases,[a,b,frame,{...c,data:{...c.data,text:'changed old stream'}},edge]);assert.equal(engine.loading.accept(tampered,versions,{...loading,receivedElements:5}).code,'loading_source_mismatch');assert.strictEqual(engine.getSnapshot(),before); + assert.equal((await engine.loading.complete(full,versions,{sourceToken:loading.sourceToken})).status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get('C').data.text,'applied unknown'); + }finally{await engine.close();} +}); + +test('background completion waits outside the network lane for a v3 unknown receipt, then reads the fresh full model',async()=>{ + let time=0,reads=0;const f=fixture(),document=createDocument(root,canvases,[a,frame,c]),loading={...state,version:3,receivedElements:3,canvasElementCounts:{[root]:4,[child]:1}},clock={now:()=>time,setTimer:()=>0,clearTimer(){}},persistence={...f.persistence,async read(signal){reads++;return f.persistence.read(signal);},async submit(request,signal){await f.persistence.submit(request,signal);throw Error('lost ACK');}},engine=await f.open({document,loading,persistence,clock}); + try{engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'wait for original receipt'}});await engine.settle();const completion=engine.loading.resume();await new Promise(resolve=>setImmediate(resolve));assert.equal(reads,0);time=1000;await engine.retry();assert.equal((await completion).status,'applied');await engine.settle();assert.equal(reads,1);assert.equal(f.calls.length,1);assert.equal(engine.getSnapshot().document.elements.get('C').data.text,'wait for original receipt');assert.equal(engine.loading.getState(),undefined);}finally{await engine.close();} +}); + +test('a fresh v3 background read fences new admissions while preserving edits made during the read',async()=>{ + let release,started;const gate=new Promise(resolve=>release=resolve),reading=new Promise(resolve=>started=resolve),f=fixture(),document=createDocument(root,canvases,[a,frame,c]),loading={...state,version:3,receivedElements:3,canvasElementCounts:{[root]:4,[child]:1}},engine=await f.open({document,loading,persistence:{...f.persistence,async read(signal){started();await gate;return f.persistence.read(signal);}}}); + try{const completion=engine.loading.resume();await reading;engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'edited during fresh read'}});await engine.settle();assert.equal(f.calls.length,0);release();assert.equal((await completion).status,'applied');await engine.settle();assert.equal(f.server().elements.get('C').data.text,'edited during fresh read');assert.equal(f.calls.length,1);}finally{release();await engine.close();} +}); + +test('a v3 certified owner cannot freeze until its source and edit are durably saved; retry wakes the same logical intent',async()=>{ + const f=fixture(),document=createDocument(root,canvases,[a,frame,c]),loading={...state,version:3,receivedElements:3,canvasElementCounts:{[root]:4,[child]:1}},engine=await f.open({document,loading}); + try{f.storage.broken=true;const edit=engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'durability first'}});await assert.rejects(engine.persist(),/quota/);await new Promise(resolve=>setImmediate(resolve));assert.equal(engine.getSnapshot().runtime.ledger.requests.size,0);assert.equal(f.calls.length,0);f.storage.broken=false;await engine.retry();await engine.settle();assert.equal(f.calls.length,1);assert.equal(f.server().elements.get('C').data.text,'durability first');assert.equal(engine.getSnapshot().runtime.ledger.operations.get(edit.operationIds[0]).phase,'resolved');}finally{f.storage.broken=false;await engine.close();} +}); + +test('closing a v3 background receipt wait unsubscribes and saves the unknown request without issuing a fresh read',async()=>{ + let reads=0;const f=fixture(),document=createDocument(root,canvases,[a,frame,c]),loading={...state,version:3,receivedElements:3,canvasElementCounts:{[root]:4,[child]:1}},engine=await f.open({document,loading,clock:{now:()=>0,setTimer:()=>0,clearTimer(){}},persistence:{...f.persistence,async read(signal){reads++;return f.persistence.read(signal);},async submit(request,signal){await f.persistence.submit(request,signal);throw Error('lost ACK');}}}); + engine.editor.dispatch({type:'element.data',target:{id:'C',incarnation:1},patch:{text:'retain on close'}});await engine.settle();const waiting=assert.rejects(engine.loading.resume());await new Promise(resolve=>setImmediate(resolve));await engine.close();await waiting;assert.equal(reads,0);assert.equal(f.calls.length,1);assert.equal(f.storage.image().formatVersion,9);assert.equal(decodeRecovery(f.storage.image(),registry).runtime.ledger.requests.size,1); +}); diff --git a/tests/headless/package.integration.mjs b/tests/headless/package.integration.mjs index e0e173e..31cbfe2 100644 --- a/tests/headless/package.integration.mjs +++ b/tests/headless/package.integration.mjs @@ -48,6 +48,7 @@ try { return accepted.status==='applied'?engine.loading.complete(document,versions,{sourceToken:state.sourceToken}):accepted; } export const certified:LoadingState={version:2,sourceToken:'consumer-read',declaredCanvasIds:['root'],receivedElements:0,totalElements:1,canvasElementCounts:{root:1}}; + export const scoped:LoadingState={...certified,version:3}; export function inspectPartial(engine:Awaited>){ const queries=engine.editor.getQueries(),lookup=queries.lookupElement('node'); const id:string|undefined=lookup.status==='present'?lookup.element.id:undefined; diff --git a/tests/headless/recovery-encoding.test.mjs b/tests/headless/recovery-encoding.test.mjs index c8a49d2..7b0fca7 100644 --- a/tests/headless/recovery-encoding.test.mjs +++ b/tests/headless/recovery-encoding.test.mjs @@ -24,5 +24,5 @@ test('cached subtree height cannot bypass the depth bound and omitted object fie const cycle={};cycle.self=cycle;assert.throws(()=>encodeRecoveryValue(cycle),/invalid_recovery_payload/);assert.throws(()=>encodeRecoveryValue({big:1n})); }); test('only factory-owned images are recognized; immutable-looking impostors retain defensive-copy handling',()=>{ - const mutable={title:'before'},image=sealRecoveryImage(2,3,2,{mutable});assert.equal(isSealedRecoveryImage(image),true);assert.equal(isSealedRecoveryImage(structuredClone(image)),false);assert.equal(isSealedRecoveryImage(Object.freeze({...image})),false);mutable.title='after';assert.equal(image.payload.mutable.title,'before');assert.ok(Object.isFrozen(image.payload.mutable));assert.throws(()=>sealRecoveryImage(2,1,2,{}),/corrupt_recovery_image/);assert.equal(isSealedRecoveryImage(sealRecoveryImage(3,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(4,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(5,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(6,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(7,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(8,1,1,{})),true);assert.throws(()=>sealRecoveryImage(9,1,1,{}),/corrupt_recovery_image/); + const mutable={title:'before'},image=sealRecoveryImage(2,3,2,{mutable});assert.equal(isSealedRecoveryImage(image),true);assert.equal(isSealedRecoveryImage(structuredClone(image)),false);assert.equal(isSealedRecoveryImage(Object.freeze({...image})),false);mutable.title='after';assert.equal(image.payload.mutable.title,'before');assert.ok(Object.isFrozen(image.payload.mutable));assert.throws(()=>sealRecoveryImage(2,1,2,{}),/corrupt_recovery_image/);assert.equal(isSealedRecoveryImage(sealRecoveryImage(3,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(4,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(5,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(6,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(7,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(8,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(9,1,1,{})),true);assert.throws(()=>sealRecoveryImage(10,1,1,{}),/corrupt_recovery_image/); }); diff --git a/tests/headless/renderer.integration.mjs b/tests/headless/renderer.integration.mjs index 1c06818..85996f8 100644 --- a/tests/headless/renderer.integration.mjs +++ b/tests/headless/renderer.integration.mjs @@ -22,6 +22,7 @@ if(performanceMode){ } const boards = [root, other].map((id, i) => ({ id, workspace_id: workspace, kind: 'board', version: 1, title: i ? 'Other board' : 'Headless board' })); const png = Buffer.from('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aB1sAAAAASUVORK5CYII=', 'base64'); +let scopeStreamRoot=null,releaseScopeStream;const scopeStreamGate=new Promise(resolve=>releaseScopeStream=resolve); let brokenStreamRoot=null,abandonedRoot=null,releaseAbandoned;const abandonedGate=new Promise(resolve=>releaseAbandoned=resolve); let releaseFirstModel;const firstModelGate=new Promise(resolve=>{releaseFirstModel=resolve;});let gatedFirstModel=!performanceMode; let releaseImport;const importGate=new Promise(resolve=>{releaseImport=resolve;});let importWriteSeen=false;const editReceipts=new Map(); @@ -48,6 +49,7 @@ const upstream = createServer(async (request, response) => { line({kind:'header',documentFormatVersion:1,projectId:value.projectId,canvases:value.canvases,versions:value.versions,tombstones:value.tombstones,removedCanvasIds:value.removedCanvasIds,elementCount:value.elements.length,canvasElementCounts:Object.fromEntries(value.canvases.map(canvas=>[canvas.id,value.elements.filter(element=>element.ownerCanvasId===canvas.id).length]))}); if(read[1]===brokenStreamRoot){line({kind:'elements',elements:value.elements.slice(0,1)});response.end('{broken tail}\n');return;} let start=0; + if(read[1]===scopeStreamRoot){line({kind:'elements',elements:value.elements.slice(0,3)});start=3;await scopeStreamGate;if(response.destroyed)return;} if(read[1]===abandonedRoot){line({kind:'elements',elements:value.elements.slice(0,1)});start=1;await abandonedGate;if(response.destroyed)return;} if(gatedFirstModel&&read[1]===root&&value.elements.length){gatedFirstModel=false;line({kind:'elements',elements:value.elements.slice(0,1)});start=1;await firstModelGate;} for(let offset=start;offset{const r=document.querySelector('.react-flow__node[data-id="scope-node"] [aria-label^="单击选中"]').getBoundingClientRect();return{x:r.x+50,y:r.y+25};})()`); + await call('Input.dispatchMouseEvent',{type:'mousePressed',...scopePoint,button:'left',buttons:1,clickCount:2});await call('Input.dispatchMouseEvent',{type:'mouseReleased',...scopePoint,button:'left',buttons:0,clickCount:2});await until(`document.activeElement===document.querySelector('.react-flow__node[data-id="scope-node"] textarea')&&!document.activeElement.readOnly`,'certified scope text did not enter editing');await call('Input.insertText',{text:' ACK before tail'});const scopeText=await evaluate('document.activeElement.value'); + for(let i=0;i<150&&documents.get(scopedRoot).elements.get('scope-node').data.text!==scopeText;i++)await pause(20);assert.equal(documents.get(scopedRoot).elements.get('scope-node').data.text,scopeText,'certified scope edit waited for another owner tail');assert.equal(versions.get(scopedRoot),2);assert.equal(versions.get(scopedChild),1);assert.equal(await evaluate(`!!document.querySelector('[data-headless-canvas][data-model-loading]')`),true);assert.equal(await evaluate(`!!document.querySelector('.react-flow__node[data-id="scope-tail"]')`),false); + await evaluate('window.__scopeTextarea=document.activeElement;true');releaseScopeStream();await until(`!!document.querySelector('[data-headless-canvas]:not([data-model-loading])')&&!!document.querySelector('.react-flow__node[data-id="scope-tail"]')`,'original stream did not complete after a scoped ACK');assert.equal(await evaluate('window.__scopeTextarea.isConnected&&document.activeElement===window.__scopeTextarea'),true);assert.equal(await evaluate('window.__scopeTextarea.value'),scopeText);assert.equal(versions.get(scopedRoot),2,'old tail caused a duplicate write');console.log('PASS certified owner edits ACK before another canvas tail, then original-source completion preserves content, focus, history and advanced versions'); + abandonedRoot=id(840);const backgroundRoot=abandonedRoot; documents.set(backgroundRoot,createDocument(backgroundRoot,[{id:backgroundRoot,parent:null}],[node('background-node',backgroundRoot,80),node('background-tail',backgroundRoot,450)]));versions.set(backgroundRoot,1);boards.push({id:backgroundRoot,workspace_id:workspace,kind:'board',version:1,title:'Background loading board'}); await call('Page.navigate',{url:`http://127.0.0.1:${port}/dashboard/${workspace}/${backgroundRoot}`});await until(`!!document.querySelector('[data-headless-canvas][data-model-loading] .react-flow__node[data-id="background-node"]')`,'background fixture did not open the prefix'); @@ -381,7 +391,7 @@ try { console.log('PASS logout retires the runtime before dropping its view without a render error'); } } finally { - releaseAbandoned();releaseFirstModel();releaseImport();releaseApply();releaseUpload(); socket?.close(); + releaseScopeStream();releaseAbandoned();releaseFirstModel();releaseImport();releaseApply();releaseUpload(); socket?.close(); if (chrome) { const exit = new Promise(resolve => { if (chrome.exitCode !== null) resolve(); else chrome.once('exit', resolve); }); chrome.kill(); await exit; } if (next?.pid) { const exit = new Promise(resolve => { if (next.exitCode !== null) resolve(); else next.once('exit', resolve); }); try { process.kill(-next.pid, 'SIGTERM'); } catch {} await Promise.race([exit, pause(3000)]); if (next.exitCode === null) try { process.kill(-next.pid, 'SIGKILL'); } catch {} } upstream.closeAllConnections(); await new Promise(resolve => upstream.close(resolve));