diff --git a/headless/README.md b/headless/README.md index 36e2c82..0b052e6 100644 --- a/headless/README.md +++ b/headless/README.md @@ -818,3 +818,32 @@ produce their own pins and completeness result; all supported transitions, including resolved evidence, are validated. Cross-schema fallback pins and owner order are preserved. The scanner is internal to one retention pass and is not stored across Roots, projects or registry changes. + +Query view construction no longer leaves a callback capturing the previous query +in the retained method context. Latest methods retain their own document/tables, +not the complete chain of discarded generations. Explicitly held historical +queries remain valid. A child-process GC check covers raw queries and the actual +editor with query coverage installed. + +Persistence uses one writer and the existing pending Root Map. `writePrefix` +captures a finite Root boundary, validates every receipt and publishes at most one +durable-generation update for that prefix. The receipt-derived maximum survives +a failed metadata publication as a writer publication obligation. The writer +promise is installed before work, so synchronous listeners cannot start another +writer. No existing Root is skipped or rewritten. + +Transport's internal `persistThrough` waits for its fixed request/dispatch Root; +it does not chase later local edits. Public `persist` still flushes all pending +Roots and the derived publication obligation, including metadata echoes. Failed +writes reject their barriers and retain the exact suffix for retry. Network +capacity grants recheck leader/identity before calling the adapter. A fenced +scheduler pump reports failed state publication without an unhandled rejection. +`settle` repeats until the Root and current task queues stabilize, preserving +future retry/held-job semantics instead of relying on two fixed passes. + +The actual-page test uses bounded CDP requests and rejects disconnected/crashed +or dialog-blocked targets without accepting dialogs. A failed protocol call cannot +hang a second metrics read. Browser cleanup escalates only the fixture's own +process after its exit grace period. Optional heap/receipt sampling and a pinned +baseline override affect only fixture copies and are diagnostic, not product APIs +or production performance claims. diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 22634e1..d3aa9b1 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -85,7 +85,7 @@ export async function openProjectEngine(options: EngineOptions) { function getDurability(){const rootRevision=editor.getSnapshot().rootRevision;if(!durabilityCache||durabilityCache.rootRevision!==rootRevision||durabilityCache.durableRootRevision!==durableRootRevision||durabilityCache.closing!==closing||durabilityCache.closed!==closed||durabilityCache.error!==durabilityError)durabilityCache=Object.freeze({rootRevision,durableRootRevision,closing,closed,...(durabilityError?{error:durabilityError}:{})});return durabilityCache;} function notifyDurability(){for(const listener of [...durabilityListeners])try{listener();}catch(error){report(error);}} const createId = options.createId ?? (() => crypto.randomUUID()); - const network = createNetworkGate(2); + const network = createNetworkGate(2,()=>{if(!leader.active)throw fault('storage_leader_lost');if(options.isCurrentIdentity&&!options.isCurrentIdentity())throw fault('identity_changed');}); const depot=options.assetDepot??(options.storage.assetDepot?createAssetDepot({scope:options.scope,port:options.storage.assetDepot,current:()=>!closing&&!closed&&(!options.isCurrentIdentity||options.isCurrentIdentity()),digest:sourceDigest}):undefined); if(depot&&(depot.scope.accountId!==options.scope.accountId||depot.scope.workspaceId!==options.scope.workspaceId))throw fault('asset_depot_scope_mismatch'); const adoptionTasks=new Map>(),adoptedFiles=new Map(); @@ -199,28 +199,47 @@ export async function openProjectEngine(options: EngineOptions) { const pending = new Map(); if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot()); let writing: Promise | undefined, scheduled = false, persistenceFailed = false; - function persist(): Promise { + // Receipt-derived publication obligation survives a failed metadata publication. + let confirmedLedgerGeneration=editor.getSnapshot().runtime.ledger.durableGeneration; + function writePrefix(throughRevision:number):Promise { if (writing) return writing; - const task = (async () => { - while (pending.size) { - const [revision, snapshot] = pending.entries().next().value!; - const sourceFiles=importFiles(snapshot.runtime.imports); - const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)||sourceFiles.has(id)).map(([id, bytes]) => ({ id, bytes })); - const deleteFiles=[...[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.id),...[...sourceFiles].filter(([id,entry])=>entry.file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id),...[...(snapshot.runtime.orphanFiles??[])].filter(([id,file])=>file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id)]; - const depot=options.assets?.authority&&options.storage.assetDepot?await prepareDepotCut({snapshot,registry:options.registry,storage:options.storage,leader,authority:options.assets.authority.scope,files:fileWrites,adoptedFiles,displays:displayPins,temporary:admissionPins(snapshot)}):undefined; - const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles,...(depot?{depot}:{}) }); - if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt'); - durableRootRevision = revision;durableImportSeals=importSeals(snapshot.runtime.imports); pending.delete(revision);notifyDurability(); - for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);} - for (const file of files) fileWrites.delete(file.id); - for(const [id,entry]of adoptedFiles)if(entry.rootRevision<=snapshot.rootRevision&&depot?.writes.some(write=>write.ref.asset.id===id&&write.bytes===entry.bytes))adoptedFiles.delete(id); - if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation)); + const task = Promise.resolve().then(async () => { + const publishDurability=()=>{if(editor.getSnapshot().runtime.ledger.durableGenerationledger.markDurable(confirmedLedgerGeneration));}; + try { + while (pending.size) { + const [revision, snapshot] = pending.entries().next().value!; + if(revision>throughRevision)break; + const sourceFiles=importFiles(snapshot.runtime.imports); + const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)||sourceFiles.has(id)).map(([id, bytes]) => ({ id, bytes })); + const deleteFiles=[...[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.id),...[...sourceFiles].filter(([id,entry])=>entry.file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id),...[...(snapshot.runtime.orphanFiles??[])].filter(([id,file])=>file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id)]; + const depot=options.assets?.authority&&options.storage.assetDepot?await prepareDepotCut({snapshot,registry:options.registry,storage:options.storage,leader,authority:options.assets.authority.scope,files:fileWrites,adoptedFiles,displays:displayPins,temporary:admissionPins(snapshot)}):undefined; + const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles,...(depot?{depot}:{}) }); + if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt'); + confirmedLedgerGeneration=Math.max(confirmedLedgerGeneration,snapshot.runtime.ledger.generation); + durableRootRevision = revision;durableImportSeals=importSeals(snapshot.runtime.imports); pending.delete(revision);notifyDurability(); + for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);} + for (const file of files) fileWrites.delete(file.id); + for(const [id,entry]of adoptedFiles)if(entry.rootRevision<=snapshot.rootRevision&&depot?.writes.some(write=>write.ref.asset.id===id&&write.bytes===entry.bytes))adoptedFiles.delete(id); + } + } catch(error) { + // Preserve the known durable prefix even if a later write fails. + publishDurability();throw error; } + // One derived publication per captured prefix, not one per receipt. + publishDurability(); persistenceFailed = false;durabilityError=undefined;notifyDurability(); - })(); + }); writing = task.catch(error => { persistenceFailed = true;const failure=error&&typeof error==='object'?error as {code?:unknown;message?:unknown}:{};durabilityError=typeof failure.code==='string'&&failure.code?failure.code:typeof failure.message==='string'&&failure.message?failure.message:'persistence_failed';notifyDurability(); throw error; }).finally(() => { writing = undefined;if(!persistenceFailed){if(loadingCompletionRevision!==undefined&&durableRootRevision>=loadingCompletionRevision&&!editor.getSnapshot().runtime.loading){const release=loadingRelease;loadingRelease=undefined;loadingCompletionRevision=undefined;release?.();}if(!closing&&!closed)scheduler.refreshAdmission();} }); return writing; } + function persistThrough(requiredRevision=editor.getSnapshot().rootRevision):Promise { + if(durableRootRevision>=requiredRevision)return Promise.resolve(); + return writePrefix(requiredRevision).then(()=>durableRootRevision { + do {await writePrefix(editor.getSnapshot().rootRevision);} + while(pending.size||writing||confirmedLedgerGeneration>editor.getSnapshot().runtime.ledger.durableGeneration); + } function collectResources(settings:{retentionMs:number;maxFiles?:number}) { assertWritable(); if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable'); @@ -442,7 +461,7 @@ export async function openProjectEngine(options: EngineOptions) { report(error); return { blocked: true }; } } - const scheduler = createScheduler({ admit:intent=>{const root=editor.getSnapshot();if(!importsMaySend(root.runtime.imports,durableImportSeals))return false;if(root.runtime.loading&&root.runtime.loading.version<3)return true;if(closing||closed||loadingCompletionRevision!==undefined&&durableRootRevision=root.rootRevision);},getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist, + const scheduler = createScheduler({ admit:intent=>{const root=editor.getSnapshot();if(!importsMaySend(root.runtime.imports,durableImportSeals))return false;if(root.runtime.loading&&root.runtime.loading.version<3)return true;if(closing||closed||loadingCompletionRevision!==undefined&&durableRootRevision=root.rootRevision);},getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist:persistThrough, createId: options.createId ?? (() => crypto.randomUUID()), participants: intent => { if(intent.historyEffect)validateCreationSizeEffect(intent.historyEffect);return operationCanvases(intent, editor.getSnapshot().document); }, compile(intent, id) { @@ -527,7 +546,7 @@ export async function openProjectEngine(options: EngineOptions) { if(code!=='run_waiting_for_receipts')report(error); } } - const runTracker=createRunTracker({clock:runClock,getRuns:()=>editor.getSnapshot().runtime.runs,persist,update:updateRun,accept:acceptRunPage, + const runTracker=createRunTracker({clock:runClock,getRuns:()=>editor.getSnapshot().runtime.runs,persist:persistThrough,update:updateRun,accept:acceptRunPage, poll:(run,signal)=>options.runs ? network.run(signal,()=>options.runs!.poll(run,signal)) : Promise.reject(fault('run_interpreter_unavailable')),onError:report}); const trackedRuns=new Set(initial.runs.keys()); const sessions = createSessions(editor); @@ -1051,7 +1070,14 @@ export async function openProjectEngine(options: EngineOptions) { getDurability, subscribeDurability(listener:()=>void){if(closed)return()=>{};durabilityListeners.add(listener);return()=>{durabilityListeners.delete(listener);};}, retry: async () => { if(recovery)await recovery.cancel();assertWritable(false);await persist();retryDefaultSizes();const runtime=editor.getSnapshot().runtime,ids=[...runtime.resources.values()].filter(resource=>!resource.digest&&runtime.ledger.operations.get(resource.operationId)?.blockers.some(code=>digestFailures.has(code))).map(resource=>resource.operationId);if(ids.length){mutate(ledger=>{for(const id of ids)ledger.block(id,['waiting_digest']);});prepareDigests();}if(options.images){editor.updateRuntime(previous=>{const resources=new Map(previous.resources);let changed=false;for(const [id,resource]of resources)if(resource.image?.version===1&&resource.image.status==='failed'){resources.set(id,freeze({...resource,image:{version:1 as const,status:'pending' as const}}));changed=true;}return changed?Object.freeze({...previous,resources:readMap(resources)}):previous;});prepareImages();}for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(sourceNeedsSeal(job)){await importLoads.get(job.id);await resetImportReader(job.id);await loadImport(job.id);}await scheduler.resume();await runTracker.retry(); }, - async settle() { await Promise.resolve();await Promise.allSettled([...adoptionTasks.values()]);await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while(verificationTasks.size)await Promise.allSettled([...verificationTasks.values()].map(task=>task.promise));while (digestTasks.size) await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; await scheduler.settle(); await runTracker.settle(); await persist();await scheduler.settle();await persist(); }, + async settle() { + let revision:number; + do { + revision=editor.getSnapshot().rootRevision; + await Promise.resolve();await Promise.allSettled([...adoptionTasks.values()]);await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle()));while(verificationTasks.size)await Promise.allSettled([...verificationTasks.values()].map(task=>task.promise));while(digestTasks.size)await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; + await scheduler.settle();await runTracker.settle();await persist();await scheduler.settle();await persist(); + } while(revision!==editor.getSnapshot().rootRevision||pending.size||writing||adoptionTasks.size||copyPreparations.size||importLoads.size||verificationTasks.size||digestTasks.size||imageTasks.size||sizingTask); + }, async close() { if (closed) return; closing = true;notifyDurability();loadingReadController?.abort();await loadingRead?.catch(()=>{});unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));digestLifecycle.abort();for(const {controller}of verificationTasks.values())controller.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]);for(const controller of adoptionTasks.keys())controller.abort();await Promise.allSettled([...adoptionTasks.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop(); diff --git a/src/lib/canvas-runtime/network.ts b/src/lib/canvas-runtime/network.ts index 3454b50..cfd4ee5 100644 --- a/src/lib/canvas-runtime/network.ts +++ b/src/lib/canvas-runtime/network.ts @@ -2,7 +2,7 @@ import { fault } from '../canvas-core/values.ts'; import { createDeque } from './structures.ts'; /** Shared bounded transport capacity for writes, uploads and run reads. */ -export function createNetworkGate(limit = 2) { +export function createNetworkGate(limit = 2, beforeRun?:()=>void) { if(!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_concurrency'); let active = 0; const waiting = createDeque<{ signal: AbortSignal; grant(): void; cancelled: boolean }>(); @@ -21,6 +21,6 @@ export function createNetworkGate(limit = 2) { } return Object.freeze({ async run(signal: AbortSignal, callback: () => Promise): Promise { await enter(signal); - try { signal.throwIfAborted();return await callback(); } finally { release(); } + try { signal.throwIfAborted();beforeRun?.();return await callback(); } finally { release(); } } }); } diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index 225f481..e0fcce6 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -210,6 +210,10 @@ export function createScheduler(options: SchedulerOptions) { 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); } } + } catch(error) { + // A retired identity can reject both the attempt and its blocked-state write. + // Report the failure without rejecting an unobserved scheduled pump promise. + report(error); } finally { pumping = false; armTimer(); if (!stopped && !holds.size && ready.size && active.size < concurrency && pumpTimer === undefined) pumpTimer = clock.setTimer(() => { pumpTimer = undefined; pump(); }, 0); diff --git a/tests/headless/engine.test.mjs b/tests/headless/engine.test.mjs index bc54a6e..dd5eb56 100644 --- a/tests/headless/engine.test.mjs +++ b/tests/headless/engine.test.mjs @@ -141,16 +141,16 @@ test('a fresh canonical read replaces a clean recovered base, keeps history and await reopened.engine.close(); }); -test('recovery waits for admitted writes, reads a complete base, fences edits and publishes without changing history', async()=>{ - const storage=storageFixture(),port=persistenceFixture(),response=deferred(),realSubmit=port.submit,readGate=deferred(); +test('recovery waits for admitted writes, reads a complete base, fences edits and publishes without changing history', {timeout:5000}, async()=>{ + const storage=storageFixture(),port=persistenceFixture(),response=deferred(),realSubmit=port.submit,readGate=deferred(),readEntered=deferred(); let reads=0;port.submit=async request=>{await response.promise;return realSubmit(request);}; - port.read=async()=>{reads++;await readGate.promise;return{document:applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'local',beforePresent:true,afterPresent:true},{kind:'field',target,field:'geometry',before:original.elements.get('a').geometry,after:{...original.elements.get('a').geometry,x:300}}]),versions:new Map([['canvas',3]])};}; + port.read=async()=>{reads++;readEntered.resolve();await readGate.promise;return{document:applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'local',beforePresent:true,afterPresent:true},{kind:'field',target,field:'geometry',before:original.elements.get('a').geometry,after:{...original.elements.get('a').geometry,x:300}}]),versions:new Map([['canvas',3]])};}; const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await tick(); const preparing=engine.prepareRecovery();await tick();assert.equal(reads,0); assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'blocked'}}).code,'recovery_in_progress'); assert.equal(engine.startRun({}).code,'recovery_in_progress');assert.equal(engine.refresh(original,new Map([['canvas',1]])).code,'recovery_in_progress'); await assert.rejects(engine.prepareRecovery(),/recovery_in_progress/); - response.resolve();await tick();assert.equal(reads,1);const history=engine.getSnapshot().history; + response.resolve();await readEntered.promise;assert.equal(reads,1);const history=engine.getSnapshot().history; readGate.resolve();const session=await preparing;assert.equal(session.conflict,undefined);assert.equal(session.canonical.document.elements.get('a').geometry.x,300); assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,0);assert.equal(engine.editor.history.undo().code,'recovery_in_progress'); assert.equal((await session.apply()).status,'applied');assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,300);assert.equal(engine.getSnapshot().history,history); @@ -296,3 +296,41 @@ test('batch-capable persistence sends an earlier real edit separately and never for(const id of [...created.operationIds,...removed.operationIds])assert.equal(engine.getSnapshot().runtime.ledger.operations.get(id).resolution,'elided');assert.equal(engine.getSnapshot().runtime.acknowledged.tombstones.has(n.id),false); }finally{await engine.close();} }); + +const idleClock=()=>({now:()=>0,setTimer:()=>Symbol('idle'),clearTimer(){}}); +test('a pending write train preserves every business root and publishes one durable-generation echo per captured prefix',async()=>{ + const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;let sends=0;port.submit=async request=>{sends++;const stored=storage.image();assert.ok(stored.payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));await network.promise;return{outcome:'unresolved'};};try{ + await engine.persist();const start=engine.getSnapshot(),saved=storage.saved.length;let held=false,markers=[],through=start.runtime.ledger.durableGeneration;const unsubscribe=engine.editor.subscribe((_,root)=>{if(root.runtime.ledger.durableGeneration>through){markers.push(root.runtime.ledger.durableGeneration);through=root.runtime.ledger.durableGeneration;}}); + storage.commit=async(...args)=>{if(!held){held=true;entered.resolve();await gate.promise;}return commit(...args);}; + for(let i=0;i<8;i++)assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'train-'+i}}).status,'applied'); + const requested=engine.getSnapshot().runtime.ledger.generation,saving=engine.persist();await entered.promise;await tick();const expected=engine.getSnapshot().runtime.ledger.generation,queued=engine.getSnapshot().rootRevision-start.rootRevision;assert.equal(sends,0);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,start.runtime.ledger.durableGeneration);gate.resolve();await saving;unsubscribe(); + assert.deepEqual(markers,[...new Set([requested,expected])]);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,expected);assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);const rows=storage.saved.slice(saved);assert.equal(rows.length,queued+markers.length);assert.deepEqual(rows.map(image=>image.rootRevision),Array.from({length:queued+markers.length},(_,i)=>start.rootRevision+i+1));assert.deepEqual(rows.slice(0,8).map(image=>image.payload.document.elements[0].data.title),Array.from({length:8},(_,i)=>'train-'+i));assert.equal(decodeRecovery(storage.image(),registry).runtime.ledger.durableGeneration,expected);await tick();assert.equal(sends,1); + }finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();} +}); +test('a later quota failure records only confirmed generation and retries the full continuous suffix',async()=>{ + const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;let sends=0;port.submit=async request=>{sends++;const stored=storage.image();assert.ok(stored.payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));await network.promise;return{outcome:'unresolved'};};try{ + await engine.persist();const start=engine.getSnapshot(),saved=storage.saved.length;let writes=0,broken=true;storage.commit=async(...args)=>{writes++;if(writes===1){entered.resolve();await gate.promise;}if(broken&&writes===4)throw Error('quota-later');return commit(...args);}; + for(let i=0;i<8;i++)engine.editor.dispatch({type:'element.data',target,patch:{title:'partial-'+i}});const saving=engine.persist();await entered.promise;await tick();const expected=engine.getSnapshot().runtime.ledger.generation,queued=engine.getSnapshot().rootRevision-start.rootRevision;assert.equal(sends,0);gate.resolve();await assert.rejects(saving,/quota-later/);const confirmed=storage.saved.at(-1);assert.equal(storage.saved.length-saved,3);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,confirmed.payload.runtime.ledger.generation);assert.ok(engine.getSnapshot().runtime.ledger.durableGenerationimage.rootRevision),Array.from({length:rows.length},(_,i)=>start.rootRevision+i+1));assert.equal(decodeRecovery(storage.image(),registry).document.elements.get('a').data.title,'partial-7');assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision); + }finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();} +}); +test('an invalid receipt cannot advance confirmed generation and the original cut remains retryable',async()=>{ + const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;let bad=true,held=false,sends=0;port.submit=async()=>{sends++;await network.promise;return{outcome:'unresolved'};};try{ + await engine.persist();const start=engine.getSnapshot(),saved=storage.saved.length;storage.commit=async(...args)=>{if(!held){held=true;entered.resolve();await gate.promise;}const receipt=await commit(...args);return bad?{durableRootRevision:receipt.durableRootRevision+1}:receipt;};engine.editor.dispatch({type:'element.data',target,patch:{title:'receipt-test'}});const saving=engine.persist();await entered.promise;await tick();gate.resolve();await assert.rejects(saving,/invalid_durable_receipt/);assert.equal(engine.getDurability().durableRootRevision,start.rootRevision);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,start.runtime.ledger.durableGeneration);assert.equal(sends,0);const first=storage.saved[saved];bad=false;await engine.persist();const retry=storage.saved[saved+1];assert.deepEqual(retry,first);assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,engine.getSnapshot().runtime.ledger.generation); + }finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();} +}); +test('a failed durability publication retains its verified receipt through retry without rewriting saved business cuts',async()=>{ + const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),network=deferred(),commit=storage.commit;let lose=true;port.submit=async()=>{await network.promise;return{outcome:'unresolved'};};try{ + await engine.persist();const leader=storage.owner(),saved=storage.saved.length;storage.commit=async(...args)=>{const receipt=await commit(...args);if(lose&&args[1].image.rootRevision===engine.getSnapshot().rootRevision){lose=false;leader.active=false;}return receipt;};engine.editor.dispatch({type:'element.data',target,patch:{title:'publication-retry'}});await assert.rejects(engine.persist(),/storage_leader_lost/);const confirmed=storage.saved.at(-1);assert.equal(engine.getDurability().durableRootRevision,confirmed.rootRevision);assert.ok(confirmed.payload.runtime.ledger.generation>engine.getSnapshot().runtime.ledger.durableGeneration);const count=storage.saved.length;leader.active=true;let activeWrites=0,maxWrites=0;const nested=[];storage.commit=async(...args)=>{activeWrites++;maxWrites=Math.max(maxWrites,activeWrites);try{await tick();return await commit(...args);}finally{activeWrites--;}};const unsubscribe=engine.editor.subscribe((_,root)=>{if(root.runtime.ledger.durableGeneration===confirmed.payload.runtime.ledger.generation)nested.push(engine.persist());});await engine.persist();await Promise.all(nested);unsubscribe();assert.equal(maxWrites,1);assert.equal(storage.saved.length,count+1);assert.equal(storage.saved.at(-1).rootRevision,confirmed.rootRevision+1);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,confirmed.payload.runtime.ledger.generation);assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);assert.ok(storage.saved.length>saved); + }finally{const leader=storage.owner();if(leader)leader.active=true;network.resolve();storage.commit=commit;await engine.close();} +}); +test('a caller arriving after an empty writer drain still waits for its own newly published Root', {timeout:5000},async()=>{ + const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;port.submit=async()=>{await network.promise;return{outcome:'unresolved'};};try{ + await engine.persist();storage.commit=async(...args)=>{entered.resolve();await gate.promise;return commit(...args);};const empty=engine.persist();let edited,required;await new Promise(resolve=>queueMicrotask(()=>{engine.editor.dispatch({type:'element.data',target,patch:{title:'after-empty-writer'}});required=engine.getSnapshot().rootRevision;edited=engine.persist();resolve();}));assert.equal(await Promise.race([entered.promise.then(()=>'write'),edited.then(()=>'returned')]),'write');assert.ok(engine.getDurability().durableRootRevision=required);assert.equal(storage.image().payload.document.elements[0].data.title,'after-empty-writer'); + }finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();} +}); +test('a durable dispatch prefix can send while a later local Root is still waiting for storage', {timeout:10000},async()=>{ + const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),body=deferred(),bodyEntered=deferred(),later=deferred(),laterEntered=deferred(),sent=deferred(),network=deferred(),commit=storage.commit;let bodyCut,after=Infinity,flushed=false;port.submit=async request=>{assert.ok(storage.image().payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));sent.resolve();await network.promise;return{outcome:'unresolved'};};try{ + await engine.persist();storage.commit=async(...args)=>{const image=args[1].image;if(bodyCut===undefined&&image.payload.runtime.ledger.requests.some(([,row])=>row.phase==='dispatch-marked')){bodyCut=image.rootRevision;bodyEntered.resolve();await body.promise;}if(image.rootRevision>after){laterEntered.resolve();await later.promise;}return commit(...args);};engine.editor.dispatch({type:'element.data',target,patch:{title:'first'}});const saving=engine.persist().then(()=>{flushed=true;});await bodyEntered.promise;after=engine.getSnapshot().rootRevision;engine.editor.dispatch({type:'element.geometry',target,patch:{x:42}});body.resolve();await laterEntered.promise;await new Promise((resolve,reject)=>{const timer=setTimeout(()=>reject(Error('durable request waited for a later Root')),2000);sent.promise.then(()=>{clearTimeout(timer);resolve();},reject);});assert.equal(flushed,false);assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,42);later.resolve();await saving;assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision); + }finally{body.resolve();later.resolve();network.resolve();storage.commit=commit;await engine.close();} +}); diff --git a/tests/headless/loading-engine.test.mjs b/tests/headless/loading-engine.test.mjs index 53088e7..cfa0547 100644 --- a/tests/headless/loading-engine.test.mjs +++ b/tests/headless/loading-engine.test.mjs @@ -129,7 +129,7 @@ test('certified scopes distinguish a complete child from an incomplete parent an for(const mutate of [image=>delete image.payload.runtime.loading.canvasElementCounts,image=>image.payload.runtime.loading.version=1,image=>image.payload.runtime.loading.version=99,image=>image.payload.runtime.loading.canvasElementCounts[child]=0]){const image=structuredClone(source);mutate(image);assert.throws(()=>decodeRecovery(image,registry));} await engine.close();engine=await f.open({loading});assert.equal(engine.editor.getQueries().membersResult(child).complete,true);assert.equal(engine.getSnapshot().document.elements.get('C').data.text,'child edit'); assert.equal(engine.loading.accept(full,versions,{...loading,receivedElements:5}).status,'applied');const before=engine.getSnapshot(),oldQueries=engine.editor.getQueries();assert.equal(oldQueries.lookupElement('missing').status,'unloaded');assert.equal(oldQueries.membersResult(root).complete,true); - assert.equal((await engine.loading.complete(full,versions,{sourceToken:loading.sourceToken})).status,'applied');assert.equal(engine.getSnapshot().revision,before.revision,'coverage-only completion must not invent a document revision');assert.equal(engine.editor.getQueries().lookupElement('missing').status,'absent');assert.equal(engine.editor.getQueries().coverage.projectComplete,true);assert.equal(oldQueries.lookupElement('missing').status,'unloaded');await engine.settle();assert.equal(f.server().elements.get('C').data.text,'child edit');assert.equal(f.storage.image().formatVersion<7,true); + const completion=await engine.loading.complete(full,versions,{sourceToken:loading.sourceToken});assert.equal(completion.status,'applied');assert.equal(completion.revision,before.revision,'coverage-only completion must not invent a document revision');assert.equal(engine.editor.getQueries().lookupElement('missing').status,'absent');assert.equal(engine.editor.getQueries().coverage.projectComplete,true);assert.equal(oldQueries.lookupElement('missing').status,'unloaded');await engine.settle();assert.equal(f.server().elements.get('C').data.text,'child edit');assert.equal(f.storage.image().formatVersion<7,true); }finally{await engine.close();} }); test('legacy prefixes do not infer scope completion; a changed count manifest cannot replace the published root',async()=>{ diff --git a/tests/headless/network.test.mjs b/tests/headless/network.test.mjs new file mode 100644 index 0000000..2ff8de0 --- /dev/null +++ b/tests/headless/network.test.mjs @@ -0,0 +1,10 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createNetworkGate} from '../../src/lib/canvas-runtime/network.ts'; +const deferred=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{resolve,promise};}; +test('transport rechecks identity after waiting for capacity and releases a rejected permit',async()=>{ + let current=true,calls=0;const gate=createNetworkGate(1,()=>{if(!current)throw Error('identity_changed');}),hold=deferred(),signal=new AbortController().signal,first=gate.run(signal,async()=>{calls++;await hold.promise;}),second=assert.rejects(gate.run(signal,async()=>{calls++;}),/identity_changed/);await Promise.resolve();current=false;hold.resolve();await first;await second;assert.equal(calls,1);current=true;await gate.run(signal,async()=>{calls++;});assert.equal(calls,2); +}); +test('an aborted queued transport cannot pass the effect guard or consume the next capacity grant',async()=>{ + let guards=0,calls=0;const gate=createNetworkGate(1,()=>guards++),hold=deferred(),signal=new AbortController().signal,abort=new AbortController(),first=gate.run(signal,async()=>{calls++;await hold.promise;}),second=assert.rejects(gate.run(abort.signal,async()=>{calls++;}),/cancelled/);await Promise.resolve();abort.abort(Error('cancelled'));await second;hold.resolve();await first;await gate.run(signal,async()=>{calls++;});assert.equal(calls,2);assert.equal(guards,2); +}); diff --git a/tests/headless/scheduler.test.mjs b/tests/headless/scheduler.test.mjs index 1e71aff..1bbc6f5 100644 --- a/tests/headless/scheduler.test.mjs +++ b/tests/headless/scheduler.test.mjs @@ -221,3 +221,7 @@ test('bounded batches drain through settle without waiting for fake timers and h 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(); }); + +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(); +});