fix(lastleg): read committed order command replays after lock waits

2026-10-03 12:13:34 +08:00
parent 34cac9c42a
commit eadb431f33
6 changed files with 100 additions and 8 deletions
@@ -62,7 +62,7 @@

 普通库存换标回归由 `phase4-overseas-pda-operations` 验证：正常箱先完成混合换标的箱标与 SKU 阶段，另两箱保留跳过/异常；首次显式异常完工后经平台 `/rework` 退回并验证幂等，返工重新完成全部箱标/SKU 事实，再生成本轮目标托盘、绑定及核对库存流水，最后正常完工和客户确认。PDA 不自行批准复核，不在客户确认后才首次组托；异常首次完工会创建关联异常记录，属于测试数据副作用，保留为验收证据。

-尾程暂扣按真实入库阶段验收：`phase4-tail-order-controls` 在 PDA 入库前验证暂扣返回冲突并回查订单零漂移，继续人工节点及整柜计划转换；`phase4-overseas-pda-operations` 接收本轮完成库存后调用同模块的 `verifyReceivedOrderHold`，重新核对本轮客户、订单和 `INBOUND_COMPLETED` 状态，再执行暂扣、同键重放/异参冲突、解除。正向暂扣未执行或前置不足仍是失败，不以入库前拒绝代替正向通过。
+尾程暂扣按真实入库阶段验收：`phase4-tail-order-controls` 在 PDA 入库前验证暂扣返回冲突并回查订单零漂移，继续人工节点及整柜计划转换；`phase4-overseas-pda-operations` 接收本轮完成库存后调用同模块的 `verifyReceivedOrderHold`，重新核对本轮客户、订单和 `INBOUND_COMPLETED` 状态，再执行暂扣、同键重放/异参冲突、解除。第一轮暂扣和解除分别同时发送两条相同幂等键、相同载荷的命令，全部返回后核对响应相同、版本只增加一次和原始库存桶；第二轮继续验证可重复周期与独立冻结保留。并发限制为两条同命令，不并发创建资源；响应不明时等待两请求结束并保留样本，不盲目释放。该证据不替代暂扣与 PDA 拣货竞争同一库存的测试。正向暂扣未执行或前置不足仍是失败，不以入库前拒绝代替正向通过。

 `pda-loading-link-regression.mjs` 通过总入口 `--modules pda-loading-link-regression` 运行，复用 `dropship-pallet-matrix.mjs` 的国内正式订单、测量、封托、订舱、证据和完成生产链，仅执行国内关联回归。覆盖单票加入后全移除换柜完成、多票共托反复换三个旧柜后完成、逐托即时提交及同键重放，以及一单多柜中仍有实际分配的未开始柜导致 `OWNER_READY/A0409`。每次重新创建带 marker 的业务数据，通过公开 HTTP 回查旧柜有效分配数、PDA 终态和入仓状态；不写业务数据库伪造前置。零分配残留由真实整托移除链观察，报告保存 `createdResources.pdaLoadingLinkRegression`。负向柜保留为 `COMPLETION_FAILED`，成功柜保留为 `COMPLETED`，不自动清理；失败报告保留且不得算通过。使用显式测试环境文件及 `--output reports/<本轮名称>.json`，须有真实 HTTP 写入授权；不启动海外收货或出库。

@@ -178,6 +178,12 @@ test('fresh immutable IDs drive two full cycles and preserve independent HOLD ac
  await verifyReceivedOrderHold(context)
  const proof=context.createdResources.phase4TailOrderControls.holdInventoryProof
  assert.equal(proof.status,'VERIFIED');assert.equal(proof.restored,true);assert.equal(order.version,8)
+  assert.equal(proof.concurrentCommands.length,2)
+  for(const command of proof.concurrentCommands) {
+    assert.equal(command.copies,2)
+    assert.deepEqual(command.replies[0],command.replies[1])
+    assert.equal(command.replies[0].version,command.expectedVersion+1)
+  }
  assert.equal(proof.snapshots.C2_RELEASED.positions.find(row=>row.id==='22').holdPackages,1)
  assert.equal(proof.snapshots.AFTER.balances[0].availablePackages,4)
  assert.equal(calls.filter(c=>c.path.endsWith('/manual')).length,2)
@@ -185,6 +191,41 @@ test('fresh immutable IDs drive two full cycles and preserve independent HOLD ac
  assert.equal(calls.filter(c=>c.method==='POST'&&c.path.endsWith('/inventory/release')).length,1)
 })

