From 49ac9222b5513ad0a5920ae0dffb38291c1e229c Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 5 Oct 2026 02:32:06 +0800 Subject: [PATCH] fix(shipping): replay immutable carrier facts without updates --- .../CarrierScheduleExceptionFactMapper.java | 3 +- .../mapper/CarrierScheduleNodeFactMapper.java | 5 +- .../CarrierScheduleQueryRepository.java | 30 ++- .../ImmutableCarrierFactPersistenceTest.java | 184 ++++++++++++++++++ ...rrierScheduleExceptionPersistenceTest.java | 23 +++ 5 files changed, 237 insertions(+), 8 deletions(-) create mode 100644 firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/ImmutableCarrierFactPersistenceTest.java diff --git a/firstleg/src/main/java/com/baiyue/firstleg/portschedule/mapper/CarrierScheduleExceptionFactMapper.java b/firstleg/src/main/java/com/baiyue/firstleg/portschedule/mapper/CarrierScheduleExceptionFactMapper.java index 6032850bb..108dc670c 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/portschedule/mapper/CarrierScheduleExceptionFactMapper.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/portschedule/mapper/CarrierScheduleExceptionFactMapper.java @@ -17,7 +17,6 @@ public interface CarrierScheduleExceptionFactMapper extends BaseMapper selectByContainer(@Param("containerBookingId") long containerBookingId); - @Select("SELECT * FROM fl_carrier_schedule_node_fact WHERE container_booking_id=#{containerBookingId} AND event_key=#{eventKey} LIMIT 1") + @Select("SELECT * FROM fl_carrier_schedule_node_fact WHERE container_booking_id=#{containerBookingId} AND event_key=#{eventKey} LIMIT 1 FOR UPDATE") CarrierScheduleNodeFactEntity selectByContainerEvent(@Param("containerBookingId") long containerBookingId, @Param("eventKey") String eventKey); } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQueryRepository.java b/firstleg/src/main/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQueryRepository.java index 407dbdcee..e71a2e5a2 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQueryRepository.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQueryRepository.java @@ -20,8 +20,10 @@ import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleNodeFactMapper; import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleExceptionFactMapper; import jakarta.annotation.Resource; import org.springframework.stereotype.Repository; +import org.springframework.dao.DuplicateKeyException; import java.util.List; +import java.util.Objects; import java.time.LocalDateTime; @Repository @@ -76,7 +78,20 @@ public class CarrierScheduleQueryRepository { .orderByAsc(CarrierScheduleNodeFactEntity::getObservedAt) .orderByAsc(CarrierScheduleNodeFactEntity::getId)); } - public boolean insertIfAbsent(CarrierScheduleNodeFactEntity value) { return nodes.insertIgnore(value) == 1; } + public boolean insertIfAbsent(CarrierScheduleNodeFactEntity value) { + try { + return nodes.insertStrict(value) == 1; + } catch (DuplicateKeyException duplicate) { + // Current reads also see a concurrent committed winner under MySQL REPEATABLE READ. + var idCollision = nodes.selectOne(new LambdaQueryWrapper() + .eq(CarrierScheduleNodeFactEntity::getId, value.getId()).last("FOR UPDATE")); + if (idCollision != null && (!Objects.equals(idCollision.getContainerBookingId(), value.getContainerBookingId()) + || !Objects.equals(idCollision.getEventKey(), value.getEventKey()))) throw duplicate; + if (nodeByContainer(value.getContainerBookingId(), value.getEventKey()) == null) throw duplicate; + // The service retains the immutable fact comparison and rejects changed payloads. + return false; + } + } public CarrierScheduleNodeFactEntity node(long queryId, String eventKey) { return nodes.selectOne(new LambdaQueryWrapper() .eq(CarrierScheduleNodeFactEntity::getQueryId, queryId) @@ -101,12 +116,21 @@ public class CarrierScheduleQueryRepository { .orderByAsc(CarrierScheduleExceptionFactEntity::getId)); } public boolean insertIfAbsent(CarrierScheduleExceptionFactEntity value) { - return exceptions.insertIfAbsent(value) == 1; + try { + return exceptions.insertStrict(value) == 1; + } catch (DuplicateKeyException duplicate) { + var idCollision = exceptions.selectOne(new LambdaQueryWrapper() + .eq(CarrierScheduleExceptionFactEntity::getId, value.getId()).last("FOR UPDATE")); + if (idCollision != null && (!Objects.equals(idCollision.getQueryId(), value.getQueryId()) + || !Objects.equals(idCollision.getEventKey(), value.getEventKey()))) throw duplicate; + if (exception(value.getQueryId(), value.getEventKey()) == null) throw duplicate; + return false; + } } public CarrierScheduleExceptionFactEntity exception(long queryId, String eventKey) { return exceptions.selectOne(new LambdaQueryWrapper() .eq(CarrierScheduleExceptionFactEntity::getQueryId, queryId) - .eq(CarrierScheduleExceptionFactEntity::getEventKey, eventKey)); + .eq(CarrierScheduleExceptionFactEntity::getEventKey, eventKey).last("FOR UPDATE")); } public boolean claimOperation(Long queryId, long operatorId, String operation, String key, String hash) { return operations.claim(IdWorker.getId(), queryId, operatorId, operation, key, hash) == 1; diff --git a/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/ImmutableCarrierFactPersistenceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/ImmutableCarrierFactPersistenceTest.java new file mode 100644 index 000000000..76f5baf2b --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/ImmutableCarrierFactPersistenceTest.java @@ -0,0 +1,184 @@ +package com.baiyue.firstleg.portschedule.repository; + +import com.baomidou.mybatisplus.annotation.TableName; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; +import com.baiyue.firstleg.portschedule.entity.CarrierScheduleNodeFactEntity; +import com.baiyue.firstleg.portschedule.entity.CarrierScheduleExceptionFactEntity; +import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleNodeFactMapper; +import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleExceptionFactMapper; +import org.apache.ibatis.annotations.Insert; +import org.h2.api.Trigger; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mybatis.spring.SqlSessionTemplate; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.util.ReflectionTestUtils; + +import java.lang.reflect.Modifier; +import java.sql.Connection; +import java.sql.SQLException; +import java.time.LocalDateTime; +import java.util.Arrays; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.*; + +/** H2 executes production Mapper SQL with an UPDATE-rejecting guard. It does not replace MySQL acceptance. */ +public class ImmutableCarrierFactPersistenceTest { + private JdbcTemplate jdbc; + private CarrierScheduleQueryRepository repository; + + public static class RejectUpdates implements Trigger { + @Override public void fire(Connection connection, Object[] oldRow, Object[] newRow) throws SQLException { + throw new SQLException("immutable carrier fact", "45000", 1644); + } + } + + @BeforeEach + void setup() throws Exception { + var dataSource = new JdbcDataSource(); + dataSource.setURL("jdbc:h2:mem:immutable-facts-" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1"); + jdbc = new JdbcTemplate(dataSource); + table(CarrierScheduleNodeFactEntity.class); table(CarrierScheduleExceptionFactEntity.class); + jdbc.execute("ALTER TABLE fl_carrier_schedule_node_fact ADD CONSTRAINT uq_node UNIQUE(container_booking_id,event_key)"); + jdbc.execute("ALTER TABLE fl_carrier_schedule_exception_fact ADD CONSTRAINT uq_exception UNIQUE(query_id,event_key)"); + jdbc.execute("ALTER TABLE fl_carrier_schedule_node_fact ALTER COLUMN provider_code SET NOT NULL"); + jdbc.execute("ALTER TABLE fl_carrier_schedule_exception_fact ADD CONSTRAINT valid_query CHECK(query_id>0)"); + for (String table : List.of("fl_carrier_schedule_node_fact", "fl_carrier_schedule_exception_fact")) + jdbc.execute("CREATE TRIGGER guard_" + table + " BEFORE UPDATE ON " + table + + " FOR EACH ROW CALL '" + RejectUpdates.class.getName() + "'"); + var configuration = new MybatisConfiguration(); configuration.setMapUnderscoreToCamelCase(true); + configuration.addMapper(CarrierScheduleNodeFactMapper.class); + configuration.addMapper(CarrierScheduleExceptionFactMapper.class); + var factory = new MybatisSqlSessionFactoryBean(); factory.setDataSource(dataSource); factory.setConfiguration(configuration); + var session = new SqlSessionTemplate(factory.getObject()); + repository = new CarrierScheduleQueryRepository(); + ReflectionTestUtils.setField(repository, "nodes", session.getMapper(CarrierScheduleNodeFactMapper.class)); + ReflectionTestUtils.setField(repository, "exceptions", session.getMapper(CarrierScheduleExceptionFactMapper.class)); + } + + @Test + void bothImmutableFactMappersUseStrictInsertWithoutAnyUpdateOrIgnoreClause() { + for (Class mapper : List.of(CarrierScheduleNodeFactMapper.class, CarrierScheduleExceptionFactMapper.class)) { + var sql = Arrays.stream(mapper.getDeclaredMethods()).filter(m -> m.isAnnotationPresent(Insert.class)) + .map(m -> String.join(" ", m.getAnnotation(Insert.class).value())).collect(Collectors.joining(" ")); + assertThat(sql).contains("INSERT INTO").doesNotContainIgnoringCase("ON DUPLICATE", "INSERT IGNORE", "UPDATE"); + } + } + + @Test + void nodeReplayAvoidsUpdateGuardAndPreservesOriginalFact() { + assertThat(repository.insertIfAbsent(node(1,3,"same"))).isTrue(); + var before = jdbc.queryForList("SELECT * FROM fl_carrier_schedule_node_fact"); + assertThat(repository.insertIfAbsent(node(2,3,"same"))).isFalse(); + assertThat(jdbc.queryForList("SELECT * FROM fl_carrier_schedule_node_fact")).isEqualTo(before); + } + + @Test + void exceptionReplayAvoidsUpdateGuardAndPreservesOriginalFact() { + assertThat(repository.insertIfAbsent(exception(1,5,"same"))).isTrue(); + var before = jdbc.queryForList("SELECT * FROM fl_carrier_schedule_exception_fact"); + assertThat(repository.insertIfAbsent(exception(2,5,"same"))).isFalse(); + assertThat(jdbc.queryForList("SELECT * FROM fl_carrier_schedule_exception_fact")).isEqualTo(before); + } + + @Test + void primaryKeyCollisionCannotBeMistakenForSameBusinessEventEvenWhenThatEventAlsoExists() { + repository.insertIfAbsent(node(1,3,"one")); repository.insertIfAbsent(node(2,4,"two")); + assertThatThrownBy(() -> repository.insertIfAbsent(node(1,4,"unrelated"))).isInstanceOf(DuplicateKeyException.class); + assertThatThrownBy(() -> repository.insertIfAbsent(node(1,4,"two"))).isInstanceOf(DuplicateKeyException.class); + repository.insertIfAbsent(exception(1,5,"one")); repository.insertIfAbsent(exception(2,6,"two")); + assertThatThrownBy(() -> repository.insertIfAbsent(exception(1,6,"unrelated"))).isInstanceOf(DuplicateKeyException.class); + assertThatThrownBy(() -> repository.insertIfAbsent(exception(1,6,"two"))).isInstanceOf(DuplicateKeyException.class); + } + + @Test + void nonDuplicateConstraintErrorsContinueToPropagate() { + var invalid = node(1,3,"invalid"); invalid.setProviderCode(null); + assertThatThrownBy(() -> repository.insertIfAbsent(invalid)).isInstanceOf(DataIntegrityViolationException.class); + assertThatThrownBy(() -> repository.insertIfAbsent(exception(1,0,"invalid"))).isInstanceOf(DataIntegrityViolationException.class); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_node_fact",Integer.class)).isZero(); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_exception_fact",Integer.class)).isZero(); + } + + @Test + void nonDuplicateSqlFailureIsNeverConvertedIntoAReplay() { + var nodes = org.mockito.Mockito.mock(CarrierScheduleNodeFactMapper.class); + var exceptions = org.mockito.Mockito.mock(CarrierScheduleExceptionFactMapper.class); + var target = new CarrierScheduleQueryRepository(); + ReflectionTestUtils.setField(target,"nodes",nodes); + ReflectionTestUtils.setField(target,"exceptions",exceptions); + var failure = new org.springframework.jdbc.UncategorizedSQLException("insert","INSERT", + new SQLException("immutable guard failure","45000",1644)); + org.mockito.Mockito.when(nodes.insertStrict(org.mockito.ArgumentMatchers.any())).thenThrow(failure); + org.mockito.Mockito.when(exceptions.insertStrict(org.mockito.ArgumentMatchers.any())).thenThrow(failure); + assertThatThrownBy(() -> target.insertIfAbsent(node(1,3,"event"))).isSameAs(failure); + assertThatThrownBy(() -> target.insertIfAbsent(exception(1,5,"event"))).isSameAs(failure); + org.mockito.Mockito.verify(nodes,org.mockito.Mockito.never()).selectOne(org.mockito.ArgumentMatchers.any()); + org.mockito.Mockito.verify(exceptions,org.mockito.Mockito.never()).selectOne(org.mockito.ArgumentMatchers.any()); + } + + @Test + void immutableUpdateGuardsRemainEffectiveForBothTables() { + repository.insertIfAbsent(node(1,3,"one")); repository.insertIfAbsent(exception(1,5,"one")); + assertThatThrownBy(() -> jdbc.update("UPDATE fl_carrier_schedule_node_fact SET id=id WHERE id=1")) + .hasMessageContaining("immutable carrier fact"); + assertThatThrownBy(() -> jdbc.update("UPDATE fl_carrier_schedule_exception_fact SET id=id WHERE id=1")) + .hasMessageContaining("immutable carrier fact"); + } + + @Test + void concurrentNodeDuplicateHasExactlyOneInsertAndOneConfirmedReplay() throws Exception { + concurrent(() -> repository.insertIfAbsent(node(1,3,"race")), () -> repository.insertIfAbsent(node(2,3,"race"))); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_node_fact",Integer.class)).isEqualTo(1); + } + + @Test + void concurrentExceptionDuplicateHasExactlyOneInsertAndOneConfirmedReplay() throws Exception { + concurrent(() -> repository.insertIfAbsent(exception(1,5,"race")), () -> repository.insertIfAbsent(exception(2,5,"race"))); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_exception_fact",Integer.class)).isEqualTo(1); + } + + private void concurrent(java.util.concurrent.Callable left, java.util.concurrent.Callable right) throws Exception { + var executor = Executors.newFixedThreadPool(2); + var ready = new CountDownLatch(2); var start = new CountDownLatch(1); + try { + var first = executor.submit(() -> { ready.countDown(); start.await(); return left.call(); }); + var second = executor.submit(() -> { ready.countDown(); start.await(); return right.call(); }); + assertThat(ready.await(5,TimeUnit.SECONDS)).isTrue(); start.countDown(); + assertThat(List.of(first.get(10,TimeUnit.SECONDS),second.get(10,TimeUnit.SECONDS))).containsExactlyInAnyOrder(true,false); + } finally { start.countDown(); executor.shutdownNow(); } + } + + private CarrierScheduleNodeFactEntity node(long id,long container,String event) { + var row=new CarrierScheduleNodeFactEntity(); row.setId(id); row.setQueryId(5L); row.setContainerBookingId(container); + row.setEventKey("MANUAL:"+event); row.setProviderCode("MANUAL"); row.setProviderEventId(event); + row.setProviderNodeCode("AT_ORIGIN_PORT"); row.setStandardNode("AT_ORIGIN_PORT"); row.setNodeSequence(20); + row.setObservedAt(LocalDateTime.of(2026,10,5,1,40)); row.setActualAt(LocalDateTime.of(2026,10,5,1,44)); + row.setRawPayloadHash("a".repeat(64)); return row; + } + + private CarrierScheduleExceptionFactEntity exception(long id,long query,String event) { + var row=new CarrierScheduleExceptionFactEntity(); row.setId(id); row.setQueryId(query); row.setEventKey("MATSON:"+event); + row.setProviderCode("MATSON"); row.setProviderEventId(event); row.setProviderStatus("HOLD"); + row.setExceptionCode("CUSTOMS_HOLD"); row.setActive(true); row.setObservedAt(LocalDateTime.of(2026,10,5,1,44)); + row.setRawPayloadHash("a".repeat(64)); return row; + } + + private void table(Class entity) { + var columns=Arrays.stream(entity.getDeclaredFields()).filter(f -> !Modifier.isStatic(f.getModifiers())) + .map(f -> f.getName().replaceAll("([a-z0-9])([A-Z])","$1_$2").toLowerCase()+" " + +(f.getType()==Long.class?"BIGINT":f.getType()==Integer.class?"INT":f.getType()==Boolean.class?"BOOLEAN":f.getType()==LocalDateTime.class?"TIMESTAMP(3)":"VARCHAR(16000)") + +(f.getName().equals("id")?" PRIMARY KEY":"")).collect(Collectors.joining(",")); + jdbc.execute("CREATE TABLE "+entity.getAnnotation(TableName.class).value()+"("+columns+")"); + } +} diff --git a/firstleg/src/test/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleExceptionPersistenceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleExceptionPersistenceTest.java index 6ec940065..8c9763e94 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleExceptionPersistenceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleExceptionPersistenceTest.java @@ -61,6 +61,12 @@ class CarrierScheduleExceptionPersistenceTest { + "operator_id,operator_name,created_at) VALUES(6,1,'NODE_OBSERVED',6,7,'IN_TRANSIT',0,'carrier-scheduler',?)", TIME); jdbc.update("INSERT INTO fl_carrier_schedule_query_history(id,query_id,action,from_version,to_version,reason," + "operator_id,operator_name,created_at) VALUES(7,1,'NODE_OBSERVED',7,7,'ARRIVED_DESTINATION_PORT',0,'carrier-scheduler',?)", TIME); + for (String table : List.of("fl_carrier_schedule_node_fact", "fl_carrier_schedule_exception_fact")) { + jdbc.execute("CREATE TRIGGER immutable_" + table + " BEFORE UPDATE ON " + table + + " FOR EACH ROW CALL '" + + com.baiyue.firstleg.portschedule.repository.ImmutableCarrierFactPersistenceTest.RejectUpdates.class.getName() + + "'"); + } MybatisConfiguration configuration = new MybatisConfiguration(); configuration.setMapUnderscoreToCamelCase(true); for (Class mapper : List.of(CarrierScheduleQueryMapper.class, CarrierScheduleQueryHistoryMapper.class, @@ -127,6 +133,23 @@ class CarrierScheduleExceptionPersistenceTest { assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_exception_fact WHERE source_reference IS NOT NULL AND raw_payload_hash IS NOT NULL", Integer.class)).isEqualTo(8); } + @Test + void originalContainerManualBodyReplaysWithoutUpdateHistoryOrFactMutationAndRejectsChangedFact() { + var body = new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ContainerManualNodeRequest( + "ARRIVED_DESTINATION_PORT", TIME.plusDays(2), null, TIME.plusDays(2), "USLAX", "Los Angeles", + "synthetic container manual fact"); + service.addContainerManualNode(3,10001,"test-operator","original-body-key",body); + var saved = allSnapshot(); + service.addContainerManualNode(3,10001,"test-operator","original-body-key",body); + assertThat(allSnapshot()).isEqualTo(saved); + assertThatThrownBy(() -> service.addContainerManualNode(3,10001,"test-operator","original-body-key", + new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ContainerManualNodeRequest( + body.standardNode(),body.observedAt(),null,body.actualAt().plusSeconds(1),body.locationCode(), + body.locationName(),body.reason()))) + .isInstanceOf(com.baiyue.common.core.exception.BizException.class).hasMessageContaining("事件事实摘要"); + assertThat(allSnapshot()).isEqualTo(saved); + } + @ParameterizedTest @ValueSource(strings = {"CUSTOMS_HOLD", "TERMINAL_HOLD", "DOCUMENT_HOLD", "UNKNOWN_HOLD"}) void sameFactsReplayPreservesAuditAndChangedFactsCannotOverwrite(String code) {