From a0417b48114dbda18e48392f66240dbfc73b5960 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 01:41:23 +0800 Subject: [PATCH] Add quiescent canonical recovery sessions without rewriting requests --- src/lib/canvas-runtime/contracts.ts | 2 + src/lib/canvas-runtime/engine.ts | 83 ++++++++++++++++++++---- src/lib/canvas-runtime/reconcile.ts | 20 ++++++ src/lib/canvas-runtime/runs.ts | 12 +++- src/lib/canvas-runtime/scheduler.ts | 21 ++++-- src/lib/canvas-service/foundry.ts | 6 ++ tests/headless/browser-storage-cases.mjs | 25 +++++++ tests/headless/engine.test.mjs | 64 ++++++++++++++++++ tests/headless/run-tracker.test.mjs | 27 ++++++++ tests/headless/scheduler.test.mjs | 23 +++++++ tests/headless/service.test.mjs | 13 ++++ 11 files changed, 276 insertions(+), 20 deletions(-) create mode 100644 tests/headless/run-tracker.test.mjs diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts index f6ab3f4..132ceb7 100644 --- a/src/lib/canvas-runtime/contracts.ts +++ b/src/lib/canvas-runtime/contracts.ts @@ -17,6 +17,8 @@ export type RuntimeState = { export type EngineSnapshot = Snapshot; export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; export type PersistencePort = { + /** Complete canonical read for a quiescent recovery session; never a legacy fallback. */ + read?: (signal: AbortSignal) => Promise; supportsRequest?: (request: FrozenRequest) => boolean; compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest; submit(request: FrozenRequest, signal: AbortSignal): Promise; diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index ddb6957..8079c21 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -3,11 +3,11 @@ import { createEditor, type EditorOptions, type Transaction } from '../canvas-co import { createDocument } from '../canvas-core/document.ts'; import { createSessions } from '../canvas-core/gestures.ts'; import { equal, fault, freeze, readMap, reject } from '../canvas-core/values.ts'; -import type { EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts'; +import type { CanonicalProposal, EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts'; import { createRunTracker, validateRunInput, runContainsTarget, type RunPort, type RunRecord, type RunPage } from './runs.ts'; import { createNetworkGate } from './network.ts'; import { createLedger, type FrozenRequest } from './ledger.ts'; -import { operationCanvases, replayPending } from './reconcile.ts'; +import { operationCanvases, replayPending, validateCanonicalRead } from './reconcile.ts'; import { decodeRecovery, encodeRecovery } from './recovery.ts'; import { createScheduler, type Clock, type OperationStatus } from './scheduler.ts'; import { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.ts'; @@ -19,6 +19,15 @@ export type EngineOptions = { historyLimit?: number; policy?: EditorOptions['policy']; traversalPolicy?: EditorOptions['traversalPolicy']; clock?: Clock; onError?: (error: unknown) => void; }; +export type RecoverySession = { + readonly baseRootRevision: number; + readonly canonical: CanonicalProposal; + readonly pendingOperationIds: readonly Id[]; + /** A conflict is diagnostic only: the existing document/history/requests remain intact. */ + readonly conflict?: { readonly code: string; readonly targets: readonly Id[] }; + apply(): Promise; + cancel(): Promise; +}; /** One project owner. Detaching a view does not close this engine or abort persistent work. */ export async function openProjectEngine(options: EngineOptions) { @@ -37,10 +46,12 @@ export async function openProjectEngine(options: EngineOptions) { const fileWrites = new Map(), digestTasks = new Map>(); let staged: readonly ResourceRecord[] = []; let applying: RunApplyInput | undefined; + let recovery: { controller: AbortController; release(): void } | undefined; + const assertWritable = () => { if(closing||closed)throw fault('engine_closing');if(recovery)throw fault('recovery_in_progress'); }; const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: readMap(options.versions), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) }); const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit, - policy(command, document) { if (closing || closed) return { enabled: false, code: 'engine_closing', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; }, - traversalPolicy(entry, direction) { return closing || closed ? { enabled: false, code: 'engine_closing', targets: [] } : options.traversalPolicy?.(entry, direction) ?? { enabled: true }; }, + policy(command, document) { if (closing || closed || recovery) return { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; }, + traversalPolicy(entry, direction) { return closing || closed || recovery ? { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] } : options.traversalPolicy?.(entry, direction) ?? { enabled: true }; }, onListenerError: report, prepareIntent: intent => applying && intent.origin==='user' ? {...intent,operationType:'run.apply',action:applying} : intent, nextOperationSequence(runtime, fallback) { let next = fallback; for (const record of runtime.ledger.operations.values()) next = Math.max(next, record.intent.sequence + 1); return next + staged.length; }, @@ -283,6 +294,51 @@ export async function openProjectEngine(options: EngineOptions) { scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]); return Object.freeze({ editor, sessions, + /** Read/preview under a scheduling hold. No request is rewritten or declared resolved here. */ + async prepareRecovery(): Promise { + assertWritable(); + if(!options.persistence.read)throw fault('canonical_reader_unavailable'); + const releaseScheduler=scheduler.hold(),releaseRuns=runTracker.hold(),controller=new AbortController(); + const active={controller,release(){if(recovery!==active)return;recovery=undefined;releaseRuns();releaseScheduler();}}; + recovery=active; + sessions.cancel('recovery'); + const assertCurrent=()=>{if(closing||closed)throw fault('engine_closing');if(recovery!==active||controller.signal.aborted)throw fault('recovery_session_closed');}; + try { + // Both holds are acquired before awaiting either lane; already admitted callbacks finish normally. + await Promise.all([scheduler.settle(),runTracker.settle()]); + while(digestTasks.size)await Promise.all([...digestTasks.values()]); + await persist();assertCurrent(); + const root=editor.getSnapshot(),previous=root.runtime; + const unresolved=[...previous.ledger.requests.values()].filter(request=>request.phase!=='resolved'); + if(unresolved.length)throw fault('recovery_waiting_for_receipts',unresolved.map(request=>request.id)); + const loaded=await network.run(controller.signal,()=>options.persistence.read!(controller.signal));assertCurrent(); + if(editor.getSnapshot().rootRevision!==root.rootRevision)throw fault('stale_plan'); + validateCanonicalRead(previous.acknowledged,previous.versions,loaded.document,loaded.versions,options.registry); + const acknowledged=createDocument(loaded.document.projectId,[...loaded.document.canvases.values()],[...loaded.document.elements.values()],loaded.document.tombstones); + const canonical=Object.freeze({document:acknowledged,versions:readMap(loaded.versions)}); + let working:Document|undefined,conflict:RecoverySession['conflict']; + try{working=replayPending(acknowledged,previous.ledger,options.registry);}catch(error){const rejected=reject(error);conflict=freeze({code:rejected.code,targets:rejected.targets});} + let committed=false; + return Object.freeze({baseRootRevision:root.rootRevision,canonical, + pendingOperationIds:Object.freeze([...previous.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)), + ...(conflict?{conflict}:{}), + async apply():Promise{ + assertCurrent(); + 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,acknowledged,versions:canonical.versions,conflicts:readMap(conflicts)}),root.rootRevision); + if(result.status==='rejected')return result; + committed=true; + } + // Persistence failure retains the hold; apply/cancel can retry this barrier without republishing. + await persist();assertCurrent();active.release(); + return{status:'applied',revision:editor.getSnapshot().revision,operationIds:[]}; + }, + async cancel(){if(recovery!==active)return;if(committed)await persist();active.release();}, + }); + }catch(error){active.release();throw error;} + }, models(boardId: Id, signal: AbortSignal) { if(closing||closed||!options.runs)return Promise.reject(fault('run_interpreter_unavailable')); if(!editor.getSnapshot().document.canvases.has(boardId))return Promise.reject(fault('canvas_missing')); @@ -290,7 +346,7 @@ export async function openProjectEngine(options: EngineOptions) { }, cancelRun(runId:Id):Result { try{ - if(closing||closed)throw fault('engine_closing');if(!options.runs)throw fault('run_interpreter_unavailable'); + assertWritable();if(!options.runs)throw fault('run_interpreter_unavailable'); const root=editor.getSnapshot(),run=root.runtime.runs.get(runId);if(!run)throw fault('run_missing'); const allowed=options.policy?.({type:'element.data',target:run.target,patch:{}},root.document);if(allowed&&!allowed.enabled)throw fault(allowed.code,allowed.targets); if([...root.runtime.ledger.operations.values()].some(record=>record.phase!=='resolved'&&record.intent.operationType==='run.cancel'&&record.intent.action.runId===runId))throw fault('run_action_pending'); @@ -299,7 +355,7 @@ export async function openProjectEngine(options: EngineOptions) { }, applyRunOutput(runId:Id,target:RunApplyInput['target'],outputId:Id):Result { try{ - if(closing||closed)throw fault('engine_closing');if(!options.runs)throw fault('run_interpreter_unavailable');if(applying)throw fault('reentrant_dispatch'); + assertWritable();if(!options.runs)throw fault('run_interpreter_unavailable');if(applying)throw fault('reentrant_dispatch'); const root=editor.getSnapshot(),run=root.runtime.runs.get(runId),element=root.document.elements.get(target.id); if(!run||run.blocked||!runContainsTarget(run,target)||!element||element.incarnation!==target.incarnation)throw fault('target_changed'); if((run.appliedOutputIds??[]).includes(outputId))throw fault('output_already_applied'); @@ -315,7 +371,7 @@ export async function openProjectEngine(options: EngineOptions) { }, startRun(input: RunInput): Result { try { - if(closing||closed)throw fault('engine_closing'); + assertWritable(); if(!options.runs)throw fault('run_interpreter_unavailable'); validateRunInput(input); const root=editor.getSnapshot(),element=root.document.elements.get(input.target.id); @@ -335,11 +391,9 @@ export async function openProjectEngine(options: EngineOptions) { refresh(document: Document, versions: ReadonlyMap): Result { const root = editor.getSnapshot(), previous = root.runtime; try { - if (closing || closed) throw fault('engine_closing'); - if (document.projectId !== root.document.projectId) throw fault('project_mismatch'); + assertWritable(); if ([...previous.ledger.requests.values()].some(request => request.phase !== 'resolved')) throw fault('refresh_waiting_for_receipts'); - for (const canvas of document.canvases.keys()) if (!Number.isSafeInteger(versions.get(canvas)) || versions.get(canvas)! < 1) throw fault('missing_server_version', [canvas]); - for (const [id, version] of previous.versions) if ((versions.get(id) ?? -1) < version) throw fault('stale_canonical_version', [id]); + validateCanonicalRead(previous.acknowledged,previous.versions,document,versions,options.registry); if (equal([...previous.versions], [...versions]) && equal([...previous.acknowledged.canvases], [...document.canvases]) && equal([...previous.acknowledged.elements], [...document.elements]) && equal([...previous.acknowledged.tombstones], [...document.tombstones])) return { status: 'applied', revision: root.revision, operationIds: [] }; const acknowledged = createDocument(document.projectId, [...document.canvases.values()], [...document.elements.values()], document.tombstones); const working = replayPending(acknowledged, previous.ledger, options.registry); @@ -347,12 +401,12 @@ export async function openProjectEngine(options: EngineOptions) { return editor.reconcile(working, Object.freeze({ ...previous, acknowledged, versions: readMap(versions), conflicts: readMap(conflicts) }), root.rootRevision); } catch (error) { const result = reject(error); - if (result.code !== 'refresh_waiting_for_receipts' && result.code !== 'engine_closing') editor.updateRuntime(previous => { const conflicts = new Map(previous.conflicts); conflicts.set('refresh', result.code); return Object.freeze({ ...previous, conflicts: readMap(conflicts) }); }); + if (!['refresh_waiting_for_receipts','engine_closing','recovery_in_progress'].includes(result.code)) editor.updateRuntime(previous => { const conflicts = new Map(previous.conflicts); conflicts.set('refresh', result.code); return Object.freeze({ ...previous, conflicts: readMap(conflicts) }); }); return result; } }, stageFiles(files: readonly StagedFile[], callback: (resources: readonly ResourceHandle[], transaction: Transaction) => unknown) { - if (closing || closed) throw fault('engine_closing'); + assertWritable(); if (staged.length) throw fault('reentrant_dispatch'); if (!options.assets) throw fault('asset_interpreter_unavailable'); const resources = files.map(file => { @@ -380,11 +434,12 @@ export async function openProjectEngine(options: EngineOptions) { }, getSnapshot: editor.getSnapshot, persist, getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }), - retry: async () => { await scheduler.resume();await runTracker.retry(); }, + retry: async () => { assertWritable();await scheduler.resume();await runTracker.retry(); }, async settle() { await Promise.resolve(); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); }, async close() { if (closed) return; closing = true; sessions.dispose(); scheduler.stop(); runTracker.stop(); + recovery?.controller.abort();recovery?.release(); for (const release of [...displays]) release(); // Failure keeps the leader/recovery material available for an explicit retry of close. await persist(); unsubscribe(); editor.dispose(); await leader.release(); closed = true; diff --git a/src/lib/canvas-runtime/reconcile.ts b/src/lib/canvas-runtime/reconcile.ts index 22b6117..1202150 100644 --- a/src/lib/canvas-runtime/reconcile.ts +++ b/src/lib/canvas-runtime/reconcile.ts @@ -3,6 +3,26 @@ import { applyChanges, validateDocument } from '../canvas-core/document.ts'; import { equal, fault } from '../canvas-core/values.ts'; import type { LedgerSnapshot } from './ledger.ts'; +/** A complete read cannot rewind known scopes or change an unchanged-version canvas. */ +export function validateCanonicalRead(previous: Document, previousVersions: ReadonlyMap, next: Document, versions: ReadonlyMap, registry: Registry) { + if(next.projectId!==previous.projectId)throw fault('project_mismatch'); + validateDocument(next,registry); + for(const [id,version]of versions)if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_canonical_version',[id]); + for(const canvas of next.canvases.keys())if(!versions.has(canvas))throw fault('missing_server_version',[canvas]); + const unchanged=new Set(); + for(const [id,version]of previousVersions){ + if((versions.get(id)??-1)previousVersions.get(id)===version)&& + (previous.tombstones.size!==next.tombstones.size||[...previous.tombstones].some(([id,incarnation])=>next.tombstones.get(id)!==incarnation)))throw fault('canonical_version_mismatch'); +} + /** The footprint follows ownership; geometry does not decide which canvas owns an object. */ export function operationCanvases(intent: OperationIntent, document: Document): readonly Id[] { if(intent.operationType==='asset.upload'||intent.operationType==='run.cancel')return[]; diff --git a/src/lib/canvas-runtime/runs.ts b/src/lib/canvas-runtime/runs.ts index fd8f8d4..57243e8 100644 --- a/src/lib/canvas-runtime/runs.ts +++ b/src/lib/canvas-runtime/runs.ts @@ -49,13 +49,14 @@ export function createRunTracker(options: { }) { const heap = createMinHeap<{ id: Id; due: number; token: number; sequence: number }>((a,b) => a.due-b.due || a.sequence-b.sequence); const tokens = new Map(); + const holds = new Set(); let sequence = 0, stopped = false, timer: unknown, active: { controller: AbortController; promise: Promise } | undefined; const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot alter cursor durability. */ } }; const eligible = (run: RunRecord) => !run.blocked && (run.hasMore || !runTerminal(run.status)); function arm() { if (timer !== undefined) options.clock.clearTimer(timer); timer = undefined; while (heap.peek() && tokens.get(heap.peek()!.id) !== heap.peek()!.token) heap.pop(); - if (stopped || active || !heap.peek()) return; + if (stopped || holds.size || active || !heap.peek()) return; timer = options.clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0,heap.peek()!.due-options.clock.now())); } function notify(ids: readonly Id[]) { @@ -68,7 +69,7 @@ export function createRunTracker(options: { arm(); } function pump() { - if (stopped || active) return; + if (stopped || holds.size || active) return; let item = heap.pop(); while (item && tokens.get(item.id)!==item.token) item=heap.pop(); if (!item) return; @@ -100,6 +101,13 @@ export function createRunTracker(options: { active={controller,promise}; } return Object.freeze({ notify, + /** Holds compose; completing a poll still commits its cursor before settle() returns. */ + hold() { + if(stopped)throw fault('run_tracker_closed'); + const token=Symbol();holds.add(token); + if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined; + return()=>{if(holds.delete(token)&&!holds.size&&!stopped)arm();}; + }, async retry() { await options.persist();for(const id of options.getRuns().keys())options.update(id,run=>({...run,blocked:false,error:undefined,nextPollAt:options.clock.now()}));notify([...options.getRuns().keys()]); }, async settle() { while(active)await active.promise; }, stop() { stopped=true;if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined;active?.controller.abort();heap.clear(); }, diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index 099efa8..f214f43 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -34,6 +34,7 @@ export function createScheduler(options: SchedulerOptions) { 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); const tokens = new Map(), active = new Map; kind: FrozenRequest['kind'] }>(); + const holds = new Set(); const laneWaiters = new Map, Set>(), deferredRetries = new Map(); let timer: unknown, pumpTimer: unknown, enqueueSequence = 0, stopped = true, closed = false, pumping = false; const delay = (attempt: number) => Math.max(1, options.retryDelay?.(attempt) ?? Math.min(30_000, 500 * 2 ** Math.min(attempt, 6))); @@ -84,11 +85,11 @@ export function createScheduler(options: SchedulerOptions) { function armTimer() { if (timer !== undefined) { clock.clearTimer(timer); timer = undefined; } while (retry.peek() && !validRetry(retry.peek()!)) retry.pop(); - if (stopped || active.size >= concurrency || !retry.peek()) return; + if (stopped || holds.size || active.size >= concurrency || !retry.peek()) return; timer = clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0, retry.peek()!.dueAt - clock.now())); } function launch(id: Id, action: 'submit' | 'query') { - if (active.has(id) || stopped) return; + if (active.has(id) || stopped || holds.size) return; const controller = new AbortController(); // Start in a microtask so the handle exists before any synchronous adapter error. const promise = Promise.resolve().then(async () => { @@ -126,7 +127,7 @@ export function createScheduler(options: SchedulerOptions) { active.set(id, { controller, promise, kind: options.getLedger().requests.get(id)?.kind }); } function pump() { - if (stopped || pumping) return; + if (stopped || holds.size || pumping) return; pumping = true; try { while (active.size < concurrency && retry.peek() && retry.peek()!.dueAt <= clock.now()) { @@ -163,7 +164,7 @@ export function createScheduler(options: SchedulerOptions) { } } finally { pumping = false; armTimer(); - if (!stopped && ready.size && active.size < concurrency && pumpTimer === undefined) pumpTimer = clock.setTimer(() => { pumpTimer = undefined; pump(); }, 0); + if (!stopped && !holds.size && ready.size && active.size < concurrency && pumpTimer === undefined) pumpTimer = clock.setTimer(() => { pumpTimer = undefined; pump(); }, 0); } } return Object.freeze({ @@ -182,6 +183,18 @@ export function createScheduler(options: SchedulerOptions) { pump(); }, notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); pump(); }, + /** Acquire synchronously, then settle(): already admitted work completes with its original identity. */ + hold() { + if (closed) throw fault('scheduler_closed'); + const token = Symbol(); holds.add(token); + if (timer !== undefined) clock.clearTimer(timer); + if (pumpTimer !== undefined) clock.clearTimer(pumpTimer); + timer = pumpTimer = undefined; + return () => { + if (!holds.delete(token) || holds.size || stopped) return; + pump(); + }; + }, /** Explicit retry after storage/auth/permission recovery; preserves frozen bodies. */ async resume() { if (stopped) return; diff --git a/src/lib/canvas-service/foundry.ts b/src/lib/canvas-service/foundry.ts index 9f319db..8730cfc 100644 --- a/src/lib/canvas-service/foundry.ts +++ b/src/lib/canvas-service/foundry.ts @@ -114,6 +114,12 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id; } return Object.freeze({ supportsRequest: (request: FrozenRequest) => request.protocolVersion === 2, + async read(signal: AbortSignal) { + const loaded=await loadFoundryModel(options.transport,options.workspaceId,options.rootId,options.registry,signal); + if(loaded.status!=='ready')throw fault('canonical_migration_required'); + if(loaded.document.projectId!==options.rootId)throw fault('project_mismatch'); + return loaded; + }, compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot) { if (snapshot.document.projectId !== options.rootId || !uuid.test(intent.id) || !uuid.test(requestId) || intent.operationType !== 'document.change' || intent.operationSchemaVersion !== 1) throw fault('operation_interpreter_unavailable'); const canvasIds = [...new Set(intent.canvasIds)].sort(), expectedVersions: Record = {}; diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index 201190d..e8d8ef2 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -5,6 +5,29 @@ import { createFoundryRuns } from '../../src/lib/canvas-service/runs.ts'; import { createFoundryAssets } from '../../src/lib/canvas-service/assets.ts'; import { createFoundryPersistence } from '../../src/lib/canvas-service/foundry.ts'; +async function canonicalRefreshCase(name,assert){ + const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()); + const target={id:'a',incarnation:1},element={...target,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:root,geometry:{x:0,y:0,width:300,height:200,rotation:0},data:{title:'a',text:'original'}}; + const original=createDocument(root,[{id:root,parent:null}],[element]),storage=createBrowserStorage({name:`${name}-canonical-refresh`}),calls=[];let server=original,version=1,engine; + const model=complete=>({documentFormatVersion:1,projectId:root,complete,removedCanvasIds:[],canvases:[...server.canvases.values()],elements:[...server.elements.values()],tombstones:{},versions:{[root]:version}}); + const transport=async request=>{calls.push(request);if(request.method==='GET')return{status:200,body:{model:model(true)}};const body=JSON.parse(request.body);for(const operation of body.operations)server=applyChanges(server,operation.changes);version++;return{status:200,body:{protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(operation=>operation.id),model:model(false)}};}; + const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1]]),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})}; + try{ + engine=await openProjectEngine(options);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle(); + const historyId=engine.getSnapshot().history.undo[0],request=[...engine.getSnapshot().runtime.ledger.requests.values()][0]; + server=applyChanges(server,[{kind:'field',target,field:'geometry',before:element.geometry,after:{...element.geometry,x:400}}]);version++; + const session=await engine.prepareRecovery();assert(session.conflict===undefined,'clean canonical refresh reported a conflict'); + assert(engine.getSnapshot().document.elements.get('a').geometry.x===0,'preview mutated the working document'); + assert(engine.editor.history.undo().code==='recovery_in_progress','history bypassed the recovery fence'); + assert((await session.apply()).status==='applied','fresh canonical state did not publish');await engine.close(); + engine=await openProjectEngine(options);const restored=engine.getSnapshot(); + assert(restored.document.elements.get('a').geometry.x===400&&restored.document.elements.get('a').data.title==='local','canonical refresh did not survive IndexedDB recovery'); + assert(restored.runtime.versions.get(root)===3&&restored.history.undo[0]===historyId,'refresh lost its version/history cut'); + assert(restored.runtime.ledger.requests.size===1&&restored.runtime.ledger.requests.get(request.id).body===request.body,'refresh changed an original request'); + assert(calls.filter(call=>call.method==='POST').length===1,'refresh or reopen resent an applied write'); + }finally{await engine?.close();storage.close();indexedDB.deleteDatabase(`${name}-canonical-refresh`);} +} + async function runRecoveryCase(name,assert){ const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),runId=crypto.randomUUID(); const registry=createRegistry(builtinComponents()),element={id:'a',incarnation:1,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:root,geometry:{x:0,y:0,width:300,height:200,rotation:0},data:{title:'a',text:'original'}},original=createDocument(root,[{id:root,parent:null}],[element]); @@ -205,6 +228,8 @@ export async function run() { results.push('protocol 2 service adapter recovers exact frozen bytes and reconciles a lost ACK through real IndexedDB'); await runRecoveryCase(name,assert); results.push('AI admission and RunRecord recover through real IndexedDB; cursor and generated content share a durable cut without a second invocation'); + await canonicalRefreshCase(name,assert); + results.push('quiescent canonical refresh preserves original requests and history; the new base/document/version cut survives IndexedDB reopening'); return results; } finally { await leader?.release(); other?.close(); storage.close(); indexedDB.deleteDatabase(name); } } diff --git a/tests/headless/engine.test.mjs b/tests/headless/engine.test.mjs index 880af38..0e8a195 100644 --- a/tests/headless/engine.test.mjs +++ b/tests/headless/engine.test.mjs @@ -140,3 +140,67 @@ test('a fresh canonical read replaces a clean recovered base, keeps history and assert.equal(reopened.engine.getSnapshot().document.elements.get('a').geometry.x, 400); await reopened.engine.close(); }); + +test('recovery waits for admitted writes, reads a complete base, fences edits and publishes without changing history', async()=>{ + const storage=storageFixture(),port=persistenceFixture(),response=deferred(),realSubmit=port.submit,readGate=deferred(); + let reads=0;port.submit=async request=>{await response.promise;return realSubmit(request);}; + port.read=async()=>{reads++;await readGate.promise;return{document:applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'local',beforePresent:true,afterPresent:true},{kind:'field',target,field:'geometry',before:original.elements.get('a').geometry,after:{...original.elements.get('a').geometry,x:300}}]),versions:new Map([['canvas',3]])};}; + const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await tick(); + const preparing=engine.prepareRecovery();await tick();assert.equal(reads,0); + assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'blocked'}}).code,'recovery_in_progress'); + assert.equal(engine.startRun({}).code,'recovery_in_progress');assert.equal(engine.refresh(original,new Map([['canvas',1]])).code,'recovery_in_progress'); + await assert.rejects(engine.prepareRecovery(),/recovery_in_progress/); + response.resolve();await tick();assert.equal(reads,1);const history=engine.getSnapshot().history; + readGate.resolve();const session=await preparing;assert.equal(session.conflict,undefined);assert.equal(session.canonical.document.elements.get('a').geometry.x,300); + assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,0);assert.equal(engine.editor.history.undo().code,'recovery_in_progress'); + assert.equal((await session.apply()).status,'applied');assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,300);assert.equal(engine.getSnapshot().history,history); + assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);await assert.rejects(session.apply(),/recovery_session_closed/); + await engine.close(); +}); + +test('unknown receipts prevent a recovery read and preserve the frozen request identity',async()=>{ + const storage=storageFixture();let reads=0;const port=persistenceFixture({async submit(){return{outcome:'unresolved'};},async read(){reads++;return{document:original,versions:new Map([['canvas',1]])};}}); + const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle(); + const before=engine.getSnapshot(),request=[...before.runtime.ledger.requests.values()][0]; + await assert.rejects(engine.prepareRecovery(),/recovery_waiting_for_receipts/);assert.equal(reads,0); + const after=engine.getSnapshot();assert.equal(after.document,before.document);assert.equal(after.history,before.history);assert.equal(after.runtime.ledger.requests.get(request.id).body,request.body); + assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'allowed'}}).enabled,true);await engine.close(); +}); + +test('recovery conflict previews preserve local data/history and cancel releases the edit fence',async()=>{ + const remote=applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'remote',beforePresent:true,afterPresent:true}]); + const storage=storageFixture(),port=persistenceFixture({async submit(){return{outcome:'not-applied',error:'version_conflict'};},async read(){return{document:remote,versions:new Map([['canvas',2]])};}}); + const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle(); + const before=engine.getSnapshot(),session=await engine.prepareRecovery();assert.ok(session.conflict);assert.equal(session.pendingOperationIds.length,1); + assert.equal((await session.apply()).status,'rejected');assert.equal(engine.getSnapshot(),before);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'local'); + await session.cancel();await session.cancel();assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'more'}}).enabled,true);await engine.close(); +}); + +test('failed recovery durability retains its fence and retries the same published root',async()=>{ + const storage=storageFixture(),port=persistenceFixture({async read(){return{document:original,versions:new Map([['canvas',1]])};}}),{engine}=await open(storage,port); + const session=await engine.prepareRecovery();storage.broken=true;await assert.rejects(session.apply(),/quota/);const revision=engine.getSnapshot().rootRevision; + assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'blocked'}}).code,'recovery_in_progress');await assert.rejects(session.cancel(),/quota/); + storage.broken=false;assert.equal((await session.apply()).status,'applied');assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal(engine.getDurability().durableRootRevision,revision); + await engine.close(); +}); + +test('closing the engine aborts a recovery read and rejects a late canonical response',async()=>{ + const storage=storageFixture(),response=deferred();let signal; + const port=persistenceFixture({async read(input){signal=input;return response.promise;}}),{engine}=await open(storage,port); + const preparing=engine.prepareRecovery();await tick();assert.ok(signal);await engine.close();assert.equal(signal.aborted,true); + response.resolve({document:original,versions:new Map([['canvas',1]])});await assert.rejects(preparing,/engine_closing/);assert.equal(storage.owner(),null); +}); + +test('complete recovery reads reject same-version rewrites and lifecycle regression before creating a session',async()=>{ + const remote=applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'remote',beforePresent:true,afterPresent:true}]); + const storage=storageFixture(),port=persistenceFixture({async read(){return{document:remote,versions:new Map([['canvas',1]])};}}),{engine}=await open(storage,port); + await assert.rejects(engine.prepareRecovery(),/canonical_version_mismatch/); + assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'a'); + assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'allowed'}}).enabled,true); + assert.equal(engine.refresh(remote,new Map([['canvas',1]])).code,'canonical_version_mismatch'); + const {validateCanonicalRead}=await import('../../src/lib/canvas-runtime/reconcile.ts'); + const withTombstone=createDocument(original.projectId,[...original.canvases.values()],[...original.elements.values()],new Map([['deleted',2]])); + assert.throws(()=>validateCanonicalRead(withTombstone,new Map([['canvas',1]]),original,new Map([['canvas',2]]),registry),/stale_canonical_lifecycle/); + assert.throws(()=>validateCanonicalRead(original,new Map([['canvas',1]]),withTombstone,new Map([['canvas',1]]),registry),/canonical_version_mismatch/); + await engine.close(); +}); diff --git a/tests/headless/run-tracker.test.mjs b/tests/headless/run-tracker.test.mjs new file mode 100644 index 0000000..2004af7 --- /dev/null +++ b/tests/headless/run-tracker.test.mjs @@ -0,0 +1,27 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRunTracker } from '../../src/lib/canvas-runtime/runs.ts'; + +const tick = async () => { for (let index=0;index<30;index++)await Promise.resolve(); }; +const deferred = () => { let resolve;const promise=new Promise(yes=>{resolve=yes;});return{promise,resolve}; }; +function fixture() { + let now=0,next=0;const timers=new Map(),runs=new Map([['run',{id:'run',status:'running',cursor:0,hasMore:true,attempt:0}]]),calls=[],saved=[],response=deferred(); + const tracker=createRunTracker({clock:{now:()=>now,setTimer(callback,delay){const id=++next;timers.set(id,{callback,due:now+delay});return id;},clearTimer:id=>timers.delete(id)},getRuns:()=>runs,async persist(){saved.push(structuredClone([...runs]));},async poll(run){calls.push(run);return calls.length===1?response.promise:{status:'succeeded',events:[],next:2,hasMore:false,outputs:[],steps:[]};},accept(run,page){runs.set(run.id,{...runs.get(run.id),status:page.status,cursor:page.next,hasMore:page.hasMore});},update(id,reduce){runs.set(id,reduce(runs.get(id)));},onError(error){throw error;}}); + return{tracker,runs,calls,saved,response,timers,async advance(ms){now+=ms;for(const[id,timer]of[...timers])if(timer.due<=now){timers.delete(id);timer.callback();}await tick();}}; +} + +test('run holds preserve a completed cursor and notifications without launching another poll',async()=>{ + const f=fixture();f.tracker.notify(['run']);await f.advance(0);assert.equal(f.calls.length,1); + const releaseA=f.tracker.hold(),releaseB=f.tracker.hold(); + f.response.resolve({status:'running',events:[],next:1,hasMore:true,outputs:[],steps:[]});await f.tracker.settle(); + assert.equal(f.runs.get('run').cursor,1);assert.equal(f.saved.at(-1)[0][1].cursor,1);assert.equal(f.timers.size,0); + await f.advance(10000);await f.tracker.retry();assert.equal(f.calls.length,1);assert.equal(f.timers.size,0); + releaseA();releaseA();await f.advance(0);assert.equal(f.calls.length,1); + releaseB();await f.advance(0);await f.tracker.settle();assert.equal(f.calls.length,2);assert.equal(f.calls[1].cursor,1);assert.equal(f.runs.get('run').cursor,2); + f.tracker.stop(); +}); + +test('releasing a run hold after stop cannot restart a poll',async()=>{ + const f=fixture(),release=f.tracker.hold();f.tracker.notify(['run']);f.tracker.stop();release();await f.advance(10000); + assert.equal(f.calls.length,0);assert.equal(f.timers.size,0);assert.throws(()=>f.tracker.hold(),/run_tracker_closed/); +}); diff --git a/tests/headless/scheduler.test.mjs b/tests/headless/scheduler.test.mjs index 39fde89..b3d1d49 100644 --- a/tests/headless/scheduler.test.mjs +++ b/tests/headless/scheduler.test.mjs @@ -143,3 +143,26 @@ test('a capacity-blocked upload retry does not spin or hide a later document ret gate.resolve(); await state.scheduler.settle(); assert.ok(calls.some(([action, id]) => action === 'query' && id === 'asset-1')); state.scheduler.stop(); }); + +test('nested holds finish admitted receipts but suppress dependent dispatch and due retries until the final release', async () => { + const response = deferred(), calls = []; + const state = setup({ concurrency: 1, submit: async request => { calls.push(['submit', request]); return request.operationIds[0] === 'first' ? response.promise : { outcome: 'applied' }; }, query: async request => { calls.push(['query', request]); return { outcome: 'applied' }; } }); + state.add('first'); state.add('dependent', ['canvas'], ['first']); state.scheduler.start(); await tick(); + const releaseA = state.scheduler.hold(), releaseB = state.scheduler.hold(); + state.add('later', ['other']); response.resolve({ outcome: 'unresolved' }); await state.scheduler.settle(); + const frozen = state.ledger().getSnapshot().requests.get('request-1'); + assert.equal(frozen.outcome, 'unresolved'); assert.equal(state.scheduler.getClaims().get('canvas'), frozen.id); + await state.advance(1000); assert.equal(calls.length, 1); assert.equal(state.timerCount(), 0); + releaseA(); releaseA(); await tick(); assert.equal(calls.length, 1); + releaseB(); await state.scheduler.settle(); + assert.deepEqual(calls.map(([action, request]) => [action, request.operationIds[0]]), [['submit', 'first'], ['query', 'first'], ['submit', 'later'], ['submit', 'dependent']]); + assert.equal(calls[1][1].id, frozen.id); assert.equal(calls[1][1].body, frozen.body); assert.equal(state.scheduler.getClaims().size, 0); + state.scheduler.stop(); +}); + +test('holds acquired before start retain notified work; stop is terminal even when a hold is later released', async () => { + const state = setup(), release = state.scheduler.hold(); state.add('first'); state.scheduler.start(); await tick(); + assert.equal(state.calls.length, 0); release(); await state.scheduler.settle(); assert.equal(state.calls.length, 1); + const finalRelease = state.scheduler.hold(); state.add('never'); state.scheduler.stop(); finalRelease(); await tick(); + assert.equal(state.calls.length, 1); assert.throws(() => state.scheduler.hold(), /scheduler_closed/); +}); diff --git a/tests/headless/service.test.mjs b/tests/headless/service.test.mjs index 691577d..965d2e8 100644 --- a/tests/headless/service.test.mjs +++ b/tests/headless/service.test.mjs @@ -98,3 +98,16 @@ test('shared service transport binds the immutable AI admission version header', for(const value of [0,-1,1.5,NaN,Number.MAX_SAFE_INTEGER+1])await assert.rejects(()=>transport({method:'POST',path:'fixture',expectedVersion:value}),/invalid_server_version/); assert.equal(calls.length,1); }); + +test('recovery reader requires a complete canonical model for its bound project and forwards cancellation',async()=>{ + const controller=new AbortController(),calls=[]; + const{port}=setup(async request=>{calls.push(request);return{status:200,body:{model:chunk()}};}); + const loaded=await port.read(controller.signal);assert.equal(loaded.document.projectId,root);assert.equal(loaded.versions.get(child),7); + assert.equal(calls[0].method,'GET');assert.equal(calls[0].signal,controller.signal);assert.equal(calls[0].body,undefined);assert.equal(calls[0].requestId,undefined); + for(const[response,error]of[ + [{status:200,body:{model:chunk(false)}},/incomplete_initial_model/], + [{status:409,body:{error:'canonical_migration_required',document_format_version:1,root_id:root,source_versions:{[root]:1}}},/canonical_migration_required/], + [{status:200,body:{model:{documentFormatVersion:1,projectId:child,canvases:[{id:child,parent:null}],elements:[],tombstones:{},versions:{[child]:1},complete:true,removedCanvasIds:[]}}},/project_mismatch/], + [{status:403,body:{error:'forbidden'}},/forbidden/], + ]){const{port:invalid}=setup(async()=>response);await assert.rejects(invalid.read(controller.signal),error);} +});