diff --git a/backend/src/services/server-poll-execution.test.ts b/backend/src/services/server-poll-execution.test.ts index cb8a98bd..78ba11cd 100644 --- a/backend/src/services/server-poll-execution.test.ts +++ b/backend/src/services/server-poll-execution.test.ts @@ -54,6 +54,42 @@ assert.equal(effects.length,plan.maintenance.length+4);assert.equal(intents.leng const nativeBefore=nativeCalls.length,effectsBefore=effects.length;assert.deepEqual(await executor.execute(plan),receipt);assert.equal(nativeCalls.length,nativeBefore);assert.equal(effects.length,effectsBefore,"Known durable receipt cannot replay poll or maintenance") state.servers[1].name="later";assert.equal((await executor.execute(plan)).configurationVerified,false,"Historical receipt does not assert current state after later drift") +reset() +let activeJournalWrites=0,nativeArrivals=0 +const nativeTogether=Promise.withResolvers() +const nativeTimeout=setTimeout(()=>nativeTogether.reject(new Error("Four native reads must remain concurrent")),1500) +const journalWrite=async(write:()=>Promise)=>{ + assert.equal(activeJournalWrites,0,"One poll must not queue its own writes against the global journal lock") + activeJournalWrites++ + try{ + await new Promise(resolve=>setImmediate(resolve)) + await write() + }finally{activeJournalWrites--} +} +const noSelfContention=createServerPollExecution({...deps, + intent:value=>journalWrite(()=>deps.intent(value)), + acknowledge:value=>journalWrite(()=>deps.acknowledge(value)), + native:async(...args)=>{ + if(++nativeArrivals===4)nativeTogether.resolve() + await nativeTogether.promise + return deps.native(...args) + }, +}) +try{ + assert.equal((await noSelfContention.execute(await noSelfContention.prepare(request()))).nativeReads,4) +}finally{clearTimeout(nativeTimeout)} + +reset() +let attemptedNativeIntents=0 +const uncertainJournal=createServerPollExecution({...deps,intent:async()=>{ + attemptedNativeIntents++ + throw new Error("Unconfirmed native intent") +}}) +await assert.rejects(uncertainJournal.execute(await uncertainJournal.prepare(request())),/Unconfirmed native intent/) +assert.equal(attemptedNativeIntents,1,"An unconfirmed intent must stop the queued journal writes") +assert.equal(nativeCalls.length,0) +assert.equal(effects.length,0) + for(const mutate of [()=>{global=false},()=>{allowed=[1]},()=>{state.settings.traffic!.retentionDays=30},()=>{state.servers[0].routingEndpointEpoch=randomUUID()},()=>{state.partitions[0].opaque="changed"},()=>{authority="b".repeat(64)}]){ reset();plan=await executor.prepare(request());mutate();await assert.rejects(executor.execute(plan));assert.equal(nativeCalls.length,0);assert.equal(effects.length,0) } diff --git a/backend/src/services/server-poll-execution.ts b/backend/src/services/server-poll-execution.ts index c2f61db6..b3ddd6b4 100644 --- a/backend/src/services/server-poll-execution.ts +++ b/backend/src/services/server-poll-execution.ts @@ -107,15 +107,25 @@ export function createServerPollExecution(deps:ServerPollDependencies){ const nativeGate=async()=>{await gate()} try{ const start=deps.now(),server=current.context.servers.find(row=>row.id===plan.request.serverId)! + // Native reads can overlap. Their durable journal writes share one storage lock. + let journalTail=Promise.resolve() + const writeNativeJournal=(write:()=>Promise)=>{ + journalTail=journalTail.then(async()=>{ + await access() + await write() + }) + return journalTail + } const observations=await Promise.allSettled(SERVER_POLL_NATIVE_READS.map(async key=>{ - await access();await deps.intent({receiptId:plan.request.receiptId,stage:"native."+key,beforeProof:current.proof,effectProof:await deps.proof({read:key,source:server})});await nativeGate() + await writeNativeJournal(async()=>deps.intent({receiptId:plan.request.receiptId,stage:"native."+key,beforeProof:current.proof,effectProof:await deps.proof({read:key,source:server})}));await nativeGate() let dispatched=false let result:unknown,readFailed=false,nativeDiagnostic:ReturnType|undefined try{result=await deps.native(key,structuredClone(server),{signal,deadlineAt:deadline,beforeDispatch:async()=>{if(dispatched)pollProblem("STATE_CONFLICT","Повторный dispatch native poll запрещён");await nativeGate();if(dispatched)pollProblem("STATE_CONFLICT","Повторный dispatch native poll запрещён");dispatched=true;nativeReads++}})}catch(error){if(error instanceof ServerPollProblem||signal?.aborted)throw error;nativeDiagnostic=capturePollError(error);readFailed=true} if(!dispatched)throw carryCapturedPollError(new ServerPollProblem("STATE_UNAVAILABLE","Native transport пропустил dispatch fence"),nativeDiagnostic,"native.fence") - await access() - if(!readFailed){let bytes:number;try{bytes=Buffer.byteLength(JSON.stringify(result))}catch{pollProblem("STATE_UNAVAILABLE","Некорректный native ответ poll")};if(bytes>1048576)pollProblem("STATE_UNAVAILABLE","Превышен native budget poll")} - await deps.acknowledge({receiptId:plan.request.receiptId,stage:"native."+key,resultProof:await deps.proof(readFailed?{unavailable:true}:result),outcome:readFailed?"read-failed":"confirmed"}) + await writeNativeJournal(async()=>{ + if(!readFailed){let bytes:number;try{bytes=Buffer.byteLength(JSON.stringify(result))}catch{pollProblem("STATE_UNAVAILABLE","Некорректный native ответ poll")};if(bytes>1048576)pollProblem("STATE_UNAVAILABLE","Превышен native budget poll")} + await deps.acknowledge({receiptId:plan.request.receiptId,stage:"native."+key,resultProof:await deps.proof(readFailed?{unavailable:true}:result),outcome:readFailed?"read-failed":"confirmed"}) + }) return readFailed?undefined:result })) const failed=observations.find(result=>result.status==="rejected");if(failed?.status==="rejected")throw failed.reason