feat(canvas): batch mutable document operations with durable coverage

This commit is contained in:
2026-11-08 04:29:56 +08:00
parent 90650873e6
commit 95a5cd5697
8 changed files with 182 additions and 21 deletions
+2
View File
@@ -26,6 +26,8 @@ export type PersistencePort = {
verifyApplied?: (request:FrozenRequest,receipt:Json) => AppliedModelProof;
supportsRequest?: (request: FrozenRequest) => boolean;
compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest;
/** Compile an ordered mutable batch; logical IDs and original intents remain independent. */
compileBatch?: (intents: readonly OperationIntent[], requestId: Id, snapshot: EngineSnapshot) => CompiledRequest;
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
/** Pure codec, merging only participants into the latest base, never an old working snapshot. */
+2 -1
View File
@@ -79,7 +79,7 @@ export async function openProjectEngine(options: EngineOptions) {
if (resource && !resource.stored) dependencies.add(resource.operationId);
}
const basePresence: Record<Id, 'absent' | 'exists'> = Object.create(null);
for (const change of intent.changes) if (change.kind === 'element') basePresence[change.id] = change.before ? 'exists' : 'absent'; else if (change.kind === 'field') basePresence[change.target.id] ??= 'exists';
for (const change of intent.changes) if (change.kind === 'element') basePresence[change.id] ??= change.before ? 'exists' : 'absent'; else if (change.kind === 'field') basePresence[change.target.id] ??= 'exists';
ledger.add(freeze({ ...intent, dependencies: [...dependencies] }), basePresence);
}
return Object.freeze({ ...previous, resources: readMap(resources), ledger: ledger.getSnapshot() });
@@ -199,6 +199,7 @@ export async function openProjectEngine(options: EngineOptions) {
if (!options.assets || !resource?.digest) throw fault('asset_not_ready');
return { kind: 'asset.upload', protocolVersion: 2, canvasIds: [], expectedVersions: {}, payload: { protocol_version: 2, request_id: id, resource_id: resource.id, operation_id: resource.operationId, sha256: resource.digest, byte_length: resource.byteLength, name: resource.name, mime_type: resource.mimeType } };
},
compileBatch: options.persistence.compileBatch ? (intents, id) => options.persistence.compileBatch!(intents, id, editor.getSnapshot()) : undefined,
async submit(request, signal) {
if (['run.start','run.cancel','run.apply'].includes(request.kind??'')) return options.runs ? network.run(signal, () => options.runs!.submit(request,signal)) : { outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' };
if (request.kind !== 'asset.upload') return network.run(signal, () => options.persistence.submit(request, signal));
+43 -6
View File
@@ -15,6 +15,7 @@ export type SchedulerOptions = {
createId(): Id;
participants(intent: OperationIntent): readonly Id[];
compile(intent: OperationIntent, requestId: Id): CompiledRequest;
compileBatch?: (intents: readonly OperationIntent[], requestId: Id) => CompiledRequest;
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
supportsRequest?: (request: FrozenRequest) => boolean;
@@ -35,6 +36,8 @@ export function createScheduler(options: SchedulerOptions) {
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<Id, number>(), active = new Map<Id, { controller: AbortController; promise: Promise<void>; kind: FrozenRequest['kind'] }>();
const holds = new Set<symbol>();
const ordered: { id: Id; sequence: number }[] = [];
let scheduledPump: Promise<void> | undefined;
const laneWaiters = new Map<NonNullable<FrozenRequest['kind']>, Set<Id>>(), deferredRetries = new Map<Id, { id: Id; dueAt: number; sequence: number; token: number }>();
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)));
@@ -53,12 +56,38 @@ export function createScheduler(options: SchedulerOptions) {
if (!record) return;
if (!indexed.has(id)) {
indexed.add(id);
let lo = 0, hi = ordered.length;
while (lo < hi) { const mid = (lo + hi) >>> 1; if (ordered[mid].sequence < record.intent.sequence) lo = mid + 1; else hi = mid; }
ordered.splice(lo, 0, { id, sequence: record.intent.sequence });
for (const dependency of record.intent.dependencies) {
const set = dependents.get(dependency) ?? new Set<Id>(); set.add(id); dependents.set(dependency, set);
}
}
enqueue(id);
}
/** Only a contiguous mutable document prefix may share a request. Task barriers stay visible. */
function batch(first: OperationIntent, canvases: Set<Id>): readonly OperationIntent[] {
const selected = [first], ids = new Set([first.id]);
if (!options.compileBatch || first.operationType !== 'document.change' || first.operationSchemaVersion !== 1) return selected;
let lo = 0, hi = ordered.length, changeCount = first.changes.length;
while (lo < hi) { const mid = (lo + hi) >>> 1; if (ordered[mid].sequence <= first.sequence) lo = mid + 1; else hi = mid; }
const ledger = options.getLedger();
for (let i = lo; i < ordered.length && selected.length < 64; i++) {
const record = ledger.operations.get(ordered[i].id);
if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.operationType !== 'document.change' || record.intent.operationSchemaVersion !== 1) break;
if (record.intent.dependencies.some(id => !ids.has(id) && ledger.operations.get(id)?.phase !== 'resolved')) break;
const scope = options.participants(record.intent);
// Independent canvas lanes keep independent failure and retry boundaries.
if (!scope.length || !scope.some(id => canvases.has(id)) || scope.some(id => claims.has(id)) || changeCount + record.intent.changes.length > 10_000) break;
selected.push(record.intent); ids.add(record.intent.id); changeCount += record.intent.changes.length;
for (const id of scope) canvases.add(id);
}
return selected;
}
function requestPump() {
if (!options.compileBatch) { pump(); return; }
if (!scheduledPump) scheduledPump = Promise.resolve().then(() => { scheduledPump = undefined; pump(); });
}
function own(request: FrozenRequest) {
for (const canvas of request.canvasIds) {
const prior = claims.get(canvas);
@@ -144,7 +173,7 @@ export function createScheduler(options: SchedulerOptions) {
const id = ready.pop(); if (!id) break;
queued.delete(id);
const record = options.getLedger().operations.get(id);
if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue;
if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue;
try {
if (!canRun(record.intent.operationType)) {
const set = laneWaiters.get(record.intent.operationType) ?? new Set<Id>(); set.add(id); laneWaiters.set(record.intent.operationType, set); continue;
@@ -156,9 +185,11 @@ export function createScheduler(options: SchedulerOptions) {
for (const canvas of occupied) { const set = waiters.get(canvas) ?? new Set<Id>(); set.add(id); waiters.set(canvas, set); }
continue;
}
const requestId = options.createId(), compiled = options.compile(record.intent, requestId);
if (canvases.length !== compiled.canvasIds.length || canvases.some(canvas => !compiled.canvasIds.includes(canvas))) throw fault('participant_mismatch');
options.mutate(ledger => ledger.freeze({ id: requestId, operationIds: [id], ...compiled }));
const scope = new Set(canvases), intents = batch(record.intent, scope);
const participants = [...scope].sort();
const requestId = options.createId(), compiled = intents.length > 1 ? options.compileBatch!(intents, requestId) : options.compile(record.intent, requestId);
if (participants.length !== compiled.canvasIds.length || participants.some(canvas => !compiled.canvasIds.includes(canvas))) throw fault('participant_mismatch');
options.mutate(ledger => ledger.freeze({ ...compiled, id: requestId, operationIds: intents.map(intent => intent.id) }));
own(options.getLedger().requests.get(requestId)!); launch(requestId, 'submit');
} catch (error) { options.mutate(ledger => ledger.block(id, [(error as Error).message || 'compile_failed'])); report(error); }
}
@@ -182,7 +213,7 @@ export function createScheduler(options: SchedulerOptions) {
}
pump();
},
notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); pump(); },
notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); requestPump(); },
/** Acquire synchronously, then settle(): already admitted work completes with its original identity. */
hold() {
if (closed) throw fault('scheduler_closed');
@@ -210,7 +241,13 @@ export function createScheduler(options: SchedulerOptions) {
pump();
},
getClaims: () => readMap(claims),
async settle() { while (active.size) await Promise.all([...active.values()].map(item => item.promise)); },
async settle() {
do {
if (scheduledPump) await scheduledPump;
if (active.size) await Promise.all([...active.values()].map(item => item.promise));
if (!stopped && !holds.size && pumpTimer !== undefined) { clock.clearTimer(pumpTimer); pumpTimer = undefined; pump(); }
} while (scheduledPump || active.size || (!stopped && !holds.size && pumpTimer !== undefined));
},
stop() { closed = true; stopped = true; if (timer !== undefined) clock.clearTimer(timer); if (pumpTimer !== undefined) clock.clearTimer(pumpTimer); timer = pumpTimer = undefined; for (const item of active.values()) item.controller.abort(); },
});
}
+17 -12
View File
@@ -112,6 +112,21 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
return { outcome: 'unresolved', error: code ?? 'transport_unknown', ...(code && code !== 'invalid_canonical_response' ? { blocked: true } : {}) };
}
}
function compileBatch(intents: readonly OperationIntent[], requestId: Id, snapshot: EngineSnapshot) {
if (!intents.length || intents.length > 64 || snapshot.document.projectId !== options.rootId || !uuid.test(requestId) || new Set(intents.map(intent => intent.id)).size !== intents.length) throw fault('operation_interpreter_unavailable');
for (const intent of intents) if (!uuid.test(intent.id) || intent.operationType !== 'document.change' || intent.operationSchemaVersion !== 1) throw fault('operation_interpreter_unavailable');
if (intents.some((intent, index) => index > 0 && intent.sequence <= intents[index - 1].sequence)) throw fault('invalid_operation_order');
const canvasIds = [...new Set(intents.flatMap(intent => intent.canvasIds))].sort(), expectedVersions: Record<Id, number> = {};
for (const canvas of canvasIds) {
const version = snapshot.runtime.versions.get(canvas);
if (version === undefined && !intents.some(intent => intent.changes.some(change => change.kind === 'canvas' && change.id === canvas && change.before === null && change.after))) throw fault('missing_server_version', [canvas]);
expectedVersions[canvas] = version ?? 0;
}
// The body is the durable coverage mapping. Never replace logical IDs with a synthetic batch ID.
const payload = JSON.parse(JSON.stringify({ protocol_version: 2, request_id: requestId, expected_versions: expectedVersions, operations: intents.map(intent => ({ id: intent.id, changes: intent.changes })) })) as Json;
if (!validJson(payload)) throw fault('invalid_operation_payload');
return freeze({ protocolVersion: 2, canvasIds, expectedVersions, payload });
}
return Object.freeze({
supportsRequest: (request: FrozenRequest) => request.protocolVersion === 2,
verifyApplied(request:FrozenRequest,receipt:Json){
@@ -127,18 +142,8 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
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<Id, number> = {};
for (const canvas of canvasIds) {
const version = snapshot.runtime.versions.get(canvas);
if (version === undefined && !intent.changes.some(change => change.kind === 'canvas' && change.id === canvas && change.before === null && change.after)) throw fault('missing_server_version', [canvas]);
expectedVersions[canvas] = version ?? 0;
}
const payload = JSON.parse(JSON.stringify({ protocol_version: 2, request_id: requestId, expected_versions: expectedVersions, operations: [{ id: intent.id, changes: intent.changes }] })) as Json;
if (!validJson(payload)) throw fault('invalid_operation_payload');
return freeze({ protocolVersion: 2, canvasIds, expectedVersions, payload });
},
compile: (intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot) => compileBatch([intent], requestId, snapshot),
compileBatch,
submit: (request: FrozenRequest, signal: AbortSignal) => execute(request, false, signal),
query: (request: FrozenRequest, signal: AbortSignal) => execute(request, true, signal),
canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot) {
+33 -2
View File
@@ -19,7 +19,7 @@ async function adoptionConsumerCase(name,assert){
else{engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'one'}},{groupId:'chain'});await engine.settle();engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'two'}},{groupId:'chain'});}
await engine.settle();let session=await engine.prepareRecovery(),first=session.plan(selected());await session.resolve(first);await engine.settle();await reopen();document=applyChanges(document,[{kind:'field',target:{id:'a',incarnation:1},field:'data.title',before:'a',after:'remote 1',beforePresent:true,afterPresent:true}]);versions.set(root,versions.get(root)+1);
session=await engine.prepareRecovery();const decisions=selected();decisions.set(first.operationIds[structural?1:0],'discard-local');if(structural)reject=false;const second=session.plan(decisions);await session.resolve(second);await engine.settle();await reopen();
if(!structural){engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{text:'after adoption'}},{groupId:'chain'});await engine.settle();document=applyChanges(document,[{kind:'field',target:{id:'a',incarnation:1},field:'data.title',before:'remote 1',after:'remote 2',beforePresent:true,afterPresent:true}]);versions.set(root,versions.get(root)+1);session=await engine.prepareRecovery();const third=session.plan(selected());reject=false;await session.resolve(third);await engine.settle();await reopen();assert(engine.getSnapshot().runtime.recoveries.size===3,'nested adoption audit lineage was lost');assert(document.elements.get('a').data.title==='two'&&document.elements.get('a').data.text==='after adoption','post-adoption native edit was lost');assert(engine.editor.history.undo().status==='applied','nested field recovery undo failed');await engine.settle();assert(document.elements.get('a').data.title==='remote 2'&&document.elements.get('a').data.text==='','undo did not restore newest reviewed server baseline');assert(engine.editor.history.redo().status==='applied','nested field recovery redo failed');await engine.settle();assert(calls.filter(call=>call.method==='POST').length===7,'nested field recovery repeated a request');}
if(!structural){engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{text:'after adoption'}},{groupId:'chain'});await engine.settle();document=applyChanges(document,[{kind:'field',target:{id:'a',incarnation:1},field:'data.title',before:'remote 1',after:'remote 2',beforePresent:true,afterPresent:true}]);versions.set(root,versions.get(root)+1);session=await engine.prepareRecovery();const third=session.plan(selected());reject=false;await session.resolve(third);await engine.settle();await reopen();assert(engine.getSnapshot().runtime.recoveries.size===3,'nested adoption audit lineage was lost');assert(document.elements.get('a').data.title==='two'&&document.elements.get('a').data.text==='after adoption','post-adoption native edit was lost');assert(engine.editor.history.undo().status==='applied','nested field recovery undo failed');await engine.settle();assert(document.elements.get('a').data.title==='remote 2'&&document.elements.get('a').data.text==='','undo did not restore newest reviewed server baseline');assert(engine.editor.history.redo().status==='applied','nested field recovery redo failed');await engine.settle();assert(calls.filter(call=>call.method==='POST').length===6,'nested field recovery did not batch its replacements');}
else{assert(engine.getSnapshot().runtime.recoveries.size===2&&document.elements.get('member').data.title==='member','structural choice did not survive adoption');assert(engine.editor.history.undo().status==='applied','adopted structure undo failed');await engine.settle();assert(!document.canvases.has(child),'adopted structure undo left a child canvas');assert(engine.editor.history.redo().status==='applied','adopted structure redo failed');await engine.settle();assert(document.elements.get('member').incarnation===2&&document.elements.get('a').data.title==='remote 1','structural redo reused a life or erased remote data');assert(calls.filter(call=>call.method==='POST').length===5,'structural adoption repeated a request');}
await reopen();for(const[id,body]of saved)assert(engine.getSnapshot().runtime.ledger.requests.get(id).body===body,'original frozen request bytes changed after final reopen');
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
@@ -33,7 +33,7 @@ async function adoptionEvidenceCase(name,assert){
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.geometry',target:{id:'a',incarnation:1},patch:{width:250}},{groupId:'adoption'});await engine.settle();const old=[...engine.getSnapshot().runtime.ledger.requests.values()][0],discard=engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'discard me'}},{groupId:'adoption'});engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'kept'}},{groupId:'adoption'});await engine.settle();const session=await engine.prepareRecovery(),selected=new Map(session.pendingOperationIds.map(id=>[id,id===discard.operationIds[0]?'discard-local':'keep-local'])),plan=session.plan(selected);reject=false;await session.resolve(plan);await engine.settle();await engine.close();engine=await openProjectEngine(options);
const snapshot=engine.getSnapshot(),audit=snapshot.runtime.recoveries.get(plan.id),backup=audit.backup.history,before={undo:backup.undo,redo:backup.redo,entries:new Map(backup.entries)},reconstructed=replayHistoryEvidence(before,audit.historyReplay,id=>snapshot.operations.has(id));assert(sameHistory(reconstructed,snapshot.history),'persisted adoption witness did not compose to its history');assert(audit.historyReplay.events.filter(event=>event.kind==='operation').length===2,'discarded contribution became a replay operation');assert(snapshot.runtime.ledger.requests.get(old.id).body===old.body,'adoption evidence changed old request bytes');assert(document.elements.get('a').geometry.width===250&&document.elements.get('a').data.title==='kept','adoption content or geometry changed');assert(engine.editor.history.undo().status==='applied','geometry evidence broke undo');await engine.settle();assert(document.elements.get('a').geometry.width===200&&document.elements.get('a').data.title==='a','adopted group inverse is incorrect');assert(engine.editor.history.redo().status==='applied','geometry evidence broke redo');await engine.settle();assert(calls.filter(call=>call.method==='POST').length===5,'adoption evidence generated an extra content write');
const snapshot=engine.getSnapshot(),audit=snapshot.runtime.recoveries.get(plan.id),backup=audit.backup.history,before={undo:backup.undo,redo:backup.redo,entries:new Map(backup.entries)},reconstructed=replayHistoryEvidence(before,audit.historyReplay,id=>snapshot.operations.has(id));assert(sameHistory(reconstructed,snapshot.history),'persisted adoption witness did not compose to its history');assert(audit.historyReplay.events.filter(event=>event.kind==='operation').length===2,'discarded contribution became a replay operation');assert(snapshot.runtime.ledger.requests.get(old.id).body===old.body,'adoption evidence changed old request bytes');assert(document.elements.get('a').geometry.width===250&&document.elements.get('a').data.title==='kept','adoption content or geometry changed');const replacementRequests=calls.filter(call=>call.method==='POST'&&plan.operationIds.every(id=>JSON.parse(call.body).operations.some(operation=>operation.id===id)));assert(replacementRequests.length===1,'adopted logical IDs did not share one frozen request');assert(engine.editor.history.undo().status==='applied','geometry evidence broke undo');await engine.settle();assert(document.elements.get('a').geometry.width===200&&document.elements.get('a').data.title==='a','adopted group inverse is incorrect');assert(engine.editor.history.redo().status==='applied','geometry evidence broke redo');await engine.settle();assert(calls.filter(call=>call.method==='POST').length===4,'adoption evidence did not use one replacement batch');
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
}
@@ -293,6 +293,35 @@ async function resourceRecoveryCase(name, assert) {
} finally { await engine?.close(); storage.close(); indexedDB.deleteDatabase(`${name}-resources`); }
}
async function batchRecoveryCase(name,assert){
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()),node={id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',geometry:{x:0,y:0,width:200,height:100,rotation:0},data:{title:'a',text:''}},document=createDocument(root,[{id:root,parent:null}],[node]);
const database=`${name}-batch-recovery`,storage=createBrowserStorage({name:database}),calls=[],receipts=new Map();let server=document,version=1,lose=true,engine;
const transport=async request=>{
calls.push(request);
if(request.method==='GET')return{status:200,body:receipts.get(request.path.split('/').at(-1))};
const body=JSON.parse(request.body);assert(body.expected_versions[root]===version,'batch lost the server version fence');
for(const operation of body.operations)server=applyChanges(server,operation.changes);version++;
const receipt={protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(operation=>operation.id),model:{documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],canvases:[...server.canvases.values()],elements:[...server.elements.values()],tombstones:Object.fromEntries(server.tombstones),versions:{[root]:version}}};receipts.set(request.requestId,receipt);
if(lose){lose=false;return{status:502,body:{error:'lost_batch_ack'}};}return{status:200,body:receipt};
};
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document,versions:new Map([[root,1]]),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})};
try{
engine=await openProjectEngine(options);
const first=engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'first'}}),second=engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'second'}});
assert(engine.getSnapshot().document.elements.get('a').data.title==='second','local publication waited for batch preparation');await engine.settle();
const frozen=[...engine.getSnapshot().runtime.ledger.requests.values()][0];
assert(calls.length===1&&frozen.operationIds.length===2&&frozen.operationIds[0]===first.operationIds[0]&&frozen.operationIds[1]===second.operationIds[0],'logical edits were not covered by one ordered request');assert(version===2,'batch advanced a canvas version per operation');
await engine.close();engine=await openProjectEngine({...options,clock:{now:()=>Date.now()+100000,setTimer:(fn,ms)=>setTimeout(fn,ms),clearTimer:id=>clearTimeout(id)}});await new Promise(resolve=>setTimeout(resolve,10));await engine.settle();
assert(calls.length===2&&calls[1].method==='GET'&&calls[1].path.endsWith(frozen.id),'batch recovery sent a new request');
assert(engine.getSnapshot().runtime.ledger.requests.get(frozen.id).body===frozen.body&&engine.getSnapshot().history.undo.length===2,'batch changed frozen bytes or collapsed history units');
for(const id of frozen.operationIds)assert(engine.getSnapshot().runtime.ledger.operations.get(id).phase==='resolved','batch ACK omitted a logical operation');
assert(engine.editor.history.undo().status==='applied','second edit undo failed');await engine.settle();assert(server.elements.get('a').data.title==='first','batch undo lost the intermediate historical value');
assert(engine.editor.history.undo().status==='applied','first edit undo failed');await engine.settle();assert(server.elements.get('a').data.title==='a','batch first inverse failed');
assert(engine.editor.history.redo().status==='applied','first edit redo failed');await engine.settle();assert(server.elements.get('a').data.title==='first','batch redo skipped an independent history unit');
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
}
async function protocolRecoveryCase(name,assert){
const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000003',account='10000000-0000-4000-8000-000000000004';
const registry=createRegistry(builtinComponents()),document=createDocument(root,[{id:root,parent:null}],[{id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',geometry:{x:0,y:0,width:260,height:200,rotation:0},data:{title:'a',text:''}}]);
@@ -415,6 +444,8 @@ export async function run() {
results.push('project engine reopens real IndexedDB with one history unit and the original request; ACK resolves without duplicate edit');
await resourceRecoveryCase(name, assert);
results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity');
await batchRecoveryCase(name,assert);
results.push('ordered mutable batches retain every logical ID and independent undo unit through a lost ACK, real IndexedDB reopening and exact receipt query');
await protocolRecoveryCase(name,assert);
results.push('protocol 2 service adapter recovers exact frozen bytes and reconciles a lost ACK through real IndexedDB');
await runRecoveryCase(name,assert);
+16
View File
@@ -204,3 +204,19 @@ test('complete recovery reads reject same-version rewrites and lifecycle regress
assert.throws(()=>validateCanonicalRead(original,new Map([['canvas',1]]),withTombstone,new Map([['canvas',1]]),registry),/canonical_version_mismatch/);
await engine.close();
});
test('one transaction captures the first presence baseline even when its created element is later deleted', async () => {
const storage = storageFixture(), port = persistenceFixture(), { engine } = await open(storage, port);
try {
const node = { ...original.elements.get('a'), id: 'temporary' };
const result = engine.editor.transaction(tx => {
tx.dispatch({ type: 'element.create', element: node });
tx.dispatch({ type: 'element.remove', target: { id: node.id, incarnation: 1 } });
});
assert.equal(result.status, 'applied'); assert.equal(result.operationIds.length, 1);
const operation = engine.getSnapshot().runtime.ledger.operations.get(result.operationIds[0]);
assert.equal(operation.basePresence.temporary, 'absent');
assert.equal(engine.getSnapshot().document.tombstones.get(node.id), 1);
await engine.settle(); assert.equal(port.calls.length, 1);
} finally { await engine.close(); }
});
+55
View File
@@ -166,3 +166,58 @@ test('holds acquired before start retain notified work; stop is terminal even wh
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/);
});
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' }; } });
state.scheduler.start(); state.add('create'); state.add('rename', ['canvas'], ['create']); state.add('move', ['canvas'], ['rename']);
assert.equal(state.ledger().getSnapshot().requests.size, 0);
await tick();
const frozen = [...state.ledger().getSnapshot().requests.values()][0];
assert.deepEqual(frozen.operationIds, ['create', 'rename', 'move']); assert.equal(frozen.phase, 'dispatch-marked'); assert.equal(sent.length, 0);
assert.deepEqual(JSON.parse(frozen.body).operations.map(op => op.id), frozen.operationIds);
barrier.resolve(); await state.scheduler.settle();
assert.equal(sent.length, 1); assert.equal(state.scheduler.getClaims().size, 0);
for (const id of frozen.operationIds) assert.equal(state.ledger().getSnapshot().operations.get(id).phase, 'resolved');
state.scheduler.stop();
});
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']);
state.scheduler.start(); await tick();
assert.deepEqual(sent.map(request => request.operationIds), [['a1'], ['b1']]);
pending.resolve({ outcome: 'applied' }); await state.scheduler.settle();
assert.deepEqual(sent.map(request => request.operationIds), [['a1'], ['b1'], ['a2']]); state.scheduler.stop();
const observed = setup({ compileBatch });
observed.add('before');
observed.ledger().add({ id: 'run', operationType: 'run.start', operationSchemaVersion: 1, sequence: 2, origin: 'user', changes: [], dependencies: ['before'] }, {});
observed.ledger().block('run', ['awaiting-input']); observed.scheduler.notify(['run']);
observed.add('after', ['canvas'], ['before']); observed.scheduler.start(); await observed.scheduler.settle();
assert.deepEqual(observed.calls.map(([, request]) => request.operationIds), [['before'], ['after']]); observed.scheduler.stop();
});
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();
const original = sent[0]; state.add('three', ['canvas'], ['two']); await state.scheduler.settle();
assert.equal(sent.length, 1); assert.equal(state.ledger().getSnapshot().operations.get('three').phase, 'pending');
await state.advance(100); await state.advance(100);
assert.equal(sent.length, 2); assert.equal(sent[1].id, original.id); assert.equal(sent[1].body, original.body); assert.deepEqual(sent[1].operationIds, ['one', 'two']);
assert.equal(state.scheduler.getClaims().get('canvas'), original.id); state.scheduler.stop();
});
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}`] : []);
await state.scheduler.settle(); assert.equal(state.calls.length, 0);
release(); await state.scheduler.settle();
assert.deepEqual(state.calls.map(([, request]) => request.operationIds.length), [64, 64, 2]);
assert.equal(state.ledger().getSnapshot().operations.get('op-129').phase, 'resolved'); state.scheduler.stop();
});
+14
View File
@@ -125,3 +125,17 @@ test('historical applied proof validates frozen identity and coverage, retains n
const tombstone=structuredClone(status.canonical);tombstone.model.tombstones={deleted:2};const lifecycle=port.verifyApplied(request,tombstone);
assert.throws(()=>lifecycle.assertCanonical({document:changed,versions:new Map([[root,5],[child,7]])}),/stale_canonical_lifecycle/);
});
test('ordered batch freezes every logical identity, conditional change and participant version, requiring complete receipt coverage', async () => {
const secondId='90000000-0000-4000-8000-000000000003';
const second={...intent,id:secondId,sequence:2,dependencies:[intent.id],changes:[{...intent.changes[0],before:'renamed',after:'final'}]};
const calls=[],port=createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport:async input=>{calls.push(input);return{status:200,body:{protocol_version:2,request_id:requestId,status:'applied',operation_ids:[intent.id,second.id],model:chunk(false)}}}});
const compiled=port.compileBatch([intent,second],requestId,snapshot),ledger=createLedger();ledger.add(intent,{a:'exists'});ledger.add(second,{a:'exists'});
const request=ledger.freeze({...compiled,id:requestId,operationIds:[intent.id,second.id]});
assert.deepEqual(JSON.parse(request.body).operations,[intent,second].map(({id,changes})=>({id,changes})));
assert.deepEqual(request.expectedVersions,{[root]:3});assert.equal((await port.submit(request,new AbortController().signal)).outcome,'applied');assert.equal(calls[0].body,request.body);
const invalid=createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport:async()=>({status:200,body:{protocol_version:2,request_id:requestId,status:'applied',operation_ids:[second.id],model:chunk(false)}})});
assert.equal((await invalid.query(request,new AbortController().signal)).error,'receipt_coverage_mismatch');
assert.throws(()=>port.compileBatch([second,intent],requestId,snapshot),/invalid_operation_order/);
assert.throws(()=>port.compileBatch([intent,intent],requestId,snapshot),/operation_interpreter_unavailable/);
});