Separate request Root durability barriers from complete storage flushes

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.durableGeneration<expected);assert.equal(engine.getDurability().durableRootRevision,confirmed.rootRevision);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'partial-7');assert.equal(sends,0);
+  broken=false;await engine.persist();assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,expected);const rows=storage.saved.slice(saved);assert.deepEqual(rows.map(image=>image.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);gate.resolve();await Promise.all([empty,edited]);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();}
+});