fix(firstleg): keep loading evidence replacement transactional
This commit is contained in:
+4
-3
@@ -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())) {
|
||||
|
||||
+1
@@ -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);
|
||||
}
|
||||
|
||||
+254
@@ -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<Boolean> 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<Map<String, Object>> snapshot() { return jdbc.queryForList("SELECT * FROM fl_loading_cabinet_evidence ORDER BY id"); }
|
||||
}
|
||||
Reference in New Issue
Block a user