68 lines
8.6 KiB
JavaScript
68 lines
8.6 KiB
JavaScript
/**
|
|
* 从真实引擎持久化序列提取各个崩溃切点，再用匹配的服务端状态重新打开。
|
|
* 比较文档、history、ledger、请求次数和冻结字节；这是镜像恢复模拟，不会终止真实进程或测试真实磁盘掉电。
|
|
*/
|
|
import {test} from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import {createDocument,createRegistry,builtinComponents,applyChanges} from '../../src/lib/canvas-core/index.ts';
|
|
import {openProjectEngine} from '../../src/lib/canvas-runtime/engine.ts';
|
|
import {decodeRecovery} from '../../src/lib/canvas-runtime/recovery.ts';
|
|
import {createFoundryPersistence} from '../../src/lib/canvas-service/foundry.ts';
|
|
|
|
// 固定项目作用域，便于服务端版本向量与恢复镜像始终指向同一根 canvas。
|
|
const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000002',account='10000000-0000-4000-8000-000000000003';
|
|
// 原始文本目标只存在第一代；所有恢复场景从同一内容基线出发。
|
|
const registry=createRegistry(builtinComponents()),target={id:'a',incarnation:1},original=createDocument(root,[{id:root,parent:null}],[{...target,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',geometry:{x:0,y:0,width:100,height:80,rotation:0},data:{title:'Original',text:''}}]);
|
|
// 用可选本地镜像、服务端文档和历史回执初始化一套独立环境，分别模拟本地与远程已经完成到哪一步。
|
|
function fixture(image,initialServer=original,initialReceipt){
|
|
let owner,saved=image&&structuredClone(image),server=initialServer,version=initialReceipt?2:1,now=1000,next=0;
|
|
// 每次 commit 捕获一个持久化切点；calls 用于区分恢复后的查询与实际写入。
|
|
const cuts=[],calls=[],timers=new Map(),receipts=new Map(initialReceipt?[[initialReceipt.request_id,structuredClone(initialReceipt)]]:[]);
|
|
// 从当前服务端状态生成局部规范模型及其版本，作为成功回执的一部分。
|
|
const model=()=>({documentFormatVersion:1,projectId:root,complete:false,canvases:[...server.canvases.values()],elements:[...server.elements.values()],tombstones:Object.fromEntries(server.tombstones),versions:{[root]:version},removedCanvasIds:[]});
|
|
// 查询只查原请求回执；首次写入应用条件操作并递增版本，重复请求直接返回已有回执。
|
|
const transport=async input=>{
|
|
calls.push(input);
|
|
if(input.method==='GET'){const receipt=receipts.get(input.path.split('/').at(-1));return receipt?{status:200,body:receipt}:{status:404,body:{status:'unknown',request_id:input.path.split('/').at(-1)}};}
|
|
if(receipts.has(input.requestId))return{status:200,body:receipts.get(input.requestId)};
|
|
const body=JSON.parse(input.body);assert.equal(body.expected_versions[root],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()};receipts.set(input.requestId,receipt);return{status:200,body:receipt};
|
|
};
|
|
// 内存 owner 保证当前引擎独占；每次保存克隆镜像，防止后续状态污染选定的崩溃切点。
|
|
const storage={async claim(scope){assert.equal(owner,undefined);return owner={scope,active:true,fence:1,async release(){owner=undefined;this.active=false;}};},async recover(){return saved&&structuredClone(saved);},async commit(leader,batch){assert.strictEqual(leader,owner);saved=structuredClone(batch.image);cuts.push(saved);return{durableRootRevision:saved.rootRevision};},async readFile(){},close(){}};
|
|
// 接入真实 Foundry 文档适配器；假时钟只登记回调，由 drain 手动推进重试。
|
|
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1]]),storage,persistence:createFoundryPersistence({registry,workspaceId:workspace,rootId:root,transport}),clock:{now:()=>now,setTimer(callback,delay){const id=++next;timers.set(id,{callback,due:now+delay});return id;},clearTimer:id=>timers.delete(id)}};
|
|
// drain 有界推进收敛和到期重试；若八轮后仍有未解决操作即失败，不以无限等待掩盖恢复停滞。
|
|
return{cuts,calls,receipts,server:()=>server,open:()=>openProjectEngine(options),async drain(engine){for(let step=0;step<8;step++){await engine.settle();if([...engine.getSnapshot().runtime.ledger.operations.values()].every(operation=>operation.phase==='resolved'))return;now+=100000;for(const[id,timer]of[...timers])if(timer.due<=now){timers.delete(id);timer.callback();}}assert.fail('recovered operations did not settle');}};
|
|
}
|
|
|
|
// 先收集 pending、frozen、dispatch-marked、resolved 四种落盘状态，再用六个子场景分别核对恢复策略和已发生的外部副作用。
|
|
test('each durable crash cut restores its own document/history/ledger and external side-effect state',async t=>{
|
|
const producer=fixture(),engine=await producer.open();await engine.persist();const initial=structuredClone(producer.cuts.at(-1));
|
|
const edited=engine.editor.dispatch({type:'element.data',target,patch:{title:'Changed'}});assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'Changed');assert.ok(engine.getDurability().rootRevision>engine.getDurability().durableRootRevision);
|
|
await producer.drain(engine);const decoded=producer.cuts.map(image=>({image,state:decodeRecovery(image,registry)})),request=[...engine.getSnapshot().runtime.ledger.requests.values()][0],receipt=producer.receipts.get(request.id),changed=producer.server();
|
|
const pending=decoded.find(({state})=>state.runtime.ledger.operations.size===1&&state.runtime.ledger.requests.size===0),frozen=decoded.find(({state})=>state.runtime.ledger.requests.get(request.id)?.phase==='frozen'),dispatch=decoded.find(({state})=>state.runtime.ledger.requests.get(request.id)?.phase==='dispatch-marked'),resolved=decoded.findLast(({state})=>state.runtime.ledger.requests.get(request.id)?.phase==='resolved');
|
|
assert.ok(pending&&frozen&&dispatch&&resolved,'producer did not expose every required durable transition');await engine.close();
|
|
// 初始镜像没有尚未落盘的可见编辑；后续每个场景都明确指定应查询/写入的次数及服务端是否已执行。
|
|
for(const scenario of[
|
|
{name:'visible edit before durability is absent after crash and was never safe to leave',image:initial,edited:false,posts:0,queries:0},
|
|
{name:'durable pending intent dispatches once with its original logical identity',image:pending.image,edited:true,posts:1,queries:0},
|
|
// frozen 切点也保守查询；unknown 仅允许原请求重试，不能改用新身份或宣称服务未发生副作用。
|
|
{name:'durable frozen request is reconciled then sends its original bytes only once',image:frozen.image,edited:true,posts:1,queries:1,exact:true},
|
|
{name:'dispatch-marked before server commit queries unknown then retries the exact request',image:dispatch.image,edited:true,posts:1,queries:1,exact:true},
|
|
{name:'server commit before local receipt queries without repeating the write',image:dispatch.image,edited:true,posts:0,queries:1,server:changed,receipt,exact:true},
|
|
{name:'durable receipt and baseline do not replay the acknowledged edit or history',image:resolved.image,edited:true,posts:0,queries:0,server:changed,receipt},
|
|
])await t.test(scenario.name,async()=>{
|
|
// 本地 dispatch 切点可对应服务执行前或执行后；若服务已经提交，不能因读取旧镜像就把服务状态回退。
|
|
const f=fixture(scenario.image,scenario.server,scenario.receipt),recovered=await f.open();
|
|
try{
|
|
await f.drain(recovered);const state=recovered.getSnapshot();assert.equal(state.document.elements.get('a').data.title,scenario.edited?'Changed':'Original');assert.equal(f.server().elements.get('a').data.title,scenario.edited?'Changed':'Original');assert.equal(state.history.undo.length,scenario.edited?1:0);assert.equal(state.operations.size,scenario.edited?1:0);
|
|
assert.equal(f.calls.filter(call=>call.method==='POST').length,scenario.posts);assert.equal(f.calls.filter(call=>call.method==='GET').length,scenario.queries);
|
|
if(scenario.edited)assert.ok(state.runtime.ledger.operations.has(edited.operationIds[0]));
|
|
// 身份核对覆盖 GET 路径和 POST requestId；实际重发还必须保留同一冻结 body。
|
|
if(scenario.exact)for(const call of f.calls){assert.equal(call.requestId??call.path.split('/').at(-1),request.id);if(call.method==='POST')assert.equal(call.body,request.body);}
|
|
assert.equal(recovered.getDurability().rootRevision,recovered.getDurability().durableRootRevision);assert.equal(recovered.sessions.getSnapshot().views.size,0);
|
|
}finally{await recovered.close();}
|
|
});
|
|
});
|