+test('hold duplicate requests are in flight together, with bounded concurrency of two',async()=>{
+  const {context}=holdModel()
+  const originalRequest=context.request
+  let active=0,peak=0
+  context.request=async(method,path,...args)=>{
+    if(method!=='PUT'||!path.includes('/orders/')) return originalRequest(method,path,...args)
+    active++;peak=Math.max(peak,active)
+    try { await new Promise(resolve=>setImmediate(resolve));return await originalRequest(method,path,...args) }
+    finally {active--}
+  }
+  await verifyReceivedOrderHold(context)
+  assert.equal(peak,2)
+  assert.equal(active,0)
+})
+
+test('ambiguous parallel hold waits for its sibling and never starts recovery writes',async()=>{
+  const {context,calls}=holdModel()
+  const originalRequest=context.request
+  let holdRequests=0,siblingSettled=false
+  context.request=async(method,path,body,...args)=>{
+    if(method==='PUT' && body?.held && ++holdRequests<=2) {
+      if(holdRequests===1) throw new Error('lost response')
+      await new Promise(resolve=>setImmediate(resolve))
+      const response=await originalRequest(method,path,body,...args)
+      siblingSettled=true
+      return response
+    }
+    return originalRequest(method,path,body,...args)
+  }
+  await assert.rejects(verifyReceivedOrderHold(context),/lost response/)
+  assert.equal(siblingSettled,true)
+  assert.equal(context.createdResources.phase4TailOrderControls.holdInventoryProof.uncertainWrite,true)
+  assert.ok(!calls.some(call=>call.path.endsWith('/inventory/release')||call.body?.held===false))
+})
+
 for(const fault of ['stale marker','no evidence','wrong immutable customer','missing position','unsafe bucket','not enough stock','moved source']) {
  test(`missing or unsafe fresh hold prerequisite blocks all writes: ${fault}`,async()=>{
    const {context,calls}=holdModel(fault)
@@ -372,7 +372,7 @@ export async function verifyReceivedOrderHold(context) {
    && originalBalances.every(row=>id(row.customerId)===customerId && id(row.overseasWarehouseId)===warehouseId)
    && originalPositions.every(row=>id(row.customerId)===customerId && id(row.overseasWarehouseId)===warehouseId))
  const proof = owned.holdInventoryProof = { marker:context.marker, orderId, customerId, warehouseId,
-    balanceIds, positionIds, snapshots:{}, status:'RUNNING', restored:false }
+    balanceIds, positionIds, snapshots:{}, concurrentCommands:[], status:'RUNNING', restored:false }
  const orderPath = query(`${ORDERS}/${encodeURIComponent(orderId)}/hold`, {customerId})
  const inventoryPath = '/api/platform/v1/last-leg/inventory'
  let uncertain = false, otherApplied = false, active = null, baseline, otherHeld, otherBody
@@ -419,14 +419,24 @@ export async function verifyReceivedOrderHold(context) {
  }
  const stock = value => ({balances:value.balances.map(holdStableStock),positions:value.positions.map(holdStableStock)})
  const expectStock = (actual,expected,label) => require(label,sameHoldFacts(stock(actual),stock(expected)))
-  const orderCommand = async (held,version,suffix) => {
+  const orderCommand = async (held,version,suffix,copies=1) => {
    const body={held,chargeable:held,reason:`${context.marker} ${suffix}`,expectedVersion:Number(version)}
    const headers={'Idempotency-Key':`${attemptKey}-${suffix}`}
-    const data=await request('PUT',held?'暂扣本轮单票尾程订单':'解除本轮单票尾程暂扣',orderPath,body,headers)
+    // Settle both writes before inspecting or recovering: one rejected promise does not
+    // mean its sibling has stopped modifying the order.
+    const outcomes=await Promise.allSettled(Array.from({length:copies},()=>
+      request('PUT',held?'暂扣本轮单票尾程订单':'解除本轮单票尾程暂扣',orderPath,body,headers)))
+    const failed=outcomes.find(result=>result.status==='rejected')
+    if(failed) { if(copies>1) uncertain=true;throw failed.reason }
+    const replies=outcomes.map(result=>result.value),data=replies[0]
    // A malformed successful response is not sufficient authority for a recovery mutation.
    if(id(data?.orderId)!==orderId || data.held!==held || Number(data.version)!==Number(version)+1) uncertain=true
    require('暂扣命令返回本订单预期版本及计费状态',!uncertain && data.chargeable===held && (held || data.reason==null))
    active=held?{version:Number(data.version),reason:body.reason}:null
+    if(copies>1) {
+      proof.concurrentCommands.push({held,copies,expectedVersion:Number(version),replies})
+      require('并发同键命令返回相同订单状态和唯一新版本',replies.every(reply=>sameHoldFacts(reply,data)))
+    }
    return {body,headers,data}
  }
  const releaseOther = async () => {
@@ -461,7 +471,7 @@ export async function verifyReceivedOrderHold(context) {
    const allHeld=shiftHoldStock(otherHeld,null,true)
    for(let cycle=1;cycle<=2;cycle++) {
      const before=await read(`C${cycle}_BEFORE`)
-      const held=await orderCommand(true,before.order.version,`F32-C${cycle}-HOLD`)
+      const held=await orderCommand(true,before.order.version,`F32-C${cycle}-HOLD`,cycle===1?2:1)
      const during=await read(`C${cycle}_HELD`)
      require('暂扣只推进单票版本并保留计费原因',during.order.detained===true && Number(during.order.version)===Number(held.data.version)
        && during.order.holdReason===held.body.reason && during.order.heldChargeable===true)
@@ -473,7 +483,7 @@ export async function verifyReceivedOrderHold(context) {
      await reject('暂扣尾程订单不能生成人工计划',`${PLANS}/manual`,manualPlanBody(representative,held.data.version,context.marker),
        {'Idempotency-Key':`${attemptKey}-F32-C${cycle}-PLAN`},'POST')
      require('人工计划拒绝后暂扣与库存不变',sameHoldFacts(await read(`C${cycle}_PLAN_REJECTED`),during))
-      const released=await orderCommand(false,during.order.version,`F32-C${cycle}-RELEASE`)
+      const released=await orderCommand(false,during.order.version,`F32-C${cycle}-RELEASE`,cycle===1?2:1)
      const after=await read(`C${cycle}_RELEASED`)
      require('解除暂扣清空计费与原因并推进版本',after.order.detained===false && after.order.heldChargeable===false
        && after.order.holdReason==null && Number(after.order.version)===Number(released.data.version))