From 261c720c77135320d5c77c8565d4275f4ce2aa4d Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 5 Oct 2026 16:35:22 +0800 Subject: [PATCH] fix(firstleg): read concurrent customs initialization with current row lock --- .../service/ContainerCustomsService.java | 12 +++- .../ContainerCustomsMapperContractTest.java | 10 ++++ .../ContainerCustomsServiceTest.java | 59 ++++++++++++++++++- 3 files changed, 77 insertions(+), 4 deletions(-) diff --git a/firstleg/src/main/java/com/baiyue/firstleg/customs/container/service/ContainerCustomsService.java b/firstleg/src/main/java/com/baiyue/firstleg/customs/container/service/ContainerCustomsService.java index f05f4adeb..faa94dbaf 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/customs/container/service/ContainerCustomsService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/customs/container/service/ContainerCustomsService.java @@ -563,7 +563,12 @@ public class ContainerCustomsService { if (!"COMPLETED".equals(plan.loadingPlanStatus())) throw conflict("装柜完成后才能处理报关资料"); boolean refund = rows.stream().anyMatch(row -> "TAX_REFUND".equals(row.taxMode())); mapper.insertContainer(IdWorker.getId(), planId, plan.containerBookingId(), refund ? WAITING : READY); - return refresh(planId); + // A concurrent INSERT IGNORE may wait for another request's insert to commit. The + // transaction's earlier plan reads can retain an RR snapshot that cannot see that row. + // Read the winning aggregate under its row lock, rather than reusing a snapshot read. + ContainerRow container = mapper.lockContainer(planId); + if (container == null) throw conflict("整柜报关记录初始化失败"); + return refresh(planId, container); } private ContainerRow ensureMutable(long planId) { @@ -573,7 +578,10 @@ public class ContainerCustomsService { } private ContainerRow refresh(long planId) { - ContainerRow value = mapper.container(planId); + return refresh(planId, mapper.container(planId)); + } + + private ContainerRow refresh(long planId, ContainerRow value) { if (value == null || !Set.of(WAITING, READY).contains(value.status())) return value; String status = mapper.missingRefundCount(planId, value.id()) == 0 ? READY : WAITING; if (!status.equals(value.status())) { diff --git a/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsMapperContractTest.java b/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsMapperContractTest.java index 1e1a59fab..a8f005c3b 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsMapperContractTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsMapperContractTest.java @@ -11,6 +11,16 @@ import java.util.List; import static org.assertj.core.api.Assertions.assertThat; class ContainerCustomsMapperContractTest { + @Test void initializationFlushesSessionCacheBeforeCurrentAggregateRead() { + var configuration = new org.apache.ibatis.session.Configuration(); + configuration.addMapper(ContainerCustomsMapper.class); + String namespace = ContainerCustomsMapper.class.getName() + "."; + assertThat(configuration.getMappedStatement(namespace + "insertContainer").isFlushCacheRequired()).isTrue(); + var currentRead = configuration.getMappedStatement(namespace + "lockContainer"); + assertThat(currentRead.getBoundSql(java.util.Map.of("planId",9L)).getSql()).contains("FOR UPDATE"); + assertThat(currentRead.getCache()).isNull(); + } + @Test void merchantOrderUsesCompletedTaxRefundPlanWithRealCompletionTimestamp() throws Exception { String value=sql("completedPlanForOrder",long.class,long.class); assertThat(value) diff --git a/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsServiceTest.java index 153637d77..d017e50fa 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsServiceTest.java @@ -52,6 +52,7 @@ class ContainerCustomsServiceTest { when(mapper.planInbounds(9L)).thenReturn(List.of()); when(mapper.plan(9L)).thenReturn(plan); when(mapper.container(9L)).thenReturn(container(ContainerCustomsService.READY,0)); + when(mapper.lockContainer(9L)).thenReturn(container(ContainerCustomsService.READY,0)); var view=service.platformView(9L); @@ -78,11 +79,12 @@ class ContainerCustomsServiceTest { when(mapper.planInbounds(9L)).thenReturn(List.of()); when(mapper.plan(9L)).thenReturn(new ContainerCustomsMapper.PlanRow(9L,"LP9",8L,"COMPLETED","CAB","C")); when(mapper.container(9L)).thenReturn(container(ContainerCustomsService.READY,0)); + when(mapper.lockContainer(9L)).thenReturn(container(ContainerCustomsService.READY,0)); assertThatThrownBy(()->service.start(9L,5L,"admin","empty-plan",new ActionRequest(0,true,null))) .isInstanceOf(BizException.class).hasMessageContaining("没有实际货物，无需报关"); - verify(mapper,never()).lockContainer(9L); + verify(mapper).lockContainer(9L); verify(mapper,never()).start(anyLong(),anyInt(),anyLong(),anyString()); } @@ -91,6 +93,7 @@ class ContainerCustomsServiceTest { when(mapper.completedPlanForOrder(7L,21L)).thenReturn(9L); when(mapper.planInbounds(9L)).thenReturn(List.of(row)); when(mapper.container(9L)).thenReturn(container(ContainerCustomsService.READY,2)); + when(mapper.lockContainer(9L)).thenReturn(container(ContainerCustomsService.READY,2)); when(mapper.actualBoxCount(9L,7L,"TAX_REFUND")).thenReturn(3L); var view=service.merchantOrderView(7L,21L); @@ -101,6 +104,56 @@ class ContainerCustomsServiceTest { assertThat(view.inbounds()).extracting(value->value.orderId()).containsExactly(21L); } + @ParameterizedTest + @CsvSource({"WAITING_CUSTOMER_DOCUMENTS", "DECLARING", "COMPLETED"}) + void concurrentFirstReadUsesCommittedWinnerInsteadOfOldSnapshot(String status) throws Exception { + var own = new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",11L,21L,"O","I",7L,"客户","TAX_REFUND"); + var other = new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",12L,22L,"OTHER","OTHER-I",8L,"其他客户","TAX_REFUND"); + when(mapper.completedPlanForOrder(7L,21L)).thenReturn(9L); + when(mapper.planInbounds(9L)).thenReturn(List.of(own, other)); + // RR snapshot predates another request's committed insert; INSERT IGNORE lost that race. + when(mapper.insertContainer(anyLong(),eq(9L),eq(8L),anyString())).thenReturn(0); + when(mapper.container(9L)).thenReturn(null); + when(mapper.lockContainer(9L)).thenReturn(container(status, 3)); + when(mapper.missingRefundCount(9L,2L)).thenReturn(2); + when(mapper.actualBoxCount(9L,7L,"TAX_REFUND")).thenReturn(100L); + + var pool = Executors.newFixedThreadPool(4); + try { + var reads = java.util.stream.IntStream.range(0,4) + .mapToObj(index -> pool.submit(() -> service.merchantOrderView(7L,21L))).toList(); + for (var read : reads) { + var view = read.get(5,TimeUnit.SECONDS); + assertThat(view.status()).isEqualTo(status); + assertThat(view.version()).isEqualTo(3); + assertThat(view.actualBoxCount()).isEqualTo(100); + assertThat(view.inbounds()).extracting(value->value.orderId()).containsExactly(21L); + } + } finally { + pool.shutdownNow(); + } + verify(mapper,times(4)).lockContainer(9L); + verify(mapper,never()).container(9L); + verify(mapper,never()).refreshStatus(anyLong(),anyInt(),anyString()); + } + + @Test void currentAggregateStillRefreshesCoverageStatusAfterInitialization() { + var row = new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",11L,21L,"O","I",7L,"客户","TAX_REFUND"); + when(mapper.completedPlanForOrder(7L,21L)).thenReturn(9L); + when(mapper.planInbounds(9L)).thenReturn(List.of(row)); + when(mapper.lockContainer(9L)).thenReturn(container(ContainerCustomsService.WAITING,3)); + when(mapper.missingRefundCount(9L,2L)).thenReturn(0); + when(mapper.refreshStatus(2L,3,ContainerCustomsService.READY)).thenReturn(1); + // This transaction's own update is visible to its subsequent read. + when(mapper.container(9L)).thenReturn(container(ContainerCustomsService.READY,4)); + + var view = service.merchantOrderView(7L,21L); + + assertThat(view.status()).isEqualTo(ContainerCustomsService.READY); + assertThat(view.version()).isEqualTo(4); + verify(mapper).refreshStatus(2L,3,ContainerCustomsService.READY); + } + @Test void uploadedSourceViewReturnsPersistedNonNullVersion(){ var inbound=new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",11L,21L,"O","I",7L,"客户","TAX_REFUND"); var aggregate=container(ContainerCustomsService.READY,2); @@ -174,6 +227,7 @@ class ContainerCustomsServiceTest { when(mapper.planInbounds(9L)).thenReturn(List.of(inbound)); when(mapper.plan(9L)).thenReturn(new ContainerCustomsMapper.PlanRow(9L,"LP9",8L,"COMPLETED","CAB","C")); when(mapper.container(9L)).thenReturn(declaring); + when(mapper.lockContainer(9L)).thenReturn(declaring); when(mapper.actualBoxCount(9L,null,null)).thenReturn(0L); when(mapper.finalDocument(31L)).thenReturn(null); when(mapper.completion(31L)).thenReturn(null); @@ -215,6 +269,7 @@ class ContainerCustomsServiceTest { var file=new MockMultipartFile("file","source.bin","application/octet-stream",new byte[]{1}); when(mapper.planInbounds(9L)).thenReturn(List.of(inbound)); when(mapper.container(9L)).thenReturn(declaring); + when(mapper.lockContainer(9L)).thenReturn(declaring); when(mapper.document(1L)).thenReturn(source); when(mapper.lockContainerById(2L)).thenReturn(declaring); @@ -425,7 +480,7 @@ class ContainerCustomsServiceTest { var completed=container(ContainerCustomsService.COMPLETED,4); var buy=new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",11L,21L,"O","I",7L,"客户","BUY_DECLARATION"); when(mapper.claimCommand(anyLong(),eq(5L),eq("COMPLETE_ALL_BUY_CUSTOMS_9"),eq("all-buy-key"),anyString())).thenReturn(1); - when(mapper.lockContainer(9L)).thenReturn(declaring); + when(mapper.lockContainer(9L)).thenReturn(declaring,completed); when(mapper.planInbounds(9L)).thenReturn(List.of(buy)); when(mapper.complete(2L,3,5L,"admin")).thenReturn(1); when(mapper.completeCommand(5L,"COMPLETE_ALL_BUY_CUSTOMS_9","all-buy-key",2L)).thenReturn(1);