diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderCommandService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderCommandService.java index 8fc8c86f4..9afa2784c 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderCommandService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderCommandService.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import jakarta.annotation.Resource; import org.springframework.http.HttpStatus; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Isolation; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.server.ResponseStatusException; @@ -55,7 +56,9 @@ public class LastlegOrderCommandService { @Resource private InventoryCommandService inventory; @Resource private ObjectMapper objectMapper; - @Transactional + // Ownership is read before waiting for the order lock. Under MySQL repeatable + // read that snapshot can hide the preceding writer's idempotency record. + @Transactional(isolation = Isolation.READ_COMMITTED) public LastlegOrderCommandDtos.HoldResponse hold(String requestedCustomerId, long orderId, String idempotencyKey, LastlegOrderCommandDtos.HoldRequest request, @@ -122,7 +125,7 @@ public class LastlegOrderCommandService { return response; } - @Transactional + @Transactional(isolation = Isolation.READ_COMMITTED) public LastlegOrderCommandDtos.ManualNodesResponse saveManualNodes(String requestedCustomerId, long orderId, String idempotencyKey, LastlegOrderCommandDtos.ManualNodesRequest request, diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inventory/command/InventoryHoldConcurrencyPersistenceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inventory/command/InventoryHoldConcurrencyPersistenceTest.java index 217dbe046..4da37378b 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inventory/command/InventoryHoldConcurrencyPersistenceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inventory/command/InventoryHoldConcurrencyPersistenceTest.java @@ -137,6 +137,32 @@ class InventoryHoldConcurrencyPersistenceTest { assertEquals(12,amount("available_packages")+amount("hold_packages")+amount("allocated_packages")+amount("picked_packages")+shipped); assertEquals(shipped,jdbc.queryForObject("SELECT COALESCE(-SUM(delta_packages),0) FROM ll_inventory_ledger WHERE business_type='SHIP'",Long.class)); } + @Test void simultaneousDuplicateHoldAndReleaseAdvanceOnceAndKeepIndependentHold() throws Exception { + inventory.hold("C1","operator",move(22,2,"OTHER-HOLD")); + ExecutorService pool=Executors.newFixedThreadPool(2); + try { + for (boolean held : new boolean[]{true,false}) { + int version=held?0:1; + String key=held?"parallel-hold":"parallel-release"; + var ready=new CountDownLatch(2);var start=new CountDownLatch(1); + Callable command=()->{ + ready.countDown();await(start); + return orders.hold("C1",11,key,new LastlegOrderCommandDtos.HoldRequest(held,false,"quality",version),"operator","Operator"); + }; + var first=pool.submit(command);var second=pool.submit(command); + await(ready);start.countDown(); + var response=first.get(8,TimeUnit.SECONDS); + assertEquals(response,second.get(8,TimeUnit.SECONDS)); + assertEquals(version+1,response.version()); + assertEquals(version+1,orderMapper.selectById(11L).getVersion()); + } + assertEquals(2,count("ll_order_command_record")); + assertEquals(2,count("ll_order_hold_history")); + assertEquals(5,count("ll_inventory_operation")); + assertEquals(5,count("ll_inventory_ledger")); + assertEquals(10,amount("available_packages"));assertEquals(2,amount("hold_packages"));conserved(0); + } finally {pool.shutdownNow();assertTrue(pool.awaitTermination(10,TimeUnit.SECONDS));} + } @Test void secondHoldCycleMustMoveInventoryAgainAndNotReuseTheFirstCycle() { hold(true,0,"hold-one"); hold(false,1,"release-one"); hold(true,2,"hold-two"); assertEquals(0,amount("available_packages")); assertEquals(12,amount("hold_packages")); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderCommandServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderCommandServiceTest.java index 456183890..c917a4b98 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderCommandServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderCommandServiceTest.java @@ -66,6 +66,18 @@ class LastlegOrderCommandServiceTest { ReflectionTestUtils.setField(service, "objectMapper", JsonMapper.builder().findAndAddModules().build()); } + @Test + void commandTransactionsReadCommittedAfterWaitingForTheOrderLock() throws Exception { + for (var entry : List.of( + new Object[]{"hold", LastlegOrderCommandDtos.HoldRequest.class}, + new Object[]{"saveManualNodes", LastlegOrderCommandDtos.ManualNodesRequest.class})) { + var method=LastlegOrderCommandService.class.getMethod((String)entry[0],String.class,long.class, + String.class,(Class)entry[1],String.class,String.class); + assertEquals(org.springframework.transaction.annotation.Isolation.READ_COMMITTED, + method.getAnnotation(org.springframework.transaction.annotation.Transactional.class).isolation()); + } + } + @Test void holdsOnlySelectedOrderWithVersionAndImmutableHistory() { LastlegOrder order = order(11L, 3); diff --git a/scripts/admin-test-env-modules/README.md b/scripts/admin-test-env-modules/README.md index 0c865f61d..f0b50d22f 100644 --- a/scripts/admin-test-env-modules/README.md +++ b/scripts/admin-test-env-modules/README.md @@ -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 写入授权；不启动海外收货或出库。 diff --git a/scripts/admin-test-env-modules/phase4-tail-order-controls-contract.test.mjs b/scripts/admin-test-env-modules/phase4-tail-order-controls-contract.test.mjs index 6ae787023..bd552627d 100644 --- a/scripts/admin-test-env-modules/phase4-tail-order-controls-contract.test.mjs +++ b/scripts/admin-test-env-modules/phase4-tail-order-controls-contract.test.mjs @@ -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) diff --git a/scripts/admin-test-env-modules/phase4-tail-order-controls.mjs b/scripts/admin-test-env-modules/phase4-tail-order-controls.mjs index 133fa0f2b..c54e81d68 100644 --- a/scripts/admin-test-env-modules/phase4-tail-order-controls.mjs +++ b/scripts/admin-test-env-modules/phase4-tail-order-controls.mjs @@ -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))