docs(tests): explain ledger scheduling replay and local elision

This commit is contained in:
2026-11-12 09:41:49 +08:00
parent c426e94b58
commit 298f4b36c0
4 changed files with 76 additions and 0 deletions
+17
View File
@@ -1,3 +1,7 @@
/**
* 未发送操作的本地抵消证明测试:完整创建到删除的封闭生命周期可以不产生请求,但必须保存身份高水位。
* 抵消需证明操作连续、没有观察者遗漏且没有可能执行的网络证据;资源仍可由撤销历史持有。
*/
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createDocument, createRegistry, builtinComponents, applyChanges } from '../../src/lib/canvas-core/index.ts';
@@ -5,15 +9,20 @@ import { createLedger } from '../../src/lib/canvas-runtime/ledger.ts';
import { elisionBatchLimit, findElision, proveElision, validateElisions, withLocalLifecycles } from '../../src/lib/canvas-runtime/elision.ts';
import { replayPending } from '../../src/lib/canvas-runtime/reconcile.ts';
import { resourcePins } from '../../src/lib/canvas-runtime/resources.ts';
// 共享只读空基线与内置定义;各用例从该基线创建新工作文档,不修改基线本身。
const registry=createRegistry(builtinComponents()),base=createDocument('project',[{id:'canvas',parent:null}],[]);
// 创建指定 ID、归属与 incarnation 的文本样本,用于生命周期及原型敏感名称检查。
const node=(id='new',ownerCanvasId='canvas',incarnation=1)=>({id,ownerCanvasId,incarnation,componentId:'text',dataVersion:1,structure:'node',data:{title:id,text:''},geometry:{x:0,y:0,width:100,height:100,rotation:0}});
// 分别构造元素创建和删除变化;前后快照为抵消证明提供完整身份及内容依据。
const create=element=>({kind:'element',id:element.id,before:null,after:element}),remove=element=>({kind:'element',id:element.id,before:element,after:null});
// 将分块变化登记为线性依赖链,并记录每个操作首次看到的对象存在性;proof 每次读取最新账本。
function fixture(chunks,document=base){
const ledger=createLedger(),intents=chunks.map((changes,index)=>({id:`op-${index}`,operationType:'document.change',operationSchemaVersion:1,sequence:index+1,origin:'user',canvasIds:['canvas'],changes,dependencies:index?[`op-${index-1}`]:[]}));
for(const intent of intents){const presence=Object.create(null);for(const change of intent.changes)if(change.kind!=='canvas'){const id=change.kind==='field'?change.target.id:change.id;presence[id]??=change.kind==='element'&&!change.before?'absent':'exists';}ledger.add(intent,presence);}
return{ledger,intents,proof:()=>findElision(intents,document,ledger.getSnapshot())};
}
// 创建、改名、删除可整体抵消,不产生请求;仅在本地重放中保留 incarnation 墓碑,操作标记为 elided。
test('closed create/update/delete lifetime retires without a request and preserves only local high-water marks',()=>{
const original=node(),changed={...original,data:{...original.data,title:'changed'}},f=fixture([[create(original)],[{kind:'field',target:{id:original.id,incarnation:1},field:'data.title',before:'new',after:'changed'}],[remove(changed)]]),proof=f.proof();
assert.deepEqual(proof.operationIds,['op-0','op-1','op-2']);assert.deepEqual(proof.tombstones,{new:1});f.ledger.elide(proof);validateElisions(f.ledger.getSnapshot());
@@ -21,17 +30,20 @@ test('closed create/update/delete lifetime retires without a request and preserv
for(const record of f.ledger.getSnapshot().operations.values()){assert.equal(record.resolution,'elided');assert.equal(record.requestId,undefined);}
});
// 已有元素的后续修改不属于封闭生命周期:只抵消前两个操作,后续标题修改继续 pending 并可重放。
test('the longest balanced prefix can retire before a later existing-element edit',()=>{
const a=node('a'),document=createDocument('project',[{id:'canvas',parent:null}],[a]),n=node(),f=fixture([[create(n)],[remove(n)],[{kind:'field',target:{id:'a',incarnation:1},field:'data.title',before:'a',after:'edited'}]],document);
const proof=f.proof();assert.deepEqual(proof.operationIds,['op-0','op-1']);f.ledger.elide(proof);
const working=replayPending(document,f.ledger.getSnapshot(),registry);assert.equal(working.elements.get('a').data.title,'edited');assert.equal(working.tombstones.get('new'),1);assert.equal(f.ledger.getSnapshot().operations.get('op-2').phase,'pending');
});
// 分区入口、子 canvas 与成员一起创建再删除时可抵消;仍记录元素墓碑,不给服务端虚构 canvas 版本。
test('closed partitions retain entry/member lifecycles without inventing server canvas versions',()=>{
const frame={...node('frame'),componentId:'frame'},member=node('member','child'),canvas={id:'child',parent:{canvasId:'canvas',entryElementId:'frame'}},changes=[create(frame),{kind:'canvas',id:'child',before:null,after:canvas},create(member)],inverse=[remove(member),{kind:'canvas',id:'child',before:canvas,after:null},remove(frame)],f=fixture([changes,inverse]);
f.ledger.elide(f.proof());const working=replayPending(base,f.ledger.getSnapshot(),registry);assert.equal(working.canvases.size,1);assert.deepEqual(Object.fromEntries(working.tombstones),{member:1,frame:1});assert.equal(base.canvases.has('child'),false);
});
// 逐一注入未知基线、阻塞、冻结和已标记派发状态,任一条件都使本地抵消证明不可用。
test('unknown, blocked, frozen and possibly-applied creation cannot be elided',()=>{
const n=node();
for(const state of ['unknown','blocked','frozen','dispatch']){
@@ -42,6 +54,7 @@ test('unknown, blocked, frozen and possibly-applied creation cannot be elided',(
}
});
// 伪造高水位、跳过中间观察者或复用同一代身份都被拒绝;失败不能提前解决原操作。
test('a skipped observer, forged tombstone or reused incarnation cannot acquire a local completion proof',()=>{
const n=node(),f=fixture([[create(n)],[remove(n)]]),proof=f.proof();assert.throws(()=>f.ledger.elide({...proof,tombstones:{new:2}}),/invalid_elision/);assert.equal(f.ledger.getSnapshot().operations.get('op-0').phase,'pending');
const createIntent={...f.intents[0],sequence:1},deleteIntent={...f.intents[1],sequence:3},ledger=createLedger();ledger.add(createIntent,{new:'absent'});ledger.add({...createIntent,id:'observer',sequence:2,changes:[],dependencies:['op-0']},{});ledger.add(deleteIntent,{new:'exists'});
@@ -49,6 +62,7 @@ test('a skipped observer, forged tombstone or reused incarnation cannot acquire
assert.throws(()=>proveElision([{...createIntent,changes:[create(n),remove(n),create(n),remove(n)]}],{}),/stale_incarnation/);
});
// 篡改已保存证明的归属、墓碑和基线时恢复校验拒绝;未来证明版本单独报告缺少解释器。
test('persisted elision groups validate ownership, scope continuity, version and exact lifecycle evidence',()=>{
const n=node(),f=fixture([[create(n)],[remove(n)]]);f.ledger.elide(f.proof());const saved=f.ledger.getSnapshot();
for(const change of [record=>({...record,elisionId:'missing'}),record=>({...record,elision:{...record.elision,tombstones:{new:2}}}),record=>({...record,basePresence:{new:'unknown'}})]){
@@ -57,6 +71,7 @@ test('persisted elision groups validate ownership, scope continuity, version and
const operations=new Map(saved.operations),anchor=operations.get('op-0');operations.set('op-0',{...anchor,elision:{...anchor.elision,version:2}});assert.throws(()=>validateElisions({operations}),/elision_interpreter_unavailable/);
});
// 即使文档操作已抵消,undo 历史仍引用图片;资源 pin 必须同时保留创建和删除历史的来源。
test('resource references remain pinned by undo history after the document operations retire',()=>{
const picture={...node('picture'),componentId:'image',data:{title:'picture',text:'',src:'asset:asset',assetId:'asset'}},f=fixture([[create(picture)],[remove(picture)]]);f.ledger.elide(f.proof());
const entries=new Map([['create',{id:'create',changes:[create(picture)],operationIds:['op-0']}],['delete',{id:'delete',changes:[remove(picture)],operationIds:['op-1']}]]),pins=resourcePins({document:base,history:{undo:['create','delete'],redo:[],entries}},f.ledger.getSnapshot(),registry,new Map());
@@ -64,6 +79,7 @@ test('resource references remain pinned by undo history after the document opera
});
// 上一段本地生命周期留下的高水位仍有效,普通和原型敏感 ID 均不能以旧 incarnation 再次抵消,失败保持原快照。
test('a subsequent cancellation group cannot reuse an incarnation from a prior local lifetime',()=>{
for(const id of ['new','constructor','__proto__']){
const n=node(id),f=fixture([[create(n)],[remove(n)]]);f.ledger.elide(f.proof());
@@ -73,6 +89,7 @@ test('a subsequent cancellation group cannot reuse an incarnation from a prior l
}
});
// 真实的已有元素修改使前批在该处截断;后续封闭片段不能忽略外部依赖单独抵消,而独立封闭链可完整入批。
test('a preceding real edit is split from a later closed segment, including explicit preceding dependencies',()=>{
const a=node('a'),n=node(),document=createDocument('project',[{id:'canvas',parent:null}],[a]),f=fixture([[{kind:'field',target:{id:'a',incarnation:1},field:'data.title',before:'a',after:'edited'}],[create(n)],[remove(n)]],document);
assert.equal(elisionBatchLimit(f.intents,document,f.ledger.getSnapshot()),1);assert.equal(findElision(f.intents.slice(1),document,f.ledger.getSnapshot()),undefined);
+12
View File
@@ -1,11 +1,19 @@
/**
* 操作账本与调度基础结构的单元测试。账本保存依赖、冻结请求和执行证据;队列及堆只决定处理顺序。
* 这里直接操作内存账本,验证状态机与压缩规则,不代表网络或持久存储已经完成。
*/
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createDeque, createMinHeap } from '../../src/lib/canvas-runtime/structures.ts';
import { createLedger, compressChanges } from '../../src/lib/canvas-runtime/ledger.ts';
// 固定的文本元素快照;完整几何和 incarnation 让字段压缩与身份检查有明确基线。
const node = { id: 'a', incarnation: 1, ownerCanvasId: 'root', componentId: 'text', dataVersion: 1, data: { title: 'a', text: '' }, structure: 'node', geometry: { x: 0, y: 0, width: 20, height: 20, rotation: 0 } };
// 构造最小文档操作;依赖由各用例显式传入,避免默认依赖掩盖错误。
const intent = (id, changes = [], dependencies = []) => ({ id, operationType: 'document.change', operationSchemaVersion: 1, sequence: 1, origin: 'user', changes, dependencies });
// 把 create 操作冻结成固定请求;后续状态转换必须保留同一请求 ID 和请求体。
const frozen = ledger => ledger.freeze({ id: 'request', operationIds: ['create'], canvasIds: ['root'], expectedVersions: { root: 1 }, payload: { changes: [] } });
// 先验证大量 FIFO 出入队后的顺序,再验证最小堆以截止时间为主、序号为次序,空容器均返回 undefined。
test('FIFO retains order across compaction; retry heap orders deadline then sequence', () => {
const queue = createDeque(); for (let i = 0; i < 4000; i++) queue.push(i);
for (let i = 0; i < 4000; i++) assert.equal(queue.pop(), i);
@@ -16,6 +24,7 @@ test('FIFO retains order across compaction; retry heap orders deadline then sequ
assert.equal(heap.pop(), undefined);
});
// 仅针对完全未发送的修改:改名和移动折叠进创建对象,随后删除可抵消整个创建过程,移动不能覆盖标题。
test('wholly unsent create-data-delete disappears but a position update preserves the title', () => {
const create = { kind: 'element', id: 'a', before: null, after: node };
const rename = { kind: 'field', target: { id: 'a', incarnation: 1 }, field: 'data.title', before: 'a', after: 'renamed', beforePresent: true, afterPresent: true };
@@ -25,6 +34,7 @@ test('wholly unsent create-data-delete disappears but a position update preserve
assert.deepEqual(compressChanges([...compressed, { kind: 'element', id: 'a', before: compressed[0].after, after: null }]), []);
});
// 未知和已受理都保留“可能执行”证据并阻塞依赖删除;只有 applied 才放行,重派发仍使用原请求体。
test('frozen identity remains exact after unknown/accepted outcomes and deletion is a separate pending operation', () => {
const ledger = createLedger(); ledger.add(intent('create'), { a: 'absent' });
const request = frozen(ledger), body = request.body;
@@ -38,6 +48,7 @@ test('frozen identity remains exact after unknown/accepted outcomes and deletion
ledger.result('request', 'applied'); assert.equal(ledger.eligible('delete'), true);
});
// 拒绝缺失依赖及同 ID 不同意图;旧代次落盘完成不能把后来新增的操作一并标记为持久化。
test('dependencies are explicit, duplicate identities conflict and late durable completion does not mark newer work durable', () => {
const ledger = createLedger();
assert.throws(() => ledger.add(intent('child', [], ['missing']), {}), /dependency_missing/);
@@ -48,6 +59,7 @@ test('dependencies are explicit, duplicate identities conflict and late durable
assert.equal(ledger.eligible('second'), false);
});
// 恢复后先保留可能执行的风险;明确零写入结果将原操作阻塞,并拒绝再把同一终态改写为成功。
test('recovery retains possible application; explicit terminal zero-write does not pretend dependent writes succeeded', () => {
const original = createLedger(); original.add(intent('create'), {}); frozen(original); original.markDispatch('request');
const restored = createLedger(original.getSnapshot());
+14
View File
@@ -1,23 +1,35 @@
/**
* ACK 基线上的本地待处理操作重放测试。已被 ACK 包含的前缀应跳过,其余修改按顺序应用到私有草稿。
* 字段存在性、canvas 归属和 incarnation 都参与检查;后段冲突不能污染原 ACK 或操作账本。
*/
import {test} from 'node:test';
import assert from 'node:assert/strict';
import {createDocument,createRegistry,builtinComponents,applyChanges,replayChanges} from '../../src/lib/canvas-core/index.ts';
import {createLedger} from '../../src/lib/canvas-runtime/ledger.ts';
import {replayPending} from '../../src/lib/canvas-runtime/reconcile.ts';
// 使用内置类型校验重放结果,避免以空 registry 绕过元素约束。
const registry=createRegistry(builtinComponents());
// 构造带独立身份和归属的文本元素,可按场景替换 ID、canvas 或 incarnation。
const node=(id='a',ownerCanvasId='canvas',incarnation=1)=>({id,ownerCanvasId,incarnation,componentId:'text',dataVersion:1,structure:'node',data:{title:'original',text:''},geometry:{x:0,y:0,width:100,height:100,rotation:0}});
// 每次返回新的基线文档,避免不同用例复用被修改的实例。
const base=()=>createDocument('project',[{id:'canvas',parent:null}],[node()]);
// 生成带前后值的标题修改,字段目标固定到 a 的第一代身份。
const title=(before,after)=>({kind:'field',target:{id:'a',incarnation:1},field:'data.title',before,after});
// 把各组修改登记为有序 pending 操作,并根据创建或既有对象提供存在性基线;这里只搭建重放输入。
function ledgerFor(groups){const ledger=createLedger();groups.forEach((changes,index)=>{const presence={};for(const change of changes)if(change.kind!=='canvas')presence[change.kind==='field'?change.target.id:change.id]=change.kind==='element'&&change.before===null?'absent':'exists';ledger.add({id:`op-${index}`,sequence:index+1,origin:'user',operationType:'document.change',operationSchemaVersion:1,canvasIds:['canvas'],dependencies:[],changes},presence);});return ledger.getSnapshot();}
// 分别从原始 ACK 和已确认第一步的 ACK 重放同一序列,最终标题都到 third,且输入 ACK 保持原值。
test('pending replay preserves sequential field baselines and skips a prefix already present in ACK',()=>{
const original=base(),ledger=ledgerFor([[title('original','first'),title('first','second')],[title('second','third')]]);
for(const acknowledged of [original,applyChanges(original,[title('original','first')])]){const result=replayPending(acknowledged,ledger,registry);assert.equal(result.elements.get('a').data.title,'third');assert.equal(acknowledged.elements.get('a').data.title,acknowledged===original?'original':'first');}
});
// 显式区分字段缺失与值为 null;先创建 null 字段再删除,并允许已完成删除的重复重放。
test('replay distinguishes missing and null fields while applying and skipping repeated deletes',()=>{
const original=base(),insert={kind:'field',target:{id:'a',incarnation:1},field:'data.optional',before:null,after:null,beforePresent:false,afterPresent:true},remove={...insert,beforePresent:true,afterPresent:false};
const added=replayPending(original,ledgerFor([[insert]]),registry);assert.ok(Object.hasOwn(added.elements.get('a').data,'optional'));assert.equal(added.elements.get('a').data.optional,null);
const removed=replayPending(original,ledgerFor([[insert],[remove],[remove]]),registry);assert.equal(Object.hasOwn(removed.elements.get('a').data,'optional'),false);
});
// 先迁移 owner 再修改子 canvas 内字段;即使值未变化,错误作用域和旧 incarnation 仍必须被拒绝。
test('a draft replays owner transitions before later scope checks and rejects stale identity even for equal values',()=>{
const frame={...node('frame'),componentId:'frame'},canvases=[{id:'canvas',parent:null},{id:'child',parent:{canvasId:'canvas',entryElementId:'frame'}}],original=createDocument('project',canvases,[frame,node()]);
const move={kind:'field',target:{id:'a',incarnation:1},field:'ownerCanvasId',before:'canvas',after:'child'};
@@ -25,9 +37,11 @@ test('a draft replays owner transitions before later scope checks and rejects st
const relocated=createDocument('project',canvases,[frame,node('a','child')]);assert.throws(()=>replayPending(relocated,ledgerFor([[title('original','original')]]),registry),/operation_context_changed/);
const restored=createDocument('project',[canvases[0]],[node('a','canvas',2)]);assert.throws(()=>replayPending(restored,ledgerFor([[title('original','original')]]),registry),/stale_incarnation/);
});
// 前序重放成功、后序前值冲突时丢弃整个私有草稿;原文档与账本不变,修正前值后才得到完整结果。
test('late conflicts discard the entire private replay draft and preserve the ACK and ledger',()=>{
const original=base(),ledger=ledgerFor([[title('original','first')],[title('wrong','last')]]),before=JSON.stringify([...ledger.operations]);assert.throws(()=>replayPending(original,ledger,registry),/field_conflict/);assert.equal(original.elements.get('a').data.title,'original');assert.equal(JSON.stringify([...ledger.operations]),before);assert.equal(replayPending(original,ledgerFor([[title('original','first')],[title('first','last')]]),registry).elements.get('a').data.title,'last');
});
// 256 次创建后再删除并以新 incarnation 恢复一个对象,对比逐组 applyChanges 的元素和墓碑结果。
test('large replay preserves create/delete/restore lifecycles and conditional failures across operation boundaries',()=>{
const original=base(),created=Array.from({length:256},(_,i)=>node(`new-${i}`)),create=created.map(after=>({kind:'element',id:after.id,before:null,after})),deleted=created[50],restored={...deleted,incarnation:2,data:{...deleted.data,title:'restored'}},groups=[create,[{kind:'element',id:deleted.id,before:deleted,after:null}],[{kind:'element',id:deleted.id,before:null,after:restored}]];
const result=replayPending(original,ledgerFor(groups),registry);let reference=original;for(const changes of groups)reference=applyChanges(reference,changes);assert.deepEqual([...result.elements],[...reference.elements]);assert.deepEqual([...result.tombstones],[...reference.tombstones]);assert.equal(result.tombstones.get(deleted.id),1);assert.equal(original.elements.size,1);
+33
View File
@@ -1,15 +1,23 @@
/**
* 调度器状态机测试:从可变操作生成冻结请求,经持久化屏障后派发,未知结果走查询及原请求重试。
* 内存账本、可控 Promise 和假时钟隔离网络、磁盘与真实时间,验证 writer 占用、依赖及停止边界。
*/
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createLedger } from '../../src/lib/canvas-runtime/ledger.ts';
import { createScheduler } from '../../src/lib/canvas-runtime/scheduler.ts';
// 手动控制 Promise 完成时刻,让测试停在落盘、上传或服务端回执尚未完成的位置。
const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; };
// 推进有限轮微任务供调度泵运行;不推进墙上时间,也不替代 scheduler.settle 的收敛等待。
const tick = async () => { for (let i = 0; i < 20; i++) await Promise.resolve(); };
// 建立可替换端口的调度环境;mutate 使用账本副本发布状态,persist 仅记录快照,默认服务立即成功。
function setup(overrides = {}, initial) {
let ledger = createLedger(initial), time = 1000, next = 0, timerId = 0;
const timers = new Map(), calls = [], persisted = [], errors = [];
const scheduler = createScheduler({
getLedger: () => ledger.getSnapshot(),
// 模拟提交边界:先在私有账本运行 reducer,成功后才替换供 scheduler 读取的当前账本。
mutate(reduce) { const draft = createLedger(ledger.getSnapshot()); reduce(draft); ledger = draft; },
async persist() { persisted.push(ledger.getSnapshot()); },
createId: () => `request-${++next}`,
@@ -18,17 +26,21 @@ function setup(overrides = {}, initial) {
async submit(request) { calls.push(['submit', request]); return { outcome: 'applied' }; },
async query(request) { calls.push(['query', request]); return { outcome: 'applied' }; },
complete(request, status) { const draft = createLedger(ledger.getSnapshot()); draft.result(request.id, status.outcome, status.error); ledger = draft; },
// 假时钟仅登记定时器,advance 才触发到期回调;retryDelay 固定为 100 便于检查查询与重试次序。
clock: { now: () => time, setTimer(callback, delay) { const id = ++timerId; timers.set(id, { callback, at: time + delay }); return id; }, clearTimer: id => timers.delete(id) },
retryDelay: () => 100,
onError: error => errors.push(error), ...overrides,
});
return { scheduler, calls, persisted, errors, ledger: () => ledger,
// 添加文档操作并通知调度器;测试用 changes 的 ID 同时表示参与 canvas,省去真实文档编译。
add(id, canvases = ['canvas'], dependencies = []) { ledger.add({ id, operationType: 'document.change', operationSchemaVersion: 1, sequence: ledger.getSnapshot().operations.size + 1, origin: 'user', changes: canvases.map(id => ({ kind: 'canvas', id, before: null, after: { id, parent: null } })), dependencies }, {}); scheduler.notify([id]); },
// 先移动时间并执行当前已到期回调,再推进微任务;有意保持网络挂起时可跳过 settle。
async advance(ms, settle = true) { time += ms; for (const [id, timer] of [...timers]) if (timer.at <= time) { timers.delete(id); timer.callback(); } await tick(); if (settle) await scheduler.settle(); },
timerCount: () => timers.size,
};
}
// 用持久化屏障挡住网络;冻结请求已占 writer,但屏障解除后才允许发送,并在成功后唤醒依赖删除。
test('freeze and dispatch evidence must be durable before submit, then dependencies wake', async () => {
const barrier = deferred(), sent = [], state = setup({ persist: () => barrier.promise, submit: async request => { sent.push(request); return { outcome: 'applied' }; } });
state.add('create'); state.add('delete', ['canvas'], ['create']); state.scheduler.start(); await tick();
@@ -40,6 +52,7 @@ test('freeze and dispatch evidence must be durable before submit, then dependenc
assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop();
});
// 一个 canvas 的未知结果保持占用;其他 canvas 可继续,accepted 仍需查询,直到 applied 才放行同 canvas 后续操作。
test('unknown request holds its canvas; another canvas proceeds; accepted stays on query path', async () => {
const calls = []; let queries = 0;
const state = setup({
@@ -56,6 +69,7 @@ test('unknown request holds its canvas; another canvas proceeds; accepted stays
assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop();
});
// 恢复后先查回执;回执未知才重发,查询和重发都必须沿用保存的 ID 与字节,不能新建第二个请求。
test('receipt missing after recovery triggers exact retry, never a new identity/body', async () => {
const original = setup({ submit: async () => ({ outcome: 'unresolved' }) });
original.add('create'); original.scheduler.start(); await original.scheduler.settle();
@@ -68,6 +82,7 @@ test('receipt missing after recovery triggers exact retry, never a new identity/
assert.equal(restored.ledger().getSnapshot().requests.size, 1); restored.scheduler.stop();
});
// 落盘失败时不发送且不释放占用;显式恢复先重新持久化,再按已有派发证据保守查询。
test('storage failure never sends and keeps claims; explicit resume retries persistence first', async () => {
let broken = true; const sent = [];
const state = setup({ persist: async () => { if (broken) throw new Error('quota'); }, submit: async request => { sent.push(request); return { outcome: 'applied' }; } });
@@ -78,6 +93,7 @@ test('storage failure never sends and keeps claims; explicit resume retries pers
assert.equal(state.calls[0][0], 'query'); assert.equal(sent.length, 0); assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop();
});
// 跨两个 canvas 的操作同时持有两个 writer;停止后的迟到成功不能写回账本,也不能重新启动已关闭实例。
test('multi-canvas claims are atomic and stop fences late results', async () => {
const response = deferred(), state = setup({ submit: () => response.promise });
state.add('cross', ['a', 'b']); state.add('single', ['b']); state.scheduler.start(); await tick();
@@ -87,6 +103,7 @@ test('multi-canvas claims are atomic and stop fences late results', async () =>
assert.throws(() => state.scheduler.start(), /scheduler_closed/);
});
// 无法解释未来协议请求时保留原字节和 writer,并标记不可解释;无关 canvas 的新操作仍可完成。
test('missing frozen-request interpreter preserves bytes and claims while unrelated work proceeds', async () => {
const initial = setup({ submit: async () => ({ outcome: 'unresolved' }) }); initial.add('create'); initial.scheduler.start(); await initial.scheduler.settle(); initial.scheduler.stop();
const saved = initial.ledger().getSnapshot(), requests = new Map(saved.requests), old = requests.get('request-1');
@@ -99,6 +116,7 @@ test('missing frozen-request interpreter preserves bytes and claims while unrela
assert.equal(restored.scheduler.getClaims().get('canvas'), old.id); restored.scheduler.stop();
});
// 身份变化等阻塞结果没有证明零写入,故保留未知占用;同时停止自动重试,避免无权限的循环请求。
test('permission/protocol rejection keeps unknown claims without an automatic retry loop', async()=>{
const calls=[],state=setup({submit:async request=>{calls.push(request);return{outcome:'unresolved',blocked:true,error:'identity_changed'}}});
state.add('create');state.scheduler.start();await state.scheduler.settle();
@@ -106,12 +124,15 @@ test('permission/protocol rejection keeps unknown claims without an automatic re
await state.advance(10000);assert.equal(calls.length,1);state.scheduler.stop();
});
// 混合测试编译器:上传不占 canvas writer,文档操作按变化 ID 占用,二者仍共享网络调度约束。
const mixedCompile = (intent, requestId) => ({ kind: intent.operationType, protocolVersion: 2, canvasIds: intent.operationType === 'asset.upload' ? [] : intent.changes.map(change => change.id), expectedVersions: Object.fromEntries(intent.changes.map(change => [change.id, 1])), payload: { requestId, operationId: intent.id } });
// 插入独立资源上传并通知调度器,用于验证上传容量与文档任务的公平性。
function addAsset(state, id) {
state.ledger().add({ id, operationType: 'asset.upload', operationSchemaVersion: 1, sequence: state.ledger().getSnapshot().operations.size + 1, origin: 'user', changes: [], canvasIds: [], dependencies: [], resourceId: `resource-${id}` }, {});
state.scheduler.notify([id]);
}
// 80 个上传排队且首个挂起时,独立文档仍能进入网络;解除上传屏障后全部完成,无额外错误。
test('an upload burst yields without filling both network slots or starving independent edits', async () => {
const gate = deferred(), calls = [];
const state = setup({ compile: mixedCompile, submit: async request => { calls.push(request); if (request.operationIds[0] === 'asset-0') await gate.promise; return { outcome: 'applied' }; } });
@@ -124,6 +145,7 @@ test('an upload burst yields without filling both network slots or starving inde
assert.equal(calls.length, 81); assert.equal(state.errors.length, 0); state.scheduler.stop();
});
// 上传容量已满时不反复唤醒同一重试;更晚到期的文档查询仍能执行,容量释放后再恢复上传查询。
test('a capacity-blocked upload retry does not spin or hide a later document retry deadline', async () => {
const gate = deferred(), calls = [];
const state = setup({ compile: mixedCompile,
@@ -144,6 +166,7 @@ test('a capacity-blocked upload retry does not spin or hide a later document ret
assert.ok(calls.some(([action, id]) => action === 'query' && id === 'asset-1')); state.scheduler.stop();
});
// 嵌套 hold 不取消已进入网络的请求,但阻止新派发和到期重试;重复释放无效,最后一个 hold 释放才恢复。
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' }; } });
@@ -160,6 +183,7 @@ test('nested holds finish admitted receipts but suppress dependent dispatch and
state.scheduler.stop();
});
// 启动前 hold 保留已通知工作;stop 是终态,之后释放 hold 也不能复活调度器。
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);
@@ -167,11 +191,13 @@ test('holds acquired before start retain notified work; stop is terminal even wh
assert.equal(state.calls.length, 1); assert.throws(() => state.scheduler.hold(), /scheduler_closed/);
});
// 批量编译仅打包逻辑操作并汇总排序后的 canvas;这里不承担真实 wire 字段压缩。
const compileBatch = (intents, requestId) => {
const canvasIds = [...new Set(intents.flatMap(intent => intent.changes.map(change => change.id)))].sort();
return { canvasIds, expectedVersions: Object.fromEntries(canvasIds.map(id => [id, 1])), payload: { requestId, operations: intents.map(intent => ({ id: intent.id, changes: intent.changes })) } };
};
// 同一微任务窗口内的依赖链可打成一批;整批共享持久化屏障,成功后所有逻辑操作一并解决。
test('mutable microtask window batches dependent document IDs behind one durable dispatch barrier', async () => {
const barrier = deferred(), sent = [];
const state = setup({ compileBatch, persist: () => barrier.promise, submit: async request => { sent.push(request); return { outcome: 'applied' }; } });
@@ -187,6 +213,7 @@ test('mutable microtask window batches dependent document IDs behind one durable
state.scheduler.stop();
});
// 先验证不同 canvas 通道不会任意并批,再加入观察前序操作的 run 记录,验证它形成批次边界。
test('batching retains independent lanes and does not cross an external dependency or task observer', async () => {
const pending = deferred(), sent = [], state = setup({ compileBatch, submit: async request => { sent.push(request); return pending.promise; } });
state.add('a1', ['a']); state.add('b1', ['b']); state.add('a2', ['a'], ['a1']);
@@ -203,6 +230,7 @@ test('batching retains independent lanes and does not cross an external dependen
assert.deepEqual(observed.calls.map(([, request]) => request.operationIds), [['before'], ['after']]); observed.scheduler.stop();
});
// 未知批次冻结全部逻辑 ID 和原始请求体;后来编辑保持 pending,查询及重发不能吸收它。
test('unknown batch preserves all IDs, claims, and exact bytes across query and retry; later edits stay mutable', async () => {
const sent = [], state = setup({ compileBatch, submit: async request => { sent.push(request); return { outcome: 'unresolved' }; }, query: async () => ({ outcome: 'unresolved' }) });
state.add('one'); state.add('two', ['canvas'], ['one']); state.scheduler.start(); await state.scheduler.settle();
@@ -213,6 +241,7 @@ test('unknown batch preserves all IDs, claims, and exact bytes across query and
assert.equal(state.scheduler.getClaims().get('canvas'), original.id); state.scheduler.stop();
});
// hold 阻止批次入场;释放后 130 个依赖操作按 64、64、2 排空,不依赖推进假定时器。
test('bounded batches drain through settle without waiting for fake timers and holds suppress scheduled admission', async () => {
const state = setup({ compileBatch }); state.scheduler.start(); const release = state.scheduler.hold();
for (let i = 0; i < 130; i++) state.add(`op-${i}`, ['canvas'], i ? [`op-${i - 1}`] : []);
@@ -222,17 +251,20 @@ test('bounded batches drain through settle without waiting for fake timers and h
assert.equal(state.ledger().getSnapshot().operations.get('op-129').phase, 'resolved'); state.scheduler.stop();
});
// 状态发布因存储 leader 丢失而失败时,错误通过 onError 报告;操作保持 pending,不能生成或发送请求。
test('a fenced pump reports failed state publication without throwing or sending through the old identity',async()=>{
const state=setup({mutate(){throw Error('storage_leader_lost');}});state.scheduler.start();state.add('fenced');await tick();await state.scheduler.settle();assert.equal(state.calls.length,0);assert.equal(state.ledger().getSnapshot().requests.size,0);assert.equal(state.ledger().getSnapshot().operations.get('fenced').phase,'pending');assert.ok(state.errors.some(error=>error.message==='storage_leader_lost'));state.scheduler.stop();
});
// Wall time can change independently of the elapsed time used by browser timers.
// 分离墙上时间与定时器经过时间:可单独跳动 wall,或推进 elapsed 并返回旧回调来模拟迟到触发。
function adjustableClock(initial=1000){
let wall=initial,elapsed=0,next=0;const timers=new Map();
return{clock:{now:()=>wall,setTimer(callback,delay){const id=++next;timers.set(id,{callback,delay,at:elapsed+delay});return id;},clearTimer:id=>timers.delete(id)},timers,
setWall(value){wall=value;},fireAfter(ms){elapsed+=ms;const due=[...timers].filter(([,timer])=>timer.at<=elapsed);for(const[id,timer]of due){timers.delete(id);timer.callback();}return due.map(([,timer])=>timer.callback);}};
}
// 墙上时钟回退时重新计算剩余等待,前跳时只查同一冻结请求;旧回调重复触发不能重复查询或越过依赖。
test('wall-clock jumps recheck retry deadlines without releasing unknown claims or bypassing dependencies',async()=>{
const clock=adjustableClock(),receipt=deferred(),calls=[];
const state=setup({clock:clock.clock,
@@ -255,6 +287,7 @@ test('wall-clock jumps recheck retry deadlines without releasing unknown claims
for(const callback of staleCallbacks)callback();await state.scheduler.settle();assert.equal(calls.length,4);assert.equal(clock.timers.size,0);assert.equal(state.scheduler.getClaims().size,0);state.scheduler.stop();
});
// 分别从回退及前跳的墙上时间恢复,验证最早有效截止时间先查询;恢复过程仍保持原请求体和逻辑 ID。
test('recovered retries recompute remaining wall time and wake the earliest valid request after either clock shift',async()=>{
const original=setup({async submit(request){return{outcome:'unresolved',retryAfterMs:request.operationIds[0]==='long'?500:100};}});
original.add('long',['a']);original.add('short',['b']);original.scheduler.start();await original.scheduler.settle();const saved=original.ledger().getSnapshot();original.scheduler.stop();