task-128: appearancepatent 事务收缩(flatten/序列化/存储/哈希纯计算抽 prepareSubmittedChunk 移出事务,双事务路径不变)+ 8 条边界测试
This commit is contained in:
+69
-25
@@ -405,7 +405,9 @@ public class AppearancePatentTaskService {
|
|||||||
|
|
||||||
private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) {
|
private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) {
|
||||||
if (transactionManager != null) {
|
if (transactionManager != null) {
|
||||||
SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(taskId, request));
|
// 纯计算(flatten 分组展开/序列化/payload 存储/哈希)在事务开始前完成
|
||||||
|
PreparedSubmittedChunk prepared = prepareSubmittedChunk(taskId, request);
|
||||||
|
SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(prepared));
|
||||||
inNewTransaction(() -> {
|
inNewTransaction(() -> {
|
||||||
completeSubmittedChunk(context);
|
completeSubmittedChunk(context);
|
||||||
return null;
|
return null;
|
||||||
@@ -638,7 +640,42 @@ public class AppearancePatentTaskService {
|
|||||||
return updatedMillis <= thresholdMillis;
|
return updatedMillis <= thresholdMillis;
|
||||||
}
|
}
|
||||||
|
|
||||||
private SubmitContext persistSubmittedChunk(Long taskId, AppearancePatentSubmitResultRequest request) {
|
/**
|
||||||
|
* /result 提交的纯计算阶段(无事务注解):读任务 + 校验归属 + 分组展开 +
|
||||||
|
* 序列化 + payload 存储 + 哈希预计算,全部在事务开始前完成。
|
||||||
|
* 重复 chunk(查重命中)时不做存储与哈希,落库由 persist 按同一查重短路。
|
||||||
|
*/
|
||||||
|
private PreparedSubmittedChunk prepareSubmittedChunk(Long taskId, AppearancePatentSubmitResultRequest request) {
|
||||||
|
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||||
|
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||||
|
throw new BusinessException("任务不存在");
|
||||||
|
}
|
||||||
|
ensureTaskOwnedByCurrentInstance(task, "submit result");
|
||||||
|
int chunkIndex = request.getChunkIndex() == null ? 0 : request.getChunkIndex();
|
||||||
|
int chunkTotal = request.getChunkTotal() == null ? 1 : request.getChunkTotal();
|
||||||
|
boolean done = Boolean.TRUE.equals(request.getDone());
|
||||||
|
String scopeKey = firstNonBlank(request.getSubmissionId(), "task:" + taskId);
|
||||||
|
String scopeHash = DigestUtil.sha256Hex(scopeKey);
|
||||||
|
|
||||||
|
TaskChunkEntity existing = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||||
|
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||||
|
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
|
||||||
|
.eq(TaskChunkEntity::getScopeHash, scopeHash)
|
||||||
|
.eq(TaskChunkEntity::getChunkIndex, chunkIndex)
|
||||||
|
.last("limit 1"));
|
||||||
|
if (existing != null) {
|
||||||
|
return new PreparedSubmittedChunk(task, scopeKey, scopeHash, chunkIndex, chunkTotal, done,
|
||||||
|
request.getError(), null, null, null);
|
||||||
|
}
|
||||||
|
List<AppearancePatentResultRowDto> rawRows = flattenSubmittedRows(request);
|
||||||
|
String payloadJson = writeJson(rawRows, "结果序列化失败");
|
||||||
|
String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson);
|
||||||
|
return new PreparedSubmittedChunk(task, scopeKey, scopeHash, chunkIndex, chunkTotal, done,
|
||||||
|
request.getError(), payloadJson, storedPayload, DigestUtil.sha256Hex(payloadJson));
|
||||||
|
}
|
||||||
|
|
||||||
|
private SubmitContext persistSubmittedChunk(PreparedSubmittedChunk prepared) {
|
||||||
|
Long taskId = prepared.task().getId();
|
||||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||||
throw new BusinessException("任务不存在");
|
throw new BusinessException("任务不存在");
|
||||||
@@ -648,48 +685,43 @@ public class AppearancePatentTaskService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
ensureTaskOwnedByCurrentInstance(task, "submit result");
|
ensureTaskOwnedByCurrentInstance(task, "submit result");
|
||||||
int chunkIndex = request.getChunkIndex() == null ? 0 : request.getChunkIndex();
|
|
||||||
int chunkTotal = request.getChunkTotal() == null ? 1 : request.getChunkTotal();
|
|
||||||
boolean done = Boolean.TRUE.equals(request.getDone());
|
|
||||||
String scopeKey = firstNonBlank(request.getSubmissionId(), "task:" + taskId);
|
|
||||||
String scopeHash = DigestUtil.sha256Hex(scopeKey);
|
|
||||||
taskCacheService.touchTaskHeartbeat(taskId);
|
taskCacheService.touchTaskHeartbeat(taskId);
|
||||||
|
|
||||||
TaskChunkEntity existing = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
TaskChunkEntity existing = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||||
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
|
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
|
||||||
.eq(TaskChunkEntity::getScopeHash, scopeHash)
|
.eq(TaskChunkEntity::getScopeHash, prepared.scopeHash())
|
||||||
.eq(TaskChunkEntity::getChunkIndex, chunkIndex)
|
.eq(TaskChunkEntity::getChunkIndex, prepared.chunkIndex())
|
||||||
.last("limit 1"));
|
.last("limit 1"));
|
||||||
if (existing == null) {
|
if (existing == null && prepared.storedPayload() != null) {
|
||||||
List<AppearancePatentResultRowDto> rawRows = flattenSubmittedRows(request);
|
|
||||||
String payloadJson = writeJson(rawRows, "结果序列化失败");
|
|
||||||
|
|
||||||
TaskChunkEntity chunk = new TaskChunkEntity();
|
TaskChunkEntity chunk = new TaskChunkEntity();
|
||||||
chunk.setTaskId(taskId);
|
chunk.setTaskId(taskId);
|
||||||
chunk.setModuleType(MODULE_TYPE);
|
chunk.setModuleType(MODULE_TYPE);
|
||||||
chunk.setScopeKey(scopeKey);
|
chunk.setScopeKey(prepared.scopeKey());
|
||||||
chunk.setScopeHash(scopeHash);
|
chunk.setScopeHash(prepared.scopeHash());
|
||||||
chunk.setChunkIndex(chunkIndex);
|
chunk.setChunkIndex(prepared.chunkIndex());
|
||||||
chunk.setChunkTotal(chunkTotal);
|
chunk.setChunkTotal(prepared.chunkTotal());
|
||||||
String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson);
|
chunk.setPayloadJson(prepared.storedPayload());
|
||||||
chunk.setPayloadJson(storedPayload);
|
chunk.setPayloadHash(prepared.payloadHash());
|
||||||
chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson));
|
|
||||||
chunk.setCreatedAt(LocalDateTime.now());
|
chunk.setCreatedAt(LocalDateTime.now());
|
||||||
chunk.setUpdatedAt(LocalDateTime.now());
|
chunk.setUpdatedAt(LocalDateTime.now());
|
||||||
try {
|
try {
|
||||||
taskChunkMapper.insert(chunk);
|
taskChunkMapper.insert(chunk);
|
||||||
} catch (DuplicateKeyException ex) {
|
} catch (DuplicateKeyException ex) {
|
||||||
log.info("[appearance-patent] duplicate chunk inserted concurrently taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex);
|
log.info("[appearance-patent] duplicate chunk inserted concurrently taskId={} scope={} chunk={}",
|
||||||
|
taskId, prepared.scopeKey(), prepared.chunkIndex());
|
||||||
}
|
}
|
||||||
} else {
|
} else if (existing != null) {
|
||||||
log.info("[appearance-patent] duplicate chunk ignored taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex);
|
log.info("[appearance-patent] duplicate chunk ignored taskId={} scope={} chunk={}",
|
||||||
|
taskId, prepared.scopeKey(), prepared.chunkIndex());
|
||||||
}
|
}
|
||||||
|
|
||||||
upsertScopeState(taskId, scopeKey, scopeHash, chunkTotal, request.getError(), done, false);
|
upsertScopeState(taskId, prepared.scopeKey(), prepared.scopeHash(), prepared.chunkTotal(),
|
||||||
|
prepared.error(), prepared.done(), false);
|
||||||
task.setUpdatedAt(LocalDateTime.now());
|
task.setUpdatedAt(LocalDateTime.now());
|
||||||
fileTaskMapper.updateById(task);
|
fileTaskMapper.updateById(task);
|
||||||
return new SubmitContext(task, scopeKey, scopeHash, chunkIndex, done, request.getError());
|
return new SubmitContext(task, prepared.scopeKey(), prepared.scopeHash(), prepared.chunkIndex(),
|
||||||
|
prepared.done(), prepared.error());
|
||||||
}
|
}
|
||||||
|
|
||||||
private void completeSubmittedChunk(SubmitContext context) {
|
private void completeSubmittedChunk(SubmitContext context) {
|
||||||
@@ -2907,6 +2939,18 @@ public class AppearancePatentTaskService {
|
|||||||
String error) {
|
String error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private record PreparedSubmittedChunk(FileTaskEntity task,
|
||||||
|
String scopeKey,
|
||||||
|
String scopeHash,
|
||||||
|
Integer chunkIndex,
|
||||||
|
Integer chunkTotal,
|
||||||
|
boolean done,
|
||||||
|
String error,
|
||||||
|
String payloadJson,
|
||||||
|
String storedPayload,
|
||||||
|
String payloadHash) {
|
||||||
|
}
|
||||||
|
|
||||||
private static class SourceRowsBuilder {
|
private static class SourceRowsBuilder {
|
||||||
private final String sourceFileKey;
|
private final String sourceFileKey;
|
||||||
private String sourceFilename;
|
private String sourceFilename;
|
||||||
|
|||||||
+320
@@ -0,0 +1,320 @@
|
|||||||
|
package com.nanri.aiimage.modules.appearancepatent.service;
|
||||||
|
|
||||||
|
import cn.hutool.crypto.digest.DigestUtil;
|
||||||
|
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.common.service.DistributedJobLockService;
|
||||||
|
import com.nanri.aiimage.config.AppearancePatentProperties;
|
||||||
|
import com.nanri.aiimage.config.InstanceMetadata;
|
||||||
|
import com.nanri.aiimage.config.StorageProperties;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentLlmClient;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentSubmitResultRequest;
|
||||||
|
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
|
||||||
|
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
|
import org.mockito.InjectMocks;
|
||||||
|
import org.mockito.Mock;
|
||||||
|
import org.mockito.Spy;
|
||||||
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
import org.springframework.test.util.ReflectionTestUtils;
|
||||||
|
import org.springframework.transaction.PlatformTransactionManager;
|
||||||
|
import org.springframework.transaction.TransactionDefinition;
|
||||||
|
import org.springframework.transaction.TransactionStatus;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyLong;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.doAnswer;
|
||||||
|
import static org.mockito.Mockito.inOrder;
|
||||||
|
import static org.mockito.Mockito.lenient;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
|
import static org.mockito.Mockito.times;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* task-128:appearancepatent 事务收缩。
|
||||||
|
* 纯计算(flatten 分组展开/序列化/payload 存储/哈希)抽为无事务的
|
||||||
|
* prepareSubmittedChunk 在事务开始前调用;双事务路径(persistSubmittedChunk +
|
||||||
|
* completeSubmittedChunk 两个独立 inNewTransaction)保持不变;落库仍在事务内。
|
||||||
|
*/
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
class AppearancePatentTaskServiceTxBoundaryTest {
|
||||||
|
|
||||||
|
private static final Long TASK_ID = 31337L;
|
||||||
|
private static final String PARSED_POINTER = "rustfs:task-parsed/appearance-patent/31337/payload.json";
|
||||||
|
private static final String CHUNK_POINTER = "rustfs:task-chunk/appearance-patent/31337/chunk.json";
|
||||||
|
private static final String STORED_CHUNK_POINTER = "\"" + CHUNK_POINTER + "\"";
|
||||||
|
|
||||||
|
@Mock private LocalFileStorageService localFileStorageService;
|
||||||
|
@Mock private OssStorageService ossStorageService;
|
||||||
|
@Mock private StorageProperties storageProperties;
|
||||||
|
@Mock private FileTaskMapper fileTaskMapper;
|
||||||
|
@Mock private FileResultMapper fileResultMapper;
|
||||||
|
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||||
|
@Mock private TaskChunkMapper taskChunkMapper;
|
||||||
|
@Spy private ObjectMapper objectMapper = new ObjectMapper();
|
||||||
|
@Mock private AppearancePatentLlmClient llmClient;
|
||||||
|
@Mock private AppearancePatentTaskCacheService taskCacheService;
|
||||||
|
@Mock private AppearancePatentProperties properties;
|
||||||
|
@Mock private TaskFileJobService taskFileJobService;
|
||||||
|
@Mock private TaskProgressSnapshotService taskProgressSnapshotService;
|
||||||
|
@Mock private TransientPayloadStorageService transientPayloadStorageService;
|
||||||
|
@Mock private PlatformTransactionManager transactionManager;
|
||||||
|
@Mock private DistributedJobLockService distributedJobLockService;
|
||||||
|
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||||
|
@Mock private InstanceMetadata instanceMetadata;
|
||||||
|
@Mock private TaskProgressLightAssembler taskProgressLightAssembler;
|
||||||
|
@Mock private TransactionStatus transactionStatus;
|
||||||
|
|
||||||
|
@InjectMocks private AppearancePatentTaskService service;
|
||||||
|
|
||||||
|
private final AtomicBoolean transactionActive = new AtomicBoolean();
|
||||||
|
private final AtomicReference<TaskChunkEntity> insertedChunk = new AtomicReference<>();
|
||||||
|
private final AtomicReference<String> storedPayloadJson = new AtomicReference<>();
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void initializeMybatisMetadata() {
|
||||||
|
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||||
|
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||||
|
}
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
void setUpTransactionAndLock() {
|
||||||
|
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
|
||||||
|
lenient().when(transactionManager.getTransaction(any(TransactionDefinition.class)))
|
||||||
|
.thenAnswer(invocation -> {
|
||||||
|
transactionActive.set(true);
|
||||||
|
return transactionStatus;
|
||||||
|
});
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
transactionActive.set(false);
|
||||||
|
return null;
|
||||||
|
}).when(transactionManager).commit(transactionStatus);
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
transactionActive.set(false);
|
||||||
|
return null;
|
||||||
|
}).when(transactionManager).rollback(transactionStatus);
|
||||||
|
lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of());
|
||||||
|
lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of());
|
||||||
|
lenient().when(taskChunkMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
lenient().when(taskScopeStateMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null);
|
||||||
|
lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null);
|
||||||
|
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||||
|
lenient().when(transientPayloadStorageService.storeChunkPayload(
|
||||||
|
eq(AppearancePatentTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString()))
|
||||||
|
.thenAnswer(invocation -> {
|
||||||
|
storedPayloadJson.set(invocation.getArgument(4));
|
||||||
|
return STORED_CHUNK_POINTER;
|
||||||
|
});
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
TaskChunkEntity chunk = invocation.getArgument(0);
|
||||||
|
assertTrue(transactionActive.get(), "chunk 落库必须在事务内");
|
||||||
|
chunk.setId(501L);
|
||||||
|
insertedChunk.set(chunk);
|
||||||
|
return 1;
|
||||||
|
}).when(taskChunkMapper).insert(any(TaskChunkEntity.class));
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
TaskScopeStateEntity scope = invocation.getArgument(0);
|
||||||
|
scope.setId(601L);
|
||||||
|
return 1;
|
||||||
|
}).when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
|
||||||
|
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void payloadHashIsPrecomputedInPrepareBeforeTransaction() throws Exception {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
|
||||||
|
Object prepared = ReflectionTestUtils.invokeMethod(
|
||||||
|
service, "prepareSubmittedChunk", TASK_ID, request(false));
|
||||||
|
assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()),
|
||||||
|
ReflectionTestUtils.getField(prepared, "payloadHash"),
|
||||||
|
"prepare 阶段必须产出预计算的 payload 哈希(事务开始前可用)");
|
||||||
|
verify(taskChunkMapper, never()).insert(any(TaskChunkEntity.class));
|
||||||
|
verify(taskScopeStateMapper, never()).insert(any(TaskScopeStateEntity.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void computeMovedOutsideTransaction() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request(false));
|
||||||
|
|
||||||
|
// prepare 存储(纯计算段)→ 事务 1 → 事务 2
|
||||||
|
var order = inOrder(transientPayloadStorageService, transactionManager, taskChunkMapper);
|
||||||
|
order.verify(transientPayloadStorageService).storeChunkPayload(
|
||||||
|
eq(AppearancePatentTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString());
|
||||||
|
order.verify(transactionManager).getTransaction(any(TransactionDefinition.class));
|
||||||
|
order.verify(taskChunkMapper).insert(any(TaskChunkEntity.class));
|
||||||
|
order.verify(transactionManager).getTransaction(any(TransactionDefinition.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void dualTransactionPathIsUnchanged() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request(false));
|
||||||
|
|
||||||
|
verify(transactionManager, times(2)).getTransaction(any(TransactionDefinition.class));
|
||||||
|
verify(transactionManager, times(2)).commit(transactionStatus);
|
||||||
|
verify(transactionManager, never()).rollback(transactionStatus);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void persistenceStaysInsideFirstTransaction() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request(false));
|
||||||
|
|
||||||
|
TaskChunkEntity chunk = insertedChunk.get();
|
||||||
|
assertEquals(TASK_ID, chunk.getTaskId());
|
||||||
|
assertEquals(STORED_CHUNK_POINTER, chunk.getPayloadJson());
|
||||||
|
assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()), chunk.getPayloadHash());
|
||||||
|
// persist 与 complete 两个事务各 touch 一次任务行(双事务路径不变)
|
||||||
|
verify(fileTaskMapper, times(2)).updateById(any(FileTaskEntity.class));
|
||||||
|
verify(taskScopeStateMapper, times(2)).insert(any(TaskScopeStateEntity.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void firstTransactionFailureSkipsSecondAndScheduling() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenThrow(
|
||||||
|
new IllegalStateException("persist tx failed"));
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class, () -> service.submitResult(TASK_ID, request(false)));
|
||||||
|
|
||||||
|
verify(transactionManager, times(1)).getTransaction(any(TransactionDefinition.class));
|
||||||
|
verify(transactionManager, never()).commit(transactionStatus);
|
||||||
|
verify(transactionManager).rollback(transactionStatus);
|
||||||
|
verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void duplicateChunkIsIgnoredWithoutStoreOrInsert() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
TaskChunkEntity existing = new TaskChunkEntity();
|
||||||
|
existing.setId(501L);
|
||||||
|
existing.setTaskId(TASK_ID);
|
||||||
|
existing.setScopeHash("existing-scope");
|
||||||
|
existing.setChunkIndex(0);
|
||||||
|
when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request(false));
|
||||||
|
|
||||||
|
verify(transientPayloadStorageService, never()).storeChunkPayload(
|
||||||
|
anyString(), anyLong(), anyString(), any(), anyString());
|
||||||
|
verify(taskChunkMapper, never()).insert(any(TaskChunkEntity.class));
|
||||||
|
verify(taskScopeStateMapper, times(2)).insert(any(TaskScopeStateEntity.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void chunkSnapshotIsStable() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request(false));
|
||||||
|
|
||||||
|
TaskChunkEntity chunk = insertedChunk.get();
|
||||||
|
assertEquals(TASK_ID, chunk.getTaskId());
|
||||||
|
assertEquals(AppearancePatentTaskService.MODULE_TYPE, chunk.getModuleType());
|
||||||
|
assertEquals("appearance-patent-31337", chunk.getScopeKey());
|
||||||
|
assertEquals(DigestUtil.sha256Hex("appearance-patent-31337"), chunk.getScopeHash());
|
||||||
|
assertEquals(0, chunk.getChunkIndex());
|
||||||
|
assertEquals(1, chunk.getChunkTotal());
|
||||||
|
assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()), chunk.getPayloadHash());
|
||||||
|
assertNotNull(chunk.getCreatedAt());
|
||||||
|
assertNotNull(chunk.getUpdatedAt());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void fullChainRunsPrepareThenTwoTransactionsThenScheduling() {
|
||||||
|
FileTaskEntity task = runningTask("instance-a");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
TaskChunkEntity chunk = new TaskChunkEntity();
|
||||||
|
chunk.setId(501L);
|
||||||
|
chunk.setTaskId(TASK_ID);
|
||||||
|
chunk.setModuleType(AppearancePatentTaskService.MODULE_TYPE);
|
||||||
|
when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk));
|
||||||
|
lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString()))
|
||||||
|
.thenReturn("{\"allItems\":[],\"items\":[],\"sourceFiles\":[]}");
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
FileResultEntity result = invocation.getArgument(0);
|
||||||
|
result.setId(701L);
|
||||||
|
return 1;
|
||||||
|
}).when(fileResultMapper).insert(any(FileResultEntity.class));
|
||||||
|
when(taskFileJobService.enqueueAssembleResult(
|
||||||
|
eq(TASK_ID), eq(AppearancePatentTaskService.MODULE_TYPE), eq(701L), anyString()))
|
||||||
|
.thenReturn(null);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request(false));
|
||||||
|
|
||||||
|
var order = inOrder(transientPayloadStorageService, transactionManager, taskFileJobService);
|
||||||
|
order.verify(transientPayloadStorageService).storeChunkPayload(
|
||||||
|
eq(AppearancePatentTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString());
|
||||||
|
order.verify(transactionManager).getTransaction(any(TransactionDefinition.class));
|
||||||
|
order.verify(transactionManager).getTransaction(any(TransactionDefinition.class));
|
||||||
|
order.verify(taskFileJobService).enqueueAssembleResult(
|
||||||
|
eq(TASK_ID), eq(AppearancePatentTaskService.MODULE_TYPE), eq(701L), anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
private AppearancePatentSubmitResultRequest request(boolean done) {
|
||||||
|
AppearancePatentSubmitResultRequest request = new AppearancePatentSubmitResultRequest();
|
||||||
|
request.setSubmissionId("appearance-patent-31337");
|
||||||
|
request.setChunkIndex(0);
|
||||||
|
request.setChunkTotal(1);
|
||||||
|
request.setDone(done);
|
||||||
|
return request;
|
||||||
|
}
|
||||||
|
|
||||||
|
private FileTaskEntity runningTask(String owner) {
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(TASK_ID);
|
||||||
|
task.setModuleType(AppearancePatentTaskService.MODULE_TYPE);
|
||||||
|
task.setStatus("RUNNING");
|
||||||
|
task.setUserId(7L);
|
||||||
|
String resultJson = "{\"parsedPayloadRef\":\"" + PARSED_POINTER + "\"";
|
||||||
|
if (owner != null) {
|
||||||
|
resultJson += ",\"ownerInstanceId\":\"" + owner + "\"";
|
||||||
|
}
|
||||||
|
task.setResultJson(resultJson + "}");
|
||||||
|
return task;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user