java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化

Browse Source
This commit is contained in:
ouyangxq
2026-08-28 09:19:18 +08:00
parent e72ac14425
commit a719a903cd
10 changed files with 174 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/admin-test-env-modules/phase3-loading-finalize.mjs
+36 -5
View File
@@ -79,6 +79,26 @@ async function request(context, method, path, body, headers = {}) {
}
}
async function waitForCompletion(context, taskDetailPath, operationsPath, attempts = 90) {
let taskResponse = null
let task = null
let operationsResponse = null
let operations = null
for (let attempt = 1; attempt <= attempts; attempt += 1) {
taskResponse = await context.request('GET', taskDetailPath)
task = taskResponse?.http === 200 && String(taskResponse?.body?.code) === '200'
? taskResponse.body.data : null
if (task?.taskStatus === 'COMPLETED' || task?.taskStatus === 'COMPLETION_FAILED') {
operationsResponse = await context.request('GET', operationsPath)
operations = operationsResponse?.http === 200 && String(operationsResponse?.body?.code) === '200'
? operationsResponse.body.data : null
return { taskResponse, task, operationsResponse, operations, attempts: attempt }
}
if (attempt < attempts) await new Promise((resolve) => setTimeout(resolve, 1_000))
}
return { taskResponse, task, operationsResponse, operations, attempts }
}
export async function run(context) {
const fixture = context.createdResources?.phase3Loading
const tenantId = id(fixture?.tenantId); const inboundId = id(fixture?.inboundId)
@@ -147,15 +167,20 @@ export async function run(context) {
const conflictedCompleteResponse = conflictedCompleteIndexes.length === 1
? concurrentCompleteResponses[conflictedCompleteIndexes[0]] : null
const winningCompleteIndex = successfulCompleteIndexes[0] ?? -1
const completeResponse = winningCompleteIndex >= 0
const acceptedCompleteResponse = winningCompleteIndex >= 0
? concurrentCompleteResponses[winningCompleteIndex] : concurrentCompleteResponses[0]
const completeBody = winningCompleteIndex >= 0
? completeAttempts[winningCompleteIndex].body : completeAttempts[0].body
const completeKey = winningCompleteIndex >= 0
? completeAttempts[winningCompleteIndex].key : completeAttempts[0].key
const operationsAfterConcurrentResponse = await context.request('GET', `${taskPath}/operations`)
const operationsAfterConcurrent = context.apiSuccess(MODULE,
'并发完成后回查 PDA 操作审计', operationsAfterConcurrentResponse, '', OPTIONS)
const completion = winningCompleteIndex >= 0
? await waitForCompletion(context, taskDetailPath, `${taskPath}/operations`)
: { taskResponse: acceptedCompleteResponse, task: null, operationsResponse: null, operations: null, attempts: 0 }
context.assert(MODULE, '异步完成命令收敛到终态', completion.taskResponse,
completion.task?.taskStatus === 'COMPLETED',
`taskStatus=${completion.task?.taskStatus || 'missing'}, workflow=${completion.task?.workflowStatus || 'missing'}, attempts=${completion.attempts}`, OPTIONS)
const operationsAfterConcurrentResponse = completion.operationsResponse
const operationsAfterConcurrent = completion.operations
const baselineCompleteOperationIds = new Set((operationsBefore || [])
.filter((item) => item?.actionType === 'COMPLETE_LOADING').map((item) => String(item?.operationId || '')))
const concurrentCompleteOperations = (operationsAfterConcurrent || []).filter((item) =>
@@ -170,10 +195,16 @@ export async function run(context) {
&& concurrentCompleteOperations.length === 1
&& concurrentCompleteOperations[0]?.operationId === completeBody.operationId
&& id(concurrentCompleteOperations[0]?.operatorId) && Boolean(concurrentCompleteOperations[0]?.createdAt)
&& completion.task?.taskStatus === 'COMPLETED' && completion.task?.workflowStatus === 'COMPLETED'
const completeResponse = completion.task?.taskStatus === 'COMPLETED'
? await request(context, 'POST', `${taskPath}/complete`, completeBody,
{ 'Idempotency-Key': completeKey })
: acceptedCompleteResponse
context.assert(MODULE, '两个不同幂等键并发 complete 仅一成功且只写一条完成审计',
completeResponse, concurrentCompleteVerified,
`results=${concurrentCompleteResponses.map((response) => `${response.http}/${codeOf(response)}`).join(',')}, success=${successfulCompleteIndexes.length}, conflict=${conflictedCompleteIndexes.length}, completeAudits=${concurrentCompleteOperations.length}, winner=${completeBody.operationId}`, OPTIONS)
if (completeResponse.http !== 200 || codeOf(completeResponse) !== '200') {
if (completeResponse.http !== 200 || codeOf(completeResponse) !== '200'
|| completion.task?.taskStatus !== 'COMPLETED') {
context.addStep(MODULE, '结算递增 inboundVersion 后 final complete 必须使用最新版本', completeResponse, false,
`expected=success after settlement refresh; HTTP=${completeResponse.http}, code=${codeOf(completeResponse)}, message=${completeResponse.body?.message || 'missing'}, taskId=${taskId}, inboundId=${inboundId}, taskVersion=${taskBefore?.version ?? 'missing'}, inboundVersion=${version}, requestId=${completeResponse.requestId || 'missing'}, traceId=${completeResponse.traceId || 'missing'}`, OPTIONS)
const taskAfterFailureResponse = await context.request('GET', taskDetailPath)