diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/OrphanJobInspector.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/OrphanJobInspector.java new file mode 100644 index 00000000..06556116 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/OrphanJobInspector.java @@ -0,0 +1,91 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskFileJobMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * 孤立 Job 巡检（task-156）。 + * + * 只读巡检：task_file_job 无对应 task 或 result 的孤儿 Job 清单输出报表 + * （日志 + 返回对象）；绝不修改 Job 状态。巡检可重复执行。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class OrphanJobInspector { + + private final TaskFileJobMapper taskFileJobMapper; + private final FileTaskMapper fileTaskMapper; + private final FileResultMapper fileResultMapper; + + public record OrphanJobEntry(Long jobId, Long taskId, Long resultId, String moduleType, + String status, String updatedAt) { + } + + public record OrphanJobReport(List entries) { + + public boolean isEmpty() { + return entries == null || entries.isEmpty(); + } + } + + public OrphanJobReport inspectOrphanJobs(int limit) { + List jobs = taskFileJobMapper.selectList(new LambdaQueryWrapper() + .orderByAsc(TaskFileJobEntity::getUpdatedAt) + .last("limit " + Math.max(1, Math.min(limit, 500)))); + if (jobs == null || jobs.isEmpty()) { + return new OrphanJobReport(List.of()); + } + Set taskIds = jobs.stream() + .map(TaskFileJobEntity::getTaskId) + .filter(id -> id != null && id > 0) + .collect(Collectors.toSet()); + Set resultIds = jobs.stream() + .map(TaskFileJobEntity::getResultId) + .filter(id -> id != null && id > 0) + .collect(Collectors.toSet()); + Set existingTaskIds = taskIds.isEmpty() ? Set.of() + : fileTaskMapper.selectBatchIds(taskIds).stream() + .map(task -> task.getId()) + .collect(Collectors.toSet()); + Set existingResultIds = resultIds.isEmpty() ? Set.of() + : fileResultMapper.selectBatchIds(resultIds).stream() + .map(result -> result.getId()) + .collect(Collectors.toSet()); + + List entries = new ArrayList<>(); + for (TaskFileJobEntity job : jobs) { + boolean taskMissing = job.getTaskId() != null && job.getTaskId() > 0 + && !existingTaskIds.contains(job.getTaskId()); + boolean resultMissing = job.getResultId() != null && job.getResultId() > 0 + && !existingResultIds.contains(job.getResultId()); + if (taskMissing || resultMissing) { + entries.add(new OrphanJobEntry( + job.getId(), job.getTaskId(), job.getResultId(), job.getModuleType(), + job.getStatus(), + job.getUpdatedAt() == null ? null : job.getUpdatedAt().toString())); + } + } + OrphanJobReport report = new OrphanJobReport(List.copyOf(entries)); + if (!report.isEmpty()) { + log.info("orphan job report: count={}", report.entries().size()); + for (OrphanJobEntry entry : report.entries()) { + log.info("orphan job: jobId={} taskId={} resultId={} moduleType={} status={} updatedAt={}", + entry.jobId(), entry.taskId(), entry.resultId(), entry.moduleType(), + entry.status(), entry.updatedAt()); + } + } + return report; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/OrphanJobInspectorTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/OrphanJobInspectorTest.java new file mode 100644 index 00000000..3e4c7bdd --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/OrphanJobInspectorTest.java @@ -0,0 +1,175 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskFileJobMapper; +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.TaskFileJobEntity; +import org.apache.ibatis.builder.MapperBuilderAssistant; +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.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * task-156：孤立 Job 巡检报表契约（plan 09）。 + * task_file_job 无对应 task/result 的孤儿 Job 清单；只读不修改状态；可重复。 + */ +@ExtendWith(MockitoExtension.class) +class OrphanJobInspectorTest { + + private static final Long ORPHAN_JOB_ID = 9001L; + private static final Long ORPHAN_TASK_ID = 9002L; + private static final Long ORPHAN_RESULT_ID = 9003L; + private static final Long VALID_JOB_ID = 9101L; + private static final Long VALID_TASK_ID = 9102L; + private static final Long VALID_RESULT_ID = 9103L; + + @Mock private TaskFileJobMapper taskFileJobMapper; + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + + private OrphanJobInspector inspector; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, TaskFileJobEntity.class); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + } + + @BeforeEach + void setUp() { + inspector = new OrphanJobInspector(taskFileJobMapper, fileTaskMapper, fileResultMapper); + lenient().when(fileTaskMapper.selectBatchIds(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectBatchIds(any())).thenReturn(List.of()); + } + + private TaskFileJobEntity job(Long id, Long taskId, Long resultId, String status) { + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(id); + job.setTaskId(taskId); + job.setResultId(resultId); + job.setModuleType("SIMILAR_ASIN"); + job.setJobType("ASSEMBLE_RESULT"); + job.setStatus(status); + job.setUpdatedAt(LocalDateTime.now()); + return job; + } + + @Test + void orphanJobDetectedWhenTaskMissing() { + when(taskFileJobMapper.selectList(any())) + .thenReturn(List.of(job(ORPHAN_JOB_ID, ORPHAN_TASK_ID, ORPHAN_RESULT_ID, "PENDING"))); + + var report = inspector.inspectOrphanJobs(50); + + assertEquals(1, report.entries().size()); + assertEquals(ORPHAN_JOB_ID, report.entries().getFirst().jobId()); + } + + @Test + void validJobNotReported() { + when(taskFileJobMapper.selectList(any())) + .thenReturn(List.of(job(VALID_JOB_ID, VALID_TASK_ID, VALID_RESULT_ID, "RUNNING"))); + FileTaskEntity task = new FileTaskEntity(); + task.setId(VALID_TASK_ID); + FileResultEntity result = new FileResultEntity(); + result.setId(VALID_RESULT_ID); + when(fileTaskMapper.selectBatchIds(any())).thenReturn(List.of(task)); + when(fileResultMapper.selectBatchIds(any())).thenReturn(List.of(result)); + + var report = inspector.inspectOrphanJobs(50); + + assertTrue(report.isEmpty(), "task/result 齐全的 Job 不得进报表"); + } + + @Test + void reportOutputsDetails() { + when(taskFileJobMapper.selectList(any())) + .thenReturn(List.of(job(ORPHAN_JOB_ID, ORPHAN_TASK_ID, ORPHAN_RESULT_ID, "FAILED"))); + + var report = inspector.inspectOrphanJobs(50); + + OrphanJobInspector.OrphanJobEntry entry = report.entries().getFirst(); + assertEquals(ORPHAN_JOB_ID, entry.jobId()); + assertEquals(ORPHAN_TASK_ID, entry.taskId()); + assertEquals(ORPHAN_RESULT_ID, entry.resultId()); + assertEquals("FAILED", entry.status()); + assertNotNull(entry.updatedAt()); + assertFalse(entry.moduleType().isBlank()); + } + + @Test + void readOnlyDoesNotModifyJobs() { + when(taskFileJobMapper.selectList(any())) + .thenReturn(List.of(job(ORPHAN_JOB_ID, ORPHAN_TASK_ID, ORPHAN_RESULT_ID, "PENDING"))); + + inspector.inspectOrphanJobs(50); + + verify(taskFileJobMapper, never()).update(any(), any()); + verify(taskFileJobMapper, never()).delete(any()); + verify(taskFileJobMapper, never()).updateById(any(TaskFileJobEntity.class)); + } + + @Test + void retryExhaustedOrphanStillReported() { + TaskFileJobEntity exhausted = job(ORPHAN_JOB_ID, ORPHAN_TASK_ID, ORPHAN_RESULT_ID, "FAILED"); + exhausted.setRetryCount(5); + when(taskFileJobMapper.selectList(any())).thenReturn(List.of(exhausted)); + + var report = inspector.inspectOrphanJobs(50); + + assertEquals(1, report.entries().size(), "重试耗尽孤儿同样检出"); + } + + @Test + void emptyReportWhenNoJobs() { + when(taskFileJobMapper.selectList(any())).thenReturn(List.of()); + + var report = inspector.inspectOrphanJobs(50); + + assertTrue(report.isEmpty()); + } + + @Test + void rerunIsSafeAndStable() { + when(taskFileJobMapper.selectList(any())) + .thenReturn(List.of(job(ORPHAN_JOB_ID, ORPHAN_TASK_ID, ORPHAN_RESULT_ID, "PENDING"))); + + var first = inspector.inspectOrphanJobs(50); + var second = inspector.inspectOrphanJobs(50); + + assertEquals(first.entries().size(), second.entries().size()); + assertEquals(first.entries().getFirst().jobId(), second.entries().getFirst().jobId()); + } + + @Test + void limitIsClamped() { + when(taskFileJobMapper.selectList(any())).thenReturn(List.of()); + + inspector.inspectOrphanJobs(0); + inspector.inspectOrphanJobs(10_000); + + verify(taskFileJobMapper, org.mockito.Mockito.times(2)).selectList(any()); + } +}