super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-156: 孤立 Job 巡检报表（task/result 缺失孤儿检出、重试耗尽同样报、只读不修改、可重复）+ 8 条测试

Browse Source
This commit is contained in:
黄自达
2026-09-02 06:53:06 +08:00
parent d64ef5284a
commit d773748ae9
2 changed files with 266 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/OrphanJobInspector.java
+91
View File
@@ -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<OrphanJobEntry> entries) {
public boolean isEmpty() {
return entries == null || entries.isEmpty();
}
}
public OrphanJobReport inspectOrphanJobs(int limit) {
List<TaskFileJobEntity> jobs = taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
.orderByAsc(TaskFileJobEntity::getUpdatedAt)
.last("limit " + Math.max(1, Math.min(limit, 500))));
if (jobs == null || jobs.isEmpty()) {
return new OrphanJobReport(List.of());
}
Set<Long> taskIds = jobs.stream()
.map(TaskFileJobEntity::getTaskId)
.filter(id -> id != null && id > 0)
.collect(Collectors.toSet());
Set<Long> resultIds = jobs.stream()
.map(TaskFileJobEntity::getResultId)
.filter(id -> id != null && id > 0)
.collect(Collectors.toSet());
Set<Long> existingTaskIds = taskIds.isEmpty() ? Set.of()
: fileTaskMapper.selectBatchIds(taskIds).stream()
.map(task -> task.getId())
.collect(Collectors.toSet());
Set<Long> existingResultIds = resultIds.isEmpty() ? Set.of()
: fileResultMapper.selectBatchIds(resultIds).stream()
.map(result -> result.getId())
.collect(Collectors.toSet());
List<OrphanJobEntry> 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;
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/task/service/OrphanJobInspectorTest.java
+175
View File
@@ -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());
}
}