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 b572fe868..140fed70c 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 @@ -149,12 +149,12 @@ public class ContainerCustomsService { @Transactional public ContainerView start(long planId, long operatorId, String operatorName, String key, ActionRequest request) { - ContainerRow current = ensure(planId); String action = "START_CONTAINER_CUSTOMS_" + planId; if(!request.confirm())throw bad("请确认开始整柜报关"); String requestHash = hash(action, request.expectedVersion(), request.confirm(), request.remark()); CommandRow replay = claim(operatorId, action, key, requestHash); if (replay != null) return platformView(planId); + ensure(planId); ContainerRow locked = mapper.lockContainer(planId); if (locked == null || !READY.equals(locked.status())) throw conflict("当前柜报关资料尚未齐全或已开始报关"); if (!Objects.equals(locked.version(), request.expectedVersion())) throw concurrent(); 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 b7e74f912..33976049a 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 @@ -13,6 +13,10 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.mock.web.MockMultipartFile; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; @@ -104,6 +108,70 @@ class ContainerCustomsServiceTest { verify(mapper,never()).freezeScopes(anyLong()); } + @Test void concurrentStartWithSameKeyReplaysOneAggregateAndDifferentPayloadConflicts() throws Exception { + var row=new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",11L,21L,"O","I",7L,"客户","TAX_REFUND"); + var ready=container(ContainerCustomsService.READY,2); + var declaring=container(ContainerCustomsService.DECLARING,3); + var aggregateState=new AtomicReference<>(ready); + var commandState=new AtomicReference(); + var claimCalls=new AtomicInteger(); + var secondClaimEntered=new CountDownLatch(1); + var commandCompleted=new CountDownLatch(1); + when(mapper.planInbounds(9L)).thenReturn(List.of(row)); + when(mapper.container(9L)).thenAnswer(invocation->aggregateState.get()); + when(mapper.missingRefundCount(9L,2L)).thenReturn(0); + when(mapper.claimCommand(anyLong(),eq(5L),eq("START_CONTAINER_CUSTOMS_9"),eq("start-key"),anyString())) + .thenAnswer(invocation->{ + int call=claimCalls.incrementAndGet(); + if(call==1){ + commandState.set(new ContainerCustomsMapper.CommandRow(invocation.getArgument(0),5L, + "START_CONTAINER_CUSTOMS_9","start-key",invocation.getArgument(4),"PROCESSING",null,null,null)); + assertThat(secondClaimEntered.await(5,TimeUnit.SECONDS)).isTrue(); + return 1; + } + secondClaimEntered.countDown(); + assertThat(commandCompleted.await(5,TimeUnit.SECONDS)).isTrue(); + return 0; + }); + when(mapper.command(5L,"START_CONTAINER_CUSTOMS_9","start-key")).thenAnswer(invocation->commandState.get()); + when(mapper.lockContainer(9L)).thenAnswer(invocation->aggregateState.get()); + when(mapper.start(2L,2,5L,"admin")).thenAnswer(invocation->{aggregateState.set(declaring);return 1;}); + when(mapper.completeCommand(eq(5L),eq("START_CONTAINER_CUSTOMS_9"),eq("start-key"),eq(2L))) + .thenAnswer(invocation->{ + var current=commandState.get(); + commandState.set(new ContainerCustomsMapper.CommandRow(current.id(),current.actorId(),current.action(), + current.idempotencyKey(),current.requestHash(),"SUCCESS",2L,null,null)); + commandCompleted.countDown(); + return 1; + }); + when(mapper.documents(2L,null)).thenReturn(List.of()); + + var request=new ActionRequest(2,true,"same"); + var executor=Executors.newFixedThreadPool(2); + try{ + var first=executor.submit(()->service.start(9L,5L,"admin","start-key",request)); + var second=executor.submit(()->service.start(9L,5L,"admin","start-key",request)); + var firstResult=first.get(8,TimeUnit.SECONDS); + var secondResult=second.get(8,TimeUnit.SECONDS); + + assertThat(firstResult.aggregateId()).isEqualTo(2L); + assertThat(secondResult.aggregateId()).isEqualTo(2L); + assertThat(firstResult.status()).isEqualTo(ContainerCustomsService.DECLARING); + assertThat(secondResult.status()).isEqualTo(ContainerCustomsService.DECLARING); + }finally{ + executor.shutdownNow(); + } + + assertThatThrownBy(()->service.start(9L,5L,"admin","start-key",new ActionRequest(2,true,"different"))) + .isInstanceOfSatisfying(BizException.class, + error->assertThat(error.getResultCode()).isEqualTo(ResultCode.CONFLICT)); + InOrder order=inOrder(mapper); + order.verify(mapper,times(2)).claimCommand(anyLong(),eq(5L),eq("START_CONTAINER_CUSTOMS_9"),eq("start-key"),anyString()); + order.verify(mapper,atLeastOnce()).planInbounds(9L); + verify(mapper,times(1)).start(2L,2,5L,"admin"); + verify(mapper,times(1)).completeCommand(5L,"START_CONTAINER_CUSTOMS_9","start-key",2L); + } + @Test void lastCompletionRequiresConfirmationBeforeCallingFile(){ var source=document(1L);var container=container(ContainerCustomsService.DECLARING,3); when(mapper.claimCommand(anyLong(),eq(5L),eq("UPLOAD_COMPLETION_1"),eq("k"),anyString())).thenReturn(1);