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

feat(task): 客户端崩溃恢复接口 POST /heartbeat/{taskId}/interrupted——客户端重启发现上次进程崩溃时立即把残留 RUNNING 任务标终态，替代最长等 30 分钟的心跳超时兜底

Browse Source
This commit is contained in:
黄自达
2026-09-11 16:09:42 +08:00
parent bb74c9a192
commit 8dc03df95b
2 changed files with 63 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/controller/TaskHeartbeatController.java
+13
View File
@@ -33,4 +33,17 @@ public class TaskHeartbeatController {
@Valid @RequestBody(required = false) TaskHeartbeatRequest request) {
return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request));
}
@PostMapping("/{taskId}/interrupted")
@Operation(
summary = "上报客户端异常中断",
description = "客户端重启后发现自己上次进程崩溃时调用：将仍在 RUNNING 的任务立即标为终态"
+ "（file_task→FAILED，brand_crawl_tasks→cancelled），替代最长 30 分钟的 stale 兜底。幂等，非 RUNNING 状态不修改。")
public ApiResponse<TaskHeartbeatVo> interrupted(
@Parameter(description = "任务 ID", required = true, example = "200")
@PathVariable Long taskId,
@RequestBody(required = false) TaskHeartbeatRequest request) {
String reason = request == null ? null : request.getPhase();
return ApiResponse.success(taskHeartbeatService.markInterrupted(taskId, reason));
}
}
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java
+50
View File
@@ -117,6 +117,56 @@ public class TaskHeartbeatService {
return fileTaskMapper.selectOne(fileQuery);
}
/**
* 客户端崩溃恢复：客户端重启后发现自己上次进程级崩溃（journal 残留），
* 调此接口把上次还在 RUNNING 的任务立即标终态，替代最长等 30 分钟的 stale 兜底。
* file_task RUNNING → FAILED；brand_crawl_tasks running → cancelled；其余状态不动（幂等）。
*/
public TaskHeartbeatVo markInterrupted(Long taskId, String reason) {
if (taskId == null || taskId <= 0) {
log.warn("[task-interrupted] ignored invalid taskId={}", taskId);
return TaskHeartbeatVo.notAlive(null, null, "invalid taskId");
}
String safeReason = (reason == null || reason.isBlank())
? "客户端异常中断，任务已自动失败"
: "客户端异常中断: " + reason.trim();
FileTaskEntity fileTask = selectFileTask(taskId);
if (fileTask != null) {
String status = fileTask.getStatus();
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, taskId)
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
.set(FileTaskEntity::getStatus, "FAILED")
.set(FileTaskEntity::getErrorMessage, safeReason)
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
if (updated > 0) {
log.warn("[task-interrupted] file task marked failed by client restart taskId={} moduleType={} reason={}",
taskId, fileTask.getModuleType(), safeReason);
return TaskHeartbeatVo.notAlive(fileTask.getModuleType(), "FAILED", "marked failed");
}
log.info("[task-interrupted] file task not in RUNNING, skipped taskId={} status={}", taskId, status);
return TaskHeartbeatVo.notAlive(fileTask.getModuleType(), status, "task is not running");
}
BrandCrawlTaskEntity brandTask = selectBrandTask(taskId);
if (brandTask != null) {
String status = brandTask.getStatus();
if ("running".equalsIgnoreCase(status) || "pending".equalsIgnoreCase(status)) {
brandTask.setStatus("cancelled");
brandTask.setErrorMessage(safeReason);
int updated = brandCrawlTaskMapper.updateById(brandTask);
if (updated > 0) {
log.warn("[task-interrupted] brand task marked cancelled by client restart taskId={} reason={}",
taskId, safeReason);
return TaskHeartbeatVo.notAlive(MODULE_BRAND, "cancelled", "marked cancelled");
}
}
log.info("[task-interrupted] brand task not in running/pending, skipped taskId={} status={}", taskId, status);
return TaskHeartbeatVo.notAlive(MODULE_BRAND, status, "task is not running");
}
log.warn("[task-interrupted] task not found taskId={}", taskId);
return TaskHeartbeatVo.notAlive(null, null, "task not found");
}
private BrandCrawlTaskEntity selectBrandTask(Long taskId) {
LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getId, taskId)