fix(mcp): упорядочить записи журнала опроса
This commit is contained in:
1 parent
765b9a1658
commit
d79fb3815a
2 files changed
+48
-2
No files matched your search
@@ -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")
|
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")
|
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<void>()
|
||||||
|
const nativeTimeout=setTimeout(()=>nativeTogether.reject(new Error("Four native reads must remain concurrent")),1500)
|
||||||
|
const journalWrite=async(write:()=>Promise<void>)=>{
|
||||||
|
assert.equal(activeJournalWrites,0,"One poll must not queue its own writes against the global journal lock")
|
||||||
|
activeJournalWrites++
|
||||||
|
try{
|
||||||
|
await new Promise<void>(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)}]){
|
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)
|
reset();plan=await executor.prepare(request());mutate();await assert.rejects(executor.execute(plan));assert.equal(nativeCalls.length,0);assert.equal(effects.length,0)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -107,15 +107,25 @@ export function createServerPollExecution(deps:ServerPollDependencies){
|
|||||||
const nativeGate=async()=>{await gate()}
|
const nativeGate=async()=>{await gate()}
|
||||||
try{
|
try{
|
||||||
const start=deps.now(),server=current.context.servers.find(row=>row.id===plan.request.serverId)!
|
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<void>)=>{
|
||||||
|
journalTail=journalTail.then(async()=>{
|
||||||
|
await access()
|
||||||
|
await write()
|
||||||
|
})
|
||||||
|
return journalTail
|
||||||
|
}
|
||||||
const observations=await Promise.allSettled(SERVER_POLL_NATIVE_READS.map(async key=>{
|
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 dispatched=false
|
||||||
let result:unknown,readFailed=false,nativeDiagnostic:ReturnType<typeof capturePollError>|undefined
|
let result:unknown,readFailed=false,nativeDiagnostic:ReturnType<typeof capturePollError>|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}
|
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")
|
if(!dispatched)throw carryCapturedPollError(new ServerPollProblem("STATE_UNAVAILABLE","Native transport пропустил dispatch fence"),nativeDiagnostic,"native.fence")
|
||||||
await access()
|
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")}
|
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 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
|
return readFailed?undefined:result
|
||||||
}))
|
}))
|
||||||
const failed=observations.find(result=>result.status==="rejected");if(failed?.status==="rejected")throw failed.reason
|
const failed=observations.find(result=>result.status==="rejected");if(failed?.status==="rejected")throw failed.reason
|
||||||
|
|||||||
Reference in new issue
Block a user