diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index 31290091..a323343f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -405,7 +405,9 @@ public class AppearancePatentTaskService { private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) { if (transactionManager != null) { - SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(taskId, request)); + // 纯计算（flatten 分组展开/序列化/payload 存储/哈希）在事务开始前完成 + PreparedSubmittedChunk prepared = prepareSubmittedChunk(taskId, request); + SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(prepared)); inNewTransaction(() -> { completeSubmittedChunk(context); return null; @@ -638,7 +640,42 @@ public class AppearancePatentTaskService { 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() + .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 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); if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { throw new BusinessException("任务不存在"); @@ -648,48 +685,43 @@ public class AppearancePatentTaskService { } 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); TaskChunkEntity existing = taskChunkMapper.selectOne(new LambdaQueryWrapper() .eq(TaskChunkEntity::getTaskId, taskId) .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) - .eq(TaskChunkEntity::getScopeHash, scopeHash) - .eq(TaskChunkEntity::getChunkIndex, chunkIndex) + .eq(TaskChunkEntity::getScopeHash, prepared.scopeHash()) + .eq(TaskChunkEntity::getChunkIndex, prepared.chunkIndex()) .last("limit 1")); - if (existing == null) { - List rawRows = flattenSubmittedRows(request); - String payloadJson = writeJson(rawRows, "结果序列化失败"); - + if (existing == null && prepared.storedPayload() != null) { TaskChunkEntity chunk = new TaskChunkEntity(); chunk.setTaskId(taskId); chunk.setModuleType(MODULE_TYPE); - chunk.setScopeKey(scopeKey); - chunk.setScopeHash(scopeHash); - chunk.setChunkIndex(chunkIndex); - chunk.setChunkTotal(chunkTotal); - String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson); - chunk.setPayloadJson(storedPayload); - chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson)); + chunk.setScopeKey(prepared.scopeKey()); + chunk.setScopeHash(prepared.scopeHash()); + chunk.setChunkIndex(prepared.chunkIndex()); + chunk.setChunkTotal(prepared.chunkTotal()); + chunk.setPayloadJson(prepared.storedPayload()); + chunk.setPayloadHash(prepared.payloadHash()); chunk.setCreatedAt(LocalDateTime.now()); chunk.setUpdatedAt(LocalDateTime.now()); try { taskChunkMapper.insert(chunk); } 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 { - log.info("[appearance-patent] duplicate chunk ignored taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex); + } else if (existing != null) { + 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()); 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) { @@ -2907,6 +2939,18 @@ public class AppearancePatentTaskService { 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 final String sourceFileKey; private String sourceFilename; diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskServiceTxBoundaryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskServiceTxBoundaryTest.java new file mode 100644 index 00000000..72905ddd --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskServiceTxBoundaryTest.java @@ -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 insertedChunk = new AtomicReference<>(); + private final AtomicReference 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; + } +}