From d0df738bbac6e18edcab455de47f39e24ef4d5f8 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Fri, 2 Oct 2026 19:50:47 +0800 Subject: [PATCH] fix(firstleg): keep loading evidence replacement transactional --- .../CabinetLoadingEvidenceService.java | 7 +- .../CabinetLoadingEvidenceServiceTest.java | 1 + ...CabinetLoadingEvidenceTransactionTest.java | 254 ++++++++++++++++++ 3 files changed, 259 insertions(+), 3 deletions(-) create mode 100644 firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceTransactionTest.java diff --git a/firstleg/src/main/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceService.java b/firstleg/src/main/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceService.java index c5673de44..d42c07b48 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceService.java @@ -84,13 +84,14 @@ public class CabinetLoadingEvidenceService { value.setCreatedBy(operatorId); value.setCreatedByName(operatorName); value.setUpdatedBy(operatorId); value.setUpdatedByName(operatorName); try { mapper.insert(value); } catch (DuplicateKeyException exception) { - CabinetLoadingEvidenceEntity concurrent = mapper.byOperation(taskId, key); - if (concurrent != null && Objects.equals(concurrent.getRequestHash(), requestHash)) return view(scope); + // A replacement may already have voided the old row in this transaction. + // Roll it back; a subsequent same-key request can read the committed winner. throw conflict("装柜证据发生并发变更，请刷新后重试"); } return view(scope); } + @Transactional public EvidenceList upload(long taskId, long warehouseId, String key, int expectedVersion, String evidenceType, String sealNo, String remark, MultipartFile multipart, long operatorId, String operatorName) { @@ -167,7 +168,7 @@ public class CabinetLoadingEvidenceService { } private Scope scope(long taskId, long warehouseId, long operatorId, boolean mutable) { - LoadingPickTaskEntity task = loadingPicks.task(taskId); + LoadingPickTaskEntity task = mutable ? loadingPicks.taskForUpdate(taskId) : loadingPicks.task(taskId); var booking = task == null ? null : loadingPicks.activeBookingByTask(taskId); if (task == null || task.getContainerBookingId() == null || booking == null || !Objects.equals(task.getContainerBookingId(), booking.getId())) { diff --git a/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceServiceTest.java index fad9fdb85..8bfb330e8 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceServiceTest.java @@ -48,6 +48,7 @@ class CabinetLoadingEvidenceServiceTest { LoadingPickAssignmentEntity assignment = new LoadingPickAssignmentEntity(); assignment.setPickTaskId(20L); assignment.setAssigneeId(9L); when(loadingPicks.task(20L)).thenReturn(task); + when(loadingPicks.taskForUpdate(20L)).thenReturn(task); when(loadingPicks.activeBookingByTask(20L)).thenReturn(booking); when(loadingPicks.assignment(20L)).thenReturn(assignment); } diff --git a/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceTransactionTest.java b/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceTransactionTest.java new file mode 100644 index 000000000..c6e23423f --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/loading/service/CabinetLoadingEvidenceTransactionTest.java @@ -0,0 +1,254 @@ +package com.baiyue.firstleg.loading.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; +import com.baiyue.common.core.api.ApiResponse; +import com.baiyue.firstleg.client.LoadingEvidenceFileClient; +import com.baiyue.firstleg.entity.*; +import com.baiyue.firstleg.loading.entity.CabinetLoadingEvidenceEntity; +import com.baiyue.firstleg.loading.mapper.CabinetLoadingEvidenceMapper; +import com.baiyue.firstleg.repository.LoadingPickRepository; +import com.baiyue.firstleg.service.InboundHoldGuard; +import com.fasterxml.jackson.databind.ObjectMapper; +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.aop.framework.ProxyFactory; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.mock.web.MockMultipartFile; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import java.lang.reflect.Modifier; +import java.lang.reflect.Proxy; +import java.lang.reflect.InvocationTargetException; +import java.time.LocalDateTime; +import java.util.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** External File/scope are stubs; evidence writes, locks, constraints and transaction advice are real. */ +class CabinetLoadingEvidenceTransactionTest { + private JdbcTemplate jdbc; + private CabinetLoadingEvidenceMapper mapper; + private CabinetLoadingEvidenceService target; + private CabinetLoadingEvidenceService service; + private LoadingEvidenceFileClient files; + private com.baiyue.firstleg.mapper.LoadingPickTaskMapper tasks; + private org.springframework.transaction.support.TransactionTemplate transaction; + private long nextFile; + private final MockMultipartFile photo = new MockMultipartFile("file", "photo.png", "image/png", new byte[]{1, 2, 3}); + + @BeforeEach + void setup() throws Exception { + var ds = new JdbcDataSource(); + ds.setURL("jdbc:h2:mem:evidence-" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1"); + jdbc = new JdbcTemplate(ds); + String columns = Arrays.stream(CabinetLoadingEvidenceEntity.class.getDeclaredFields()) + .filter(field -> !Modifier.isStatic(field.getModifiers())).map(field -> { + String name = field.getName().replaceAll("([a-z0-9])([A-Z])", "$1_$2").toLowerCase(); + String type = field.getType() == Long.class ? "BIGINT" : field.getType() == Integer.class ? "INT" + : field.getType() == LocalDateTime.class ? "TIMESTAMP(3)" : "VARCHAR(2000)"; + return name + " " + type + (name.equals("id") ? " PRIMARY KEY" : ""); + }).collect(Collectors.joining(",")); + jdbc.execute("CREATE TABLE fl_loading_cabinet_evidence(" + columns + ")"); + jdbc.execute("ALTER TABLE fl_loading_cabinet_evidence ADD CONSTRAINT uq_operation UNIQUE(source_task_id,operation_key)"); + jdbc.execute("ALTER TABLE fl_loading_cabinet_evidence ADD CONSTRAINT uq_seal UNIQUE(seal_no,active_flag)"); + String taskColumns = Arrays.stream(LoadingPickTaskEntity.class.getDeclaredFields()) + .filter(field -> !Modifier.isStatic(field.getModifiers())).map(field -> + field.getName().replaceAll("([a-z0-9])([A-Z])", "$1_$2").toLowerCase() + " " + + (field.getType() == Long.class ? "BIGINT" : field.getType() == Integer.class ? "INT" + : field.getType() == LocalDateTime.class ? "TIMESTAMP(3)" : "VARCHAR(2000)") + + (field.getName().equals("id") ? " PRIMARY KEY" : "")).collect(Collectors.joining(",")); + jdbc.execute("CREATE TABLE fl_loading_pick_task(" + taskColumns + ")"); + jdbc.update("INSERT INTO fl_loading_pick_task(id,container_booking_id,task_status,version) VALUES(20,99,'LOADING',0)"); + var config = new MybatisConfiguration(); config.setMapUnderscoreToCamelCase(true); + config.addMapper(CabinetLoadingEvidenceMapper.class); + config.addMapper(com.baiyue.firstleg.mapper.LoadingPickTaskMapper.class); + var factory = new MybatisSqlSessionFactoryBean(); factory.setDataSource(ds); factory.setConfiguration(config); + var session = new SqlSessionTemplate(factory.getObject()); + mapper = session.getMapper(CabinetLoadingEvidenceMapper.class); + tasks = session.getMapper(com.baiyue.firstleg.mapper.LoadingPickTaskMapper.class); + var picks = spy(new LoadingPickRepository()); + ReflectionTestUtils.setField(picks, "tasks", tasks); + var booking = new ContainerBookingEntity(); booking.setId(99L); booking.setWarehouseId(8L); + var assignment = new LoadingPickAssignmentEntity(); assignment.setAssigneeId(9L); + doReturn(booking).when(picks).activeBookingByTask(20L); + doReturn(assignment).when(picks).assignment(20L); + files = mock(LoadingEvidenceFileClient.class); + when(files.upload(anyString(), eq(9L), eq(99L), anyString(), any())).thenAnswer(call -> ApiResponse.success( + new LoadingEvidenceFileClient.UploadResult(nextFile, "PLATFORM", 0L, "LOADING_" + call.getArgument(3), "99", "photo.png", "image/png", 3L))); + when(files.validate(anyString(), eq(99L), anyString(), anyLong())).thenAnswer(call -> ApiResponse.success( + new LoadingEvidenceFileClient.FileMetadata(call.getArgument(3), "PLATFORM", 0L, "LOADING_" + call.getArgument(2), "99", "photo.png", "image/png"))); + target = new CabinetLoadingEvidenceService(); + ReflectionTestUtils.setField(target, "mapper", mapper); ReflectionTestUtils.setField(target, "loadingPicks", picks); + ReflectionTestUtils.setField(target, "files", files); ReflectionTestUtils.setField(target, "internalToken", "offline"); + ReflectionTestUtils.setField(target, "inboundHoldGuard", mock(InboundHoldGuard.class)); + ReflectionTestUtils.setField(target, "objectMapper", new ObjectMapper().findAndRegisterModules()); + var proxy = new ProxyFactory(target); + var manager = new DataSourceTransactionManager(ds); + transaction = new org.springframework.transaction.support.TransactionTemplate(manager); + proxy.addAdvice(new TransactionInterceptor(manager, new AnnotationTransactionAttributeSource())); + service = (CabinetLoadingEvidenceService) proxy.getProxy(); + } + + @Test + void replacementInsertFailureRollsBackOldActiveRowAndSameRequestCanRetry() { + nextFile = 301; upload("original"); + var before = snapshot(); + jdbc.execute("ALTER TABLE fl_loading_cabinet_evidence ADD CONSTRAINT reject_new_file CHECK(file_id <> 302)"); + nextFile = 302; + assertThatThrownBy(() -> upload("replacement")).isInstanceOf(RuntimeException.class); + assertThat(snapshot()).isEqualTo(before); + jdbc.execute("ALTER TABLE fl_loading_cabinet_evidence DROP CONSTRAINT reject_new_file"); + upload("replacement"); + assertThat(mapper.byContainer(99)).hasSize(2).filteredOn(row -> "ACTIVE".equals(row.getStatus())) + .singleElement().satisfies(row -> assertThat(row.getFileId()).isEqualTo(302L)); + } + + @Test + void duplicateKeyAfterStaleReplayReadsMustRollBackPriorVoid() { + nextFile = 301; upload("original"); + nextFile = 302; upload("winner"); + var before = snapshot(); + // Simulate two stale idempotency reads from an overlapping invocation. + // Every write and the duplicate operation-key violation still execute via the real mapper/database. + AtomicInteger reads = new AtomicInteger(); + var staleReads = (CabinetLoadingEvidenceMapper) Proxy.newProxyInstance(getClass().getClassLoader(), + new Class[]{CabinetLoadingEvidenceMapper.class}, (proxy, method, args) -> { + if (method.getName().equals("byOperation") && reads.incrementAndGet() <= 2) return null; + try { return method.invoke(mapper, args); } catch (InvocationTargetException ex) { throw ex.getCause(); } + }); + ReflectionTestUtils.setField(target, "mapper", staleReads); + nextFile = 303; + assertThatThrownBy(() -> upload("winner")).isInstanceOf(RuntimeException.class); + assertThat(snapshot()).isEqualTo(before); + ReflectionTestUtils.setField(target, "mapper", mapper); + upload("winner"); + assertThat(snapshot()).isEqualTo(before); + } + + @Test + void sequentialSameKeyReplayDoesNotUploadAgainAndDifferentPayloadCannotMutate() { + nextFile = 301; upload("original"); var before = snapshot(); + upload("original"); assertThat(snapshot()).isEqualTo(before); + verify(files, times(1)).upload(anyString(), anyLong(), anyLong(), anyString(), any()); + assertThatThrownBy(() -> service.upload(20, 8, "original", 0, "PHOTO_STEP_1", null, "changed", photo, 9, "operator")) + .hasMessageContaining("幂等键"); + assertThat(snapshot()).isEqualTo(before); + } + + @Test + void realConcurrentSameKeyCommandsConvergeWithoutVoidingCommittedWinner() throws Exception { + nextFile = 301; upload("original"); + var barrier = new java.util.concurrent.CyclicBarrier(2); + var ids = new java.util.concurrent.atomic.AtomicLong(301); + when(files.upload(anyString(), eq(9L), eq(99L), anyString(), any())).thenAnswer(call -> { + long fileId = ids.incrementAndGet(); + return ApiResponse.success(new LoadingEvidenceFileClient.UploadResult(fileId, "PLATFORM", 0L, + "LOADING_PHOTO_STEP_1", "99", "photo.png", "image/png", 3L)); + }); + var pool = java.util.concurrent.Executors.newFixedThreadPool(2); + try { + java.util.concurrent.Callable command = () -> { + barrier.await(10, java.util.concurrent.TimeUnit.SECONDS); + try { upload("concurrent"); return true; } + catch (com.baiyue.common.core.exception.BizException conflict) { + assertThat(conflict.getMessage()).matches(".*(并发|版本).*" ); return false; + } + }; + var first = pool.submit(command); var second = pool.submit(command); + boolean firstPassed = first.get(20, java.util.concurrent.TimeUnit.SECONDS); + boolean secondPassed = second.get(20, java.util.concurrent.TimeUnit.SECONDS); + assertThat(firstPassed || secondPassed).isTrue(); + assertThat(mapper.byContainer(99)).hasSize(2).filteredOn(row -> "ACTIVE".equals(row.getStatus())) + .singleElement().satisfies(row -> assertThat(row.getOperationKey()).isEqualTo("concurrent")); + var committed = snapshot(); + upload("concurrent"); + assertThat(snapshot()).isEqualTo(committed); + } finally { pool.shutdownNow(); } + } + + @Test + void wrongWarehouseAndVersionRetainOriginalRows() { + nextFile = 301; upload("original"); var before = snapshot(); + assertThatThrownBy(() -> service.upload(20, 7, "wrong-warehouse", 0, "PHOTO_STEP_1", null, "fixture", photo, 9, "operator")) + .hasMessageContaining("仓库"); + nextFile = 302; + assertThatThrownBy(() -> service.upload(20, 8, "wrong-version", 9, "PHOTO_STEP_1", null, "fixture", photo, 9, "operator")) + .hasMessageContaining("版本"); + assertThat(snapshot()).isEqualTo(before); + } + + @Test + void sealOwnedByAnotherContainerCannotBeReused() { + var foreign = new CabinetLoadingEvidenceEntity(); + foreign.setId(600L); foreign.setContainerBookingId(100L); foreign.setSourceTaskId(21L); + foreign.setEvidenceType("SEAL"); foreign.setFileId(601L); foreign.setSealNo("OTHER-SEAL"); + foreign.setOperationKey("foreign"); foreign.setStatus("ACTIVE"); foreign.setActiveFlag(1); foreign.setVersion(0); + mapper.insert(foreign); + var before = snapshot(); nextFile = 301; + assertThatThrownBy(() -> service.upload(20, 8, "seal", 0, "SEAL", "OTHER-SEAL", "fixture", photo, 9, "operator")) + .hasMessageContaining("锁码已用于其他柜次"); + assertThat(snapshot()).isEqualTo(before); + } + + @Test + void heldTaskFailsBeforeFileUploadAndDoesNotChangeEvidence() { + nextFile = 301; upload("original"); var before = snapshot(); + var guard = (InboundHoldGuard) ReflectionTestUtils.getField(target, "inboundHoldGuard"); + doThrow(new com.baiyue.common.core.exception.BizException(com.baiyue.common.core.api.ResultCode.CONFLICT, "已扣货")) + .when(guard).requireTaskNotHeld(20L); + assertThatThrownBy(() -> upload("held")).hasMessageContaining("扣货"); + assertThat(snapshot()).isEqualTo(before); + verify(files, times(1)).upload(anyString(), anyLong(), anyLong(), anyString(), any()); + } + + @Test + void completedTaskRejectsUploadBeforeFileAndEvidenceWrites() { + jdbc.update("UPDATE fl_loading_pick_task SET task_status='COMPLETED' WHERE id=20"); + assertThatThrownBy(() -> upload("late")).hasMessageContaining("不在可补证状态"); + verifyNoInteractions(files); + assertThat(snapshot()).isEmpty(); + } + + @Test + void concurrentCompletionHoldsTaskLockAndLateUploadSeesCommittedTerminalStatus() throws Exception { + var locked = new java.util.concurrent.CountDownLatch(1); + var release = new java.util.concurrent.CountDownLatch(1); + var started = new java.util.concurrent.CountDownLatch(1); + var pool = java.util.concurrent.Executors.newFixedThreadPool(2); + try { + var completion = pool.submit(() -> transaction.executeWithoutResult(status -> { + tasks.selectForUpdate(20); + assertThat(tasks.acceptCompletion(20, 0)).isEqualTo(1); + assertThat(tasks.completeAccepted(20, 9, "operator")).isEqualTo(1); + locked.countDown(); + try { if (!release.await(10, java.util.concurrent.TimeUnit.SECONDS)) throw new IllegalStateException("test timeout"); } + catch (InterruptedException e) { throw new RuntimeException(e); } + })); + assertThat(locked.await(10, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + nextFile = 301; + var upload = pool.submit(() -> { started.countDown(); upload("late-concurrent"); }); + assertThat(started.await(10, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + try { + assertThatThrownBy(() -> upload.get(200, java.util.concurrent.TimeUnit.MILLISECONDS)) + .isInstanceOf(java.util.concurrent.TimeoutException.class); + } finally { release.countDown(); } + completion.get(10, java.util.concurrent.TimeUnit.SECONDS); + assertThatThrownBy(() -> upload.get(10, java.util.concurrent.TimeUnit.SECONDS)) + .hasCauseInstanceOf(com.baiyue.common.core.exception.BizException.class) + .hasStackTraceContaining("不在可补证状态"); + assertThat(snapshot()).isEmpty(); + verifyNoInteractions(files); + } finally { release.countDown(); pool.shutdownNow(); } + } + + private void upload(String key) { service.upload(20, 8, key, 0, "PHOTO_STEP_1", null, "fixture", photo, 9, "operator"); } + private List> snapshot() { return jdbc.queryForList("SELECT * FROM fl_loading_cabinet_evidence ORDER BY id"); } +}