193 lines
22 KiB
TypeScript
193 lines
22 KiB
TypeScript
/**
|
|
* 用户明确选择后的冲突恢复规划，不是后台静默重试。
|
|
* 以校验过的服务端文档为起点，逐项保留或丢弃本地未决意图，重建文档、历史及 ledger，
|
|
* 同时保存旧文档备份和决策审计。此模块只返回方案，真正发布和持久化由 engine 协调。
|
|
* 保留本地意图时创建新操作身份；原操作/请求证据不会被改写成一次看似未发生的请求。
|
|
*/
|
|
import {validateLoadingCompletion} from './loading.ts';
|
|
import { withLocalLifecycles } from './elision.ts';
|
|
import type { Change, Document, FieldChange, HistoryState, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
|
|
import { describeHistoryTransition, historyTransitionChanges, type HistoryTransition } from '../canvas-core/history-transition.ts';
|
|
import { invert } from '../canvas-core/document.ts';
|
|
import { planHistoryChanges } from '../canvas-core/history.ts';
|
|
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
|
|
import type { CanonicalProposal, EngineSnapshot } from './contracts.ts';
|
|
import { createLedger, type Presence } from './ledger.ts';
|
|
import { intentContexts, rebaseIntent } from './change-rebase.ts';
|
|
import type { HistoryReplay } from './history-replay.ts';
|
|
import { createHistoryJournalRecovery } from './history-journal-recovery.ts';
|
|
import { replanTraversals } from './traversal-recovery.ts';
|
|
import { operationCanvases, validateCanonicalRead } from './reconcile.ts';
|
|
import { referencedResources } from './resources.ts';
|
|
|
|
/** 每个待处理操作都必须有用户选择；不隐式替未选择的依赖作决定。 */
|
|
export type RecoveryChoice = 'keep-local' | 'discard-local';
|
|
/** 旧操作的结论：被新操作替代、被丢弃，或目标已满足而不需要新操作。 */
|
|
export type RecoveryDecision = { readonly operationId: Id; readonly choice: RecoveryChoice; readonly replacementId?: Id; readonly resolution: 'superseded' | 'discarded' | 'satisfied' };
|
|
/** 恢复审计记录：绑定规划时的根 revision，保留备份及可用的历史重放证据。 */
|
|
export type RecoveryAudit = { readonly id: Id; readonly formatVersion: 1; readonly baseRootRevision: number; readonly decisions: readonly RecoveryDecision[]; readonly backup: Json; readonly historyTransition?:HistoryTransition; readonly historyReplay?:HistoryReplay };
|
|
/** 待采纳的文档/历史/操作/账本候选及审计，非完整 EngineSnapshot；引擎还需结合 canonical 版本等事实发布。
|
|
* baseRootRevision 用于拒绝过时方案，resourceIds 是本次已知保留集合，operationIds 只列新替代操作。 */
|
|
export type ConflictPlan = { readonly id: Id; readonly baseRootRevision: number; readonly document: Document; readonly history: HistoryState; readonly operations: EngineSnapshot['operations']; readonly ledger: EngineSnapshot['runtime']['ledger']; readonly audit: RecoveryAudit; readonly resourceIds: readonly Id[]; readonly operationIds: readonly Id[] };
|
|
/** 字段身份包括元素代次，避免把删除后重建的同 ID 元素算成同一字段历史。 */
|
|
const fieldKey = (change: FieldChange) => JSON.stringify([change.target.id,change.target.incarnation,change.field]);
|
|
/** 读取恢复起点的字段值，并区分 data 字段缺失与显式 null。
|
|
* allowMissing 仅供历史分析记录目标缺失，不将旧代次目标重定向到新元素。 */
|
|
function value(document: Document, change: FieldChange, allowMissing=false): {before:Json;beforePresent?:boolean;missing?:boolean} {
|
|
const element=document.elements.get(change.target.id);
|
|
if(!element||element.incarnation!==change.target.incarnation){if(allowMissing)return{before:null,beforePresent:false,missing:true};throw fault('stale_incarnation',[change.target.id]);}
|
|
const key=change.field.startsWith('data.')?change.field.slice(5):undefined;
|
|
return {before:(key===undefined?(element as unknown as Record<string,Json>)[change.field]:element.data[key]??null),beforePresent:key===undefined?undefined:Object.hasOwn(element.data,key)};
|
|
}
|
|
/** 把文档 Map 转成备份用的 JSON 形状；序列化的是数据，不是运行时方法或索引。 */
|
|
const documentValue = (document: Document) => ({projectId:document.projectId,canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:[...document.tombstones]});
|
|
|
|
/** 在不可变 snapshot 上规划恢复：先验 canonical 单调性/完整性，并要求 ledger 中请求 phase 全为 resolved。
|
|
* 不自己查询回执、认证来源或等待磁盘；canonical/账本已捕获且可信由 engine/端口负责。
|
|
* 取全部非 resolved 且非 asset.upload 意图，要求 choices 精确覆盖每项；依次重规划、重建 history、退休旧项并加入替代项。
|
|
* 不发布/持久化/提交请求；createId 可消耗外部编号，失败不会回滚该回调副作用。
|
|
* 小输入/模型按既有不可变合同使用，备份 JSON 编码与全部恢复规划同步完成，没有时间片/容量预算。 */
|
|
export function planConflictRecovery(snapshot: EngineSnapshot, canonical: CanonicalProposal, choices: ReadonlyMap<Id,RecoveryChoice>, registry: Registry, createId:()=>Id,historyLimit=100): ConflictPlan {
|
|
if(snapshot.runtime.loading)validateLoadingCompletion(snapshot.runtime.acknowledged,snapshot.runtime.versions,snapshot.runtime.loading,canonical.document,canonical.versions,registry,true);else validateCanonicalRead(snapshot.runtime.acknowledged,snapshot.runtime.versions,canonical.document,canonical.versions,registry);
|
|
if([...snapshot.runtime.ledger.requests.values()].some(request=>request.phase!=='resolved'))throw fault('recovery_waiting_for_receipts');
|
|
// 上传资源不在这次文档取舍中；其它所有未决操作都纳入选择集合，并检查依赖的时间顺序。
|
|
const records=[...snapshot.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved'&&record.intent.operationType!=='asset.upload').sort((a,b)=>a.intent.sequence-b.intent.sequence);
|
|
if(!records.length)throw fault('recovery_nothing_to_resolve');
|
|
for(const record of records)for(const dependency of record.intent.dependencies)if((snapshot.runtime.ledger.operations.get(dependency)?.intent.sequence??Infinity)>=record.intent.sequence)throw fault('recovery_dependency_order',[record.intent.id]);
|
|
if(choices.size!==records.length||records.some(record=>!['keep-local','discard-local'].includes(choices.get(record.intent.id)??'')))throw fault('recovery_choices_incomplete');
|
|
// 新审计和替代操作都在同一命名空间避碰，包含旧历史、请求和恢复记录中的身份。
|
|
const used=new Set([...snapshot.operations.keys(),...snapshot.runtime.ledger.operations.keys(),...snapshot.runtime.ledger.requests.keys(),...snapshot.history.entries.keys(),...(snapshot.runtime.recoveries?.keys()??[])]);
|
|
/** 每次分配都登记使用过的 ID，拒绝生成器重复值；不靠覆盖 Map 来解决碰撞。 */
|
|
function unique(){const id=createId();if(!id||used.has(id))throw fault('duplicate_operation_id');used.add(id);return id;}
|
|
const id=unique(),ledger=createLedger(snapshot.runtime.ledger),operations=new Map(snapshot.operations),decisions:RecoveryDecision[]=[],replacements=new Map<Id,OperationIntent|undefined>();
|
|
// 边界记录历史单元恢复前后的文档；starts 保存首次触及字段时的基线，供混合 ACK 历史核对。
|
|
const boundaries=new Map<Id,{before:Document;after:Document}>(),starts=new Map<string,ReturnType<typeof value>>(),historyOwner=new Map<Id,Id>(),contexts=intentContexts(snapshot.document,records.map(record=>record.intent));
|
|
for(const entry of snapshot.history.entries.values())for(const operationId of entry.operationIds){if(historyOwner.has(operationId))throw fault('recovery_history_ambiguous');historyOwner.set(operationId,entry.id);}
|
|
// 全部 content 有原生 transition、某个遍历有 transition，或命中已有 adoption 时，进入严格日志恢复。
|
|
// 这只是分支选择，不证明证据齐全/可回放；journal 构造和后续 project/accept 仍可拒绝不完整/矛盾记录。
|
|
const adopted=new Set([...snapshot.runtime.recoveries.values()].filter(audit=>audit.historyReplay!==undefined).flatMap(audit=>audit.decisions.flatMap(decision=>decision.replacementId?[decision.replacementId]:[])));
|
|
const content=records.filter(record=>record.intent.changes.length),hasEvidence=content.length>0&&content.every(record=>record.intent.historyTransition)||records.some(record=>['undo','redo'].includes(record.intent.origin)&&record.intent.historyTransition)||content.some(record=>adopted.has(record.intent.id));
|
|
const journal=hasEvidence?createHistoryJournalRecovery(snapshot,records.map(record=>record.intent),registry,historyLimit):undefined;
|
|
// 权威文档叠加本地生命周期事实，避免尚未发出的创建/删除身份被错误复活。
|
|
let document=withLocalLifecycles(canonical.document,snapshot.runtime.ledger),sequence=0;for(const record of snapshot.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence);
|
|
// 串行投影使后一个意图看到前一项保留/丢弃的结果，而不是分别套在同一旧快照上。
|
|
for(const record of records){
|
|
const intent=record.intent,choice=choices.get(intent.id)!;
|
|
if(intent.operationSchemaVersion!==1)throw fault('operation_interpreter_unavailable',[intent.id]);
|
|
if(intent.changes.length&&!['user','undo','redo'].includes(intent.origin)&&!(intent.origin==='asset-resolution'&&intent.historyEffect))throw fault('recovery_history_traversal_not_supported',[intent.id]);
|
|
const before=document,entryId=historyOwner.get(intent.id);
|
|
if(entryId&&!boundaries.has(entryId))boundaries.set(entryId,{before,after:before});
|
|
for(const original of intent.changes)if(original.kind==='field'){
|
|
const owner=historyOwner.get(intent.id),key=owner?JSON.stringify([owner,fieldKey(original)]):undefined;
|
|
if(key&&!starts.has(key))starts.set(key,value(document,original,true));
|
|
}
|
|
// 用户新建 canvas 的身份若已被服务端占用，不能悄悄视为同一对象。
|
|
if(choice==='keep-local'&&intent.origin==='user')for(const change of intent.changes)if(change.kind==='canvas'&&!change.before&&canonical.versions.has(change.id))throw fault('recovery_identity_collision',[change.id]);
|
|
// 每项边界 before→after 反映此前全部审阅结果；keep 在该新候选上规划，discard 不施加原文档变化。
|
|
// journal.project 还可能先推进已有结算历史；没有 journal 时直接走 rebaseIntent，不能并行按同一旧基线处理。
|
|
const rebased=journal?journal.project(document,intent,contexts.get(intent.id)!,choice==='keep-local'):choice==='keep-local'?rebaseIntent(document,intent,contexts.get(intent.id)!,registry):{document,changes:[]};
|
|
const changes=rebased.changes;document=rebased.document;
|
|
if(entryId)boundaries.get(entryId)!.after=document;
|
|
let replacement:OperationIntent|undefined;
|
|
if(choice==='keep-local'){
|
|
// 生成启动、取消等动作需重新发起命令；冲突审阅不能顺手重新执行可能付费的任务。
|
|
if(intent.operationType!=='document.change'&&intent.operationType!=='run.apply')throw fault('recovery_action_requires_new_command',[intent.id]);
|
|
if(changes.length){
|
|
// 原依赖若已审阅：改指对应 replacement，或在丢弃/已满足时移除；未审阅的既有依赖保留原 ID。
|
|
// 再依赖所有此前新 replacement，保证后项看到前项的串行效果；新 sequence 从账本最大值后递增。
|
|
const dependencyIds=intent.dependencies.flatMap(dependency=>replacements.has(dependency)?replacements.get(dependency)?[replacements.get(dependency)!.id]:[]:[dependency]);
|
|
// New dependencies reference new logical facts; a discarded parent never silently starts a paid run.
|
|
const dependencySet=new Set(dependencyIds);
|
|
for(const previous of replacements.values())if(previous)dependencySet.add(previous.id);
|
|
const replacementId=unique();
|
|
// 旧历史转移绑定旧状态，不能直接复制给新操作；保留其余逻辑属性并重算画布范围。
|
|
const {historyTransition:_oldHistory,...originalFact}=intent;
|
|
const basic={...originalFact,id:replacementId,sequence:++sequence,changes:freeze(changes),dependencies:[...dependencySet],canvasIds:operationCanvases({...intent,canvasIds:[],changes},before,document)};
|
|
replacement=intent.operationType==='run.apply'?freeze({...basic,operationType:'run.apply',action:{...intent.action,targetCanvasId:document.elements.get(intent.action.target.id)!.ownerCanvasId}}) as OperationIntent:freeze(basic);
|
|
operations.set(replacement.id,replacement);
|
|
}
|
|
}
|
|
journal?.accept(intent,replacement,choice==='keep-local',before,document);
|
|
replacements.set(intent.id,replacement);
|
|
// satisfied 只表示这次规划无需新 changes；不是在此收到另一份服务器 applied 回执。
|
|
decisions.push(freeze({operationId:intent.id,choice,resolution:choice==='discard-local'?'discarded':replacement?'superseded':'satisfied',...(replacement?{replacementId:replacement.id}:{})}));
|
|
}
|
|
// 根据现有证据选日志重放、撤销/重做重规划或兼容历史重写，拒绝无法证明安全的路径。
|
|
const pending=records.map(record=>record.intent),history=journal?journal.finish():pending.some(intent=>['undo','redo'].includes(intent.origin)&&intent.changes.length)?replanTraversals(snapshot,pending,replacements,new Set(decisions.filter(decision=>decision.choice==='keep-local').map(decision=>decision.operationId)),boundaries,registry):rewriteHistory(snapshot,pending,replacements,starts,boundaries,registry);
|
|
// 旧逻辑事实转为带审计的终结状态，新替代操作单独加入账本并记录存在/不存在基线。
|
|
ledger.retire(id,decisions);
|
|
for(const replacement of replacements.values())if(replacement){
|
|
const presence:Record<Id,Presence>={};for(const change of replacement.changes)if(change.kind==='field')presence[change.target.id]??='exists';else if(change.kind==='element')presence[change.id]??=change.before?'exists':'absent';
|
|
ledger.add(replacement,presence);
|
|
}
|
|
const historyReplay=journal?.evidence();
|
|
// 已知资源保留集合：先纳入本地 resources 全部 ID，再补原/替代 changes 与历史 replay 中可解释引用。
|
|
// 不在此推断 opaque payload 的所有私有引用，也不把资源 ID 清单当作已保存 Blob 字节的证明。
|
|
const resourceIds=new Set(snapshot.runtime.resources.keys());
|
|
for(const record of records)for(const resource of referencedResources(record.intent.changes,registry))resourceIds.add(resource);
|
|
for(const replacement of replacements.values())if(replacement)for(const resource of referencedResources(replacement.changes,registry))resourceIds.add(resource);
|
|
if(historyReplay)for(const transition of [historyReplay.anchor,...historyReplay.events.map(event=>event.transition)])for(const resource of referencedResources(historyTransitionChanges(transition),registry))resourceIds.add(resource);
|
|
// backup 捕获旧 working/ACK/版本/history 和资源 ID；不嵌入 Blob、服务回执或完整 workspace 快照。
|
|
// 顶层 audit transition 描述旧→新历史，journal evidence 另解释内部事件；两者都须与恢复 Root 同次保存。
|
|
const backup=JSON.parse(JSON.stringify({document:documentValue(snapshot.document),acknowledged:documentValue(snapshot.runtime.acknowledged),versions:[...snapshot.runtime.versions],resourceIds:[...resourceIds],history:{undo:snapshot.history.undo,redo:snapshot.history.redo,entries:[...snapshot.history.entries]}})) as Json;
|
|
const historyTransition=describeHistoryTransition(snapshot.history,history,snapshot.rootRevision+1,undefined,!!historyReplay);
|
|
const audit=freeze({id,formatVersion:1 as const,baseRootRevision:snapshot.rootRevision,decisions,backup,...(historyTransition?{historyTransition}:{}),...(historyReplay?{historyReplay}:{})});
|
|
return Object.freeze({id,baseRootRevision:snapshot.rootRevision,document,history,operations:readMap(operations),ledger:ledger.getSnapshot(),audit,resourceIds:Object.freeze([...resourceIds]),operationIds:Object.freeze([...replacements.values()].flatMap(intent=>intent?[intent.id]:[]))});
|
|
}
|
|
|
|
/** 无 journal 的兼容分支，仅重写受本批 pending 影响的 undo 项，且其关联操作都必须是已知 user origin。
|
|
* 全项未决时直接用 replacements 拼回；混合已确认贡献时，结构项要求确认贡献为完整前缀，纯字段按身份键分别审阅。
|
|
* 不生成完整事件 replay，不重规划 redo 或任意交错遍历；不满足证据条件即拒绝本次方案，交外层保持未发布。
|
|
* 返回私有 entry 表/undo 候选，未影响 entry 与 redo 数组沿用不可变快照引用。 */
|
|
function rewriteHistory(snapshot:EngineSnapshot,pending:readonly OperationIntent[],replacements:ReadonlyMap<Id,OperationIntent|undefined>,starts:ReadonlyMap<string,ReturnType<typeof value>>,boundaries:ReadonlyMap<Id,{before:Document;after:Document}>,registry:Registry):HistoryState {
|
|
const entries=new Map(snapshot.history.entries),affected=new Set(pending.map(intent=>intent.id));
|
|
for(const entry of snapshot.history.entries.values()){
|
|
const intents=pending.filter(intent=>entry.operationIds.includes(intent.id));if(!intents.length)continue;
|
|
if(!snapshot.history.undo.includes(entry.id)||entry.operationIds.some(id=>snapshot.operations.get(id)?.origin!=='user'))throw fault('recovery_history_traversal_not_supported',[entry.id]);
|
|
// 整个历史单元都未确认时，可直接由替代 changes 重建；没有剩余改动则移除该单元。
|
|
if(entry.operationIds.every(id=>affected.has(id))){
|
|
const changes=intents.flatMap(intent=>replacements.get(intent.id)?.changes??[]),operationIds=intents.flatMap(intent=>replacements.get(intent.id)?[replacements.get(intent.id)!.id]:[]);
|
|
if(changes.length)entries.set(entry.id,freeze({...entry,changes,operationIds}));else entries.delete(entry.id);
|
|
continue;
|
|
}
|
|
// 所有权和连线端点也属于结构，不能按普通字段独立拼接。
|
|
const structural=(changes:readonly Change[])=>changes.some(change=>change.kind!=='field'||['ownerCanvasId','source','target'].includes(change.field));
|
|
if(structural(entry.changes)||intents.some(intent=>structural(intent.changes))){
|
|
// A grouped entry is chronological. Never splice a confirmed operation from after a pending one into its prefix.
|
|
const first=entry.operationIds.findIndex(id=>affected.has(id));
|
|
if(entry.operationIds.slice(first).some(id=>!affected.has(id)))throw fault('recovery_history_nonprefix_ack',[entry.id]);
|
|
const oldPending=intents.flatMap(intent=>intent.changes),prefix=entry.changes.slice(0,entry.changes.length-oldPending.length);
|
|
if(entry.changes.length<oldPending.length||!equal(entry.changes.slice(prefix.length),oldPending))throw fault('recovery_history_mismatch',[entry.id]);
|
|
const boundary=boundaries.get(entry.id)!;
|
|
const changes=[...prefix,...intents.flatMap(intent=>replacements.get(intent.id)?.changes??[])];
|
|
// The acknowledged prefix is evidence, not a fresh command. Check its exact inverse rather than broadening it to remote additions.
|
|
try{planHistoryChanges(boundary.before,invert(prefix),registry);planHistoryChanges(boundary.after,invert(changes),registry);}
|
|
catch{throw fault('recovery_history_prefix_conflict',[entry.id]);}
|
|
const operationIds=entry.operationIds.flatMap(id=>affected.has(id)?replacements.get(id)?[replacements.get(id)!.id]:[]:[id]);
|
|
if(changes.length)entries.set(entry.id,freeze({...entry,changes,operationIds}));else entries.delete(entry.id);
|
|
continue;
|
|
}
|
|
// 纯字段按 ID/代次/field 分组，先验证该字段历史尾部确为 oldPending，再分离已确认 prefix。
|
|
// starts 记录本次审阅首次触及该字段的 fresh 值/存在性；只在它仍等于旧 pending 的 before 时保留确认前缀。
|
|
const keys=new Map<string,FieldChange>();for(const intent of intents)for(const change of intent.changes as readonly FieldChange[])keys.set(fieldKey(change),change);
|
|
const fields:FieldChange[]=[];
|
|
for(const [key,example]of keys){
|
|
if(entry.changes.some(change=>change.kind!=='field'&&change.id===example.target.id||change.kind==='field'&&change.target.id===example.target.id&&change.field==='ownerCanvasId'))throw fault('recovery_history_structure_not_supported',[entry.id]);
|
|
const original=entry.changes.filter((change):change is FieldChange=>change.kind==='field'&&fieldKey(change)===key);
|
|
const oldPending=intents.flatMap(intent=>intent.changes as readonly FieldChange[]).filter(change=>fieldKey(change)===key);
|
|
const prefix=original.slice(0,original.length-oldPending.length);
|
|
if(original.length<oldPending.length||!equal(original.slice(prefix.length),oldPending))throw fault('recovery_history_mismatch',[entry.id]);
|
|
// 若服务端基线已改变，原先确认前缀不再代表当前起点，不盲目把它纳入新的撤销内容。
|
|
const start=starts.get(JSON.stringify([entry.id,key]))!,first=oldPending[0];
|
|
if(!start.missing&&equal(start.before,first.before)&&(start.beforePresent??true)===(first.beforePresent??true))fields.push(...prefix);
|
|
fields.push(...intents.flatMap(intent=>replacements.get(intent.id)?.changes as readonly FieldChange[]??[]).filter(change=>fieldKey(change)===key));
|
|
}
|
|
// 未触及字段保留；重建字段链追加到后面。此分支已排除 owner/端点等结构变化，不用该方式重排结构依赖。
|
|
const changes:Change[]=[...entry.changes.filter(change=>change.kind!=='field'||!keys.has(fieldKey(change))),...fields];
|
|
const operationIds=entry.operationIds.flatMap(operationId=>affected.has(operationId)?replacements.get(operationId)?[replacements.get(operationId)!.id]:[]:[operationId]);
|
|
if(changes.length)entries.set(entry.id,freeze({...entry,changes,operationIds}));else entries.delete(entry.id);
|
|
}
|
|
// 只移除已消失的 undo 项；redo 的安全重规划不在此兼容分支中暗自执行。
|
|
return Object.freeze({undo:Object.freeze(snapshot.history.undo.filter(id=>entries.has(id))),redo:snapshot.history.redo,entries:readMap(entries)});
|
|
}
|