diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts index 64fd250..fb52e47 100644 --- a/src/lib/canvas-runtime/contracts.ts +++ b/src/lib/canvas-runtime/contracts.ts @@ -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; query(request: FrozenRequest, signal: AbortSignal): Promise; /** Pure codec, merging only participants into the latest base, never an old working snapshot. */ diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index b20afa3..927abc6 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -79,7 +79,7 @@ export async function openProjectEngine(options: EngineOptions) { if (resource && !resource.stored) dependencies.add(resource.operationId); } const basePresence: Record = 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)); diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index f214f43..c0be443 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -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; query(request: FrozenRequest, signal: AbortSignal): Promise; 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(), active = new Map; kind: FrozenRequest['kind'] }>(); const holds = new Set(); + const ordered: { id: Id; sequence: number }[] = []; + let scheduledPump: Promise | undefined; const laneWaiters = new Map, Set>(), deferredRetries = new Map(); 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(); 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): 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(); 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(); 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(); }, }); } diff --git a/src/lib/canvas-service/foundry.ts b/src/lib/canvas-service/foundry.ts index 1f750da..9cd0215 100644 --- a/src/lib/canvas-service/foundry.ts +++ b/src/lib/canvas-service/foundry.ts @@ -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 = {}; + 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 = {}; - 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) { diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index 9d5d3a8..a3e2560 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -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); diff --git a/tests/headless/engine.test.mjs b/tests/headless/engine.test.mjs index 0e8a195..79ffe6e 100644 --- a/tests/headless/engine.test.mjs +++ b/tests/headless/engine.test.mjs @@ -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(); } +}); diff --git a/tests/headless/scheduler.test.mjs b/tests/headless/scheduler.test.mjs index b3d1d49..1e71aff 100644 --- a/tests/headless/scheduler.test.mjs +++ b/tests/headless/scheduler.test.mjs @@ -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(); +}); diff --git a/tests/headless/service.test.mjs b/tests/headless/service.test.mjs index 79950d7..85c458c 100644 --- a/tests/headless/service.test.mjs +++ b/tests/headless/service.test.mjs @@ -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/); +});