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

fix(outbound): atomically confirm platform mixed-container dispatch and tracking

Browse Source
This commit is contained in:
ouyangxq
2026-09-30 12:37:00 +08:00
parent 5cb2a5145f
commit 125ca34ba4
5 changed files with 109 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/sql/头程数据库统一增量-MySQL8.sql
+1 -1
View File
@@ -5259,7 +5259,7 @@ CREATE TRIGGER trg_carrier_schedule_node_before_update BEFORE UPDATE ON fl_carri
CREATE TRIGGER trg_carrier_schedule_query_before_insert BEFORE INSERT ON fl_carrier_schedule_query FOR EACH ROW BEGIN
IF NOT EXISTS (SELECT 1 FROM fl_ocean_booking booking
WHERE booking.id=NEW.ocean_booking_id AND booking.customer_id=NEW.customer_id) THEN
WHERE booking.id=NEW.ocean_booking_id AND COALESCE(booking.customer_id,0)=NEW.customer_id) THEN
SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='船司查询客户与订舱不一致';
END IF;
IF NEW.container_booking_id IS NOT NULL AND NOT EXISTS (
firstleg/src/main/java/com/baiyue/firstleg/outbound/dispatch/service/OutboundDispatchAutoConfirmService.java
+5 -11
View File
@@ -1,10 +1,10 @@
package com.baiyue.firstleg.outbound.dispatch.service;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.firstleg.outbound.dispatch.domain.OutboundDispatchStatus;
import com.baiyue.firstleg.outbound.dispatch.dto.OutboundDispatchDtos.ActionRequest;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/** Real proxy-owned transaction boundary for repeated system convergence after customs completion. */
@@ -12,19 +12,13 @@ import org.springframework.transaction.annotation.Transactional;
public class OutboundDispatchAutoConfirmService {
@Resource private OutboundDispatchService dispatch;
@Transactional
@Transactional(propagation = Propagation.REQUIRES_NEW)
public boolean confirmIfReady(long containerBookingId) {
var current=dispatch.detail(containerBookingId);
if(OutboundDispatchStatus.WAITING_FOR_LOADING.name().equals(current.aggregate().dispatchStatus()))return true;
if(!current.aggregate().gate().passed())return false;
try{
dispatch.confirm(containerBookingId,0L,"SYSTEM","CUSTOMS-AUTO-DISPATCH-"+containerBookingId,
new ActionRequest(current.aggregate().version(),"整柜报关完成且全部离仓门禁满足，系统自动确认离仓"));
return true;
}catch(BizException ex){
var after=dispatch.detail(containerBookingId);
if(OutboundDispatchStatus.WAITING_FOR_LOADING.name().equals(after.aggregate().dispatchStatus()))return true;
throw ex;
}
dispatch.confirm(containerBookingId,0L,"SYSTEM","CUSTOMS-AUTO-DISPATCH-"+containerBookingId,
new ActionRequest(current.aggregate().version(),"整柜报关完成且全部离仓门禁满足，系统自动确认离仓"));
return true;
}
}
firstleg/src/main/resources/db/phase3-carrier-schedule-query-schema.sql
+1 -1
View File
@@ -209,7 +209,7 @@ DELIMITER $$
CREATE TRIGGER trg_carrier_schedule_query_before_insert BEFORE INSERT ON fl_carrier_schedule_query FOR EACH ROW
BEGIN
IF NOT EXISTS (SELECT 1 FROM fl_ocean_booking booking
WHERE booking.id=NEW.ocean_booking_id AND booking.customer_id=NEW.customer_id) THEN
WHERE booking.id=NEW.ocean_booking_id AND COALESCE(booking.customer_id,0)=NEW.customer_id) THEN
SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='船司查询客户与订舱不一致';
END IF;
IF NEW.container_booking_id IS NOT NULL AND NOT EXISTS (
firstleg/src/test/java/com/baiyue/firstleg/outbound/dispatch/OutboundDispatchAutoConfirmServiceTest.java
+71 -1
View File
@@ -2,15 +2,33 @@ package com.baiyue.firstleg.outbound.dispatch;
import com.baiyue.firstleg.outbound.dispatch.dto.OutboundDispatchDtos.*;
import com.baiyue.firstleg.outbound.dispatch.service.*;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.*;
import org.mockito.*;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.NameMatchTransactionAttributeSource;
import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.*;
class OutboundDispatchAutoConfirmServiceTest {
@@ -20,6 +38,58 @@ class OutboundDispatchAutoConfirmServiceTest {
@Test void waitsWithoutWritingUntilEveryOtherGatePasses(){when(dispatch.detail(22L)).thenReturn(detail(false,"READY_FOR_DISPATCH",3));assertThat(service.confirmIfReady(22L)).isFalse();verify(dispatch,never()).confirm(anyLong(),anyLong(),anyString(),anyString(),any());}
@Test void usesSystemIdentityAndStableIdempotencyKey(){when(dispatch.detail(22L)).thenReturn(detail(true,"READY_FOR_DISPATCH",3));assertThat(service.confirmIfReady(22L)).isTrue();verify(dispatch).confirm(eq(22L),eq(0L),eq("SYSTEM"),eq("CUSTOMS-AUTO-DISPATCH-22"),argThat(r->r.expectedVersion()==3));}
@Test void concurrentReplayAlreadyConfirmedConvergesWithoutSecondMutation(){when(dispatch.detail(22L)).thenReturn(detail(true,"WAITING_FOR_LOADING",4));assertThat(service.confirmIfReady(22L)).isTrue();verify(dispatch,never()).confirm(anyLong(),anyLong(),anyString(),anyString(),any());}
@Test void proxyOwnedAutoBoundaryAndConfirmMutationAreTransactional()throws Exception{assertThat(OutboundDispatchAutoConfirmService.class.getMethod("confirmIfReady",long.class).getAnnotation(Transactional.class)).isNotNull();assertThat(OutboundDispatchService.class.getMethod("confirm",long.class,long.class,String.class,String.class,ActionRequest.class).getAnnotation(Transactional.class)).isNotNull();}
@Test void proxyOwnedAutoBoundaryAndConfirmMutationAreTransactional()throws Exception{assertThat(OutboundDispatchAutoConfirmService.class.getMethod("confirmIfReady",long.class).getAnnotation(Transactional.class).propagation()).isEqualTo(Propagation.REQUIRES_NEW);assertThat(OutboundDispatchService.class.getMethod("confirm",long.class,long.class,String.class,String.class,ActionRequest.class).getAnnotation(Transactional.class)).isNotNull();}
@Test void anUncommittedDepartureCannotHideCarrierTaskCreationFailure(){
when(dispatch.detail(22L)).thenReturn(detail(true,"READY_FOR_DISPATCH",3), detail(true,"WAITING_FOR_LOADING",4));
BizException failure=new BizException(ResultCode.CONFLICT,"船司查询客户与订舱不一致");
when(dispatch.confirm(anyLong(),anyLong(),anyString(),anyString(),any())).thenThrow(failure);
assertThatThrownBy(()->service.confirmIfReady(22L)).isSameAs(failure);
verify(dispatch,times(1)).detail(22L);
}
@Test void afterCommitFailureRollsBackDepartureAndTaskTogetherThenIndependentRetryCommitsBoth(){
JdbcDataSource database=new JdbcDataSource();
database.setURL("jdbc:h2:mem:auto-dispatch-"+UUID.randomUUID()+";DB_CLOSE_DELAY=-1");
JdbcTemplate jdbc=new JdbcTemplate(database);
jdbc.execute("CREATE TABLE business_fact(kind VARCHAR(32) PRIMARY KEY)");
var manager=new DataSourceTransactionManager(database);
AtomicBoolean rejectCarrier=new AtomicBoolean(true);
when(dispatch.detail(22L)).thenAnswer(invocation->detail(true,
jdbc.queryForObject("SELECT COUNT(*) FROM business_fact WHERE kind='OUTBOUND'",Integer.class)>0
? "WAITING_FOR_LOADING" : "READY_FOR_DISPATCH",3));
when(dispatch.confirm(anyLong(),anyLong(),anyString(),anyString(),any())).thenAnswer(invocation->{
jdbc.update("INSERT INTO business_fact(kind) VALUES('OUTBOUND')");
if(rejectCarrier.get())throw new BizException(ResultCode.CONFLICT,"carrier task insert failed");
jdbc.update("INSERT INTO business_fact(kind) VALUES('CARRIER_QUERY')");
return detail(true,"WAITING_FOR_LOADING",4);
});
var dispatchAttributes=new NameMatchTransactionAttributeSource();
dispatchAttributes.addTransactionalMethod("confirm",new RuleBasedTransactionAttribute());
ProxyFactory dispatchFactory=new ProxyFactory(dispatch);
dispatchFactory.setProxyTargetClass(true);
dispatchFactory.addAdvice(new TransactionInterceptor(manager,dispatchAttributes));
ReflectionTestUtils.setField(service,"dispatch",dispatchFactory.getProxy());
ProxyFactory autoFactory=new ProxyFactory(service);
autoFactory.setProxyTargetClass(true);
autoFactory.addAdvice(new TransactionInterceptor(manager,new AnnotationTransactionAttributeSource()));
var auto=(OutboundDispatchAutoConfirmService)autoFactory.getProxy();
AtomicReference<RuntimeException> callbackFailure=new AtomicReference<>();
new TransactionTemplate(manager).executeWithoutResult(status->{
jdbc.update("INSERT INTO business_fact(kind) VALUES('CUSTOMS_COMPLETED')");
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization(){
@Override public void afterCommit(){
try{auto.confirmIfReady(22L);}catch(RuntimeException failure){callbackFailure.set(failure);}
}
});
});
assertThat(callbackFailure.get()).isInstanceOf(BizException.class).hasMessage("carrier task insert failed");
assertThat(jdbc.queryForList("SELECT kind FROM business_fact",String.class)).containsExactly("CUSTOMS_COMPLETED");
rejectCarrier.set(false);
assertThat(auto.confirmIfReady(22L)).isTrue();
assertThat(jdbc.queryForList("SELECT kind FROM business_fact",String.class))
.containsExactlyInAnyOrder("CUSTOMS_COMPLETED","OUTBOUND","CARRIER_QUERY");
assertThat(auto.confirmIfReady(22L)).isTrue();
verify(dispatch,times(2)).confirm(anyLong(),anyLong(),anyString(),anyString(),any());
}
private static Detail detail(boolean gate,String status,int version){var result=new ContainerGateResult(gate,List.of(),LocalDateTime.now());var aggregate=new Aggregate(22L,"CAB","C",44L,"COMPLETED",66L,"RELEASED",77L,"DEPARTED",status,version,List.of(),result);return new Detail(aggregate,List.of());}
}
firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQuerySqlContractTest.java
+31 -1
View File
@@ -4,6 +4,8 @@ import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleQueryMapper;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import org.junit.jupiter.api.Test;
import org.h2.jdbcx.JdbcDataSource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.core.io.ClassPathResource;
import java.nio.charset.StandardCharsets;
@@ -14,6 +16,34 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
class CarrierScheduleQuerySqlContractTest {
@Test
void deployedTriggerPredicateMapsPlatformBookingToScopeZeroAndStillRejectsOtherCustomers() throws Exception {
JdbcDataSource database = new JdbcDataSource();
database.setURL("jdbc:h2:mem:carrier-scope-" + java.util.UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
JdbcTemplate jdbc = new JdbcTemplate(database);
jdbc.execute("CREATE TABLE fl_ocean_booking(id BIGINT PRIMARY KEY,customer_id BIGINT)");
jdbc.update("INSERT INTO fl_ocean_booking(id,customer_id) VALUES(1,NULL),(2,10001),(3,10002)");
String unified = Files.readString(Path.of("..", "deploy", "sql", "头程数据库统一增量-MySQL8.sql"),
StandardCharsets.UTF_8);
for (String source : List.of(moduleSql(), unified)) {
int triggerStart = source.indexOf("CREATE TRIGGER trg_carrier_schedule_query_before_insert");
int start = source.indexOf("WHERE booking.id=NEW.ocean_booking_id", triggerStart);
int end = source.indexOf(") THEN", start);
assertThat(triggerStart).isGreaterThanOrEqualTo(0);
assertThat(start).isGreaterThan(triggerStart);
assertThat(end).isGreaterThan(start);
String predicate = source.substring(start, end)
.replace("NEW.ocean_booking_id", "?").replace("NEW.customer_id", "?");
String query = "SELECT EXISTS(SELECT 1 FROM fl_ocean_booking booking " + predicate + ")";
assertThat(jdbc.queryForObject(query, Boolean.class, 1L, 0L)).isTrue();
assertThat(jdbc.queryForObject(query, Boolean.class, 1L, 10001L)).isFalse();
assertThat(jdbc.queryForObject(query, Boolean.class, 2L, 10001L)).isTrue();
assertThat(jdbc.queryForObject(query, Boolean.class, 2L, 0L)).isFalse();
assertThat(jdbc.queryForObject(query, Boolean.class, 2L, 10002L)).isFalse();
assertThat(jdbc.queryForObject(query, Boolean.class, 99L, 0L)).isFalse();
}
}
@Test
void batchesActualContainerFactsWithoutEstimatedOrPerNodeQueries() throws Exception {
String sql = String.join("\n", CarrierScheduleQueryMapper.class
@@ -95,7 +125,7 @@ class CarrierScheduleQuerySqlContractTest {
@Test
void enforcesRealBookingContainerAndPermissions() throws Exception {
assertThat(moduleSql())
.contains("booking.id=NEW.ocean_booking_id AND booking.customer_id=NEW.customer_id")
.contains("booking.id=NEW.ocean_booking_id AND COALESCE(booking.customer_id,0)=NEW.customer_id")
.contains("allocation.container_booking_id=NEW.container_booking_id")
.contains("id IN (5102184,5102186,5102188)")
.contains("platform:firstleg:port-schedule-query:manual")