Add quiescent canonical recovery sessions without rewriting requests

2026-11-08 01:41:23 +08:00
parent 0c7fae6e32
commit a0417b4811
11 changed files with 276 additions and 20 deletions
@@ -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<Result>;
+  cancel(): Promise<void>;
+};

 /** 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<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
    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<RecoverySession> {
+        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<Result>{
+              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<Id, number>): 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;