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

fix(firstleg): read concurrent customs initialization with current row lock

Browse Source
This commit is contained in:
ouyangxq
2026-10-05 16:35:22 +08:00
parent 090f57bfb1
commit 261c720c77
3 changed files with 77 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
firstleg/src/main/java/com/baiyue/firstleg/customs/container/service/ContainerCustomsService.java
+10 -2
View File
@@ -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())) {
firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsMapperContractTest.java
+10
View File
@@ -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)
firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsServiceTest.java
+57 -2
View File
@@ -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);