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

task-180: Worker 日志字段对齐规范（TaskResultFileJobWorker 关键阶段补 stage=/errorType=，保留 [task-file-job] 前缀）+ 8 条测试

Browse Source 
- dispatch/success/failure/orphan/heartbeat/requeue/defer/waittxllm/finalize 日志追加规范 stage 字面量
- 失败/孤儿分支追加 errorType={}（exception 类名）；保留 jobId/taskId/moduleType 与既有语义，仅消息文本追加、无逻辑改动
- 审计测试锁定字段名一致、阶段字面量白名单、无敏感字段
This commit is contained in:
黄自达
2026-09-04 23:44:39 +08:00
parent 5fa59680f2
commit 39fb23bd16
2 changed files with 137 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java
+11 -11
View File
@@ -117,7 +117,7 @@ public class TaskResultFileJobWorker {
// 重试耗尽后把归属机上跑得好好的任务判成失败。
// job 不会因此卡死：归属实例的 runPendingJobs 每 15 秒扫一次自己 owner 的
// PENDING/FAILED job，enqueueAssembleResult / stuck 扫描也会补发 dispatch。
log.info("[task-file-job] skip owner-scoped job because owner is another instance jobId={} taskId={} moduleType={} status={} owner={} current={}",
log.info("[task-file-job] skip owner-scoped job because owner is another instance jobId={} taskId={} moduleType={} status={} owner={} current={} stage=SKIP_OWNER",
job.getId(), job.getTaskId(), job.getModuleType(), job.getStatus(),
ownerFromScopeKey(job.getScopeKey()), currentInstanceId());
return;
@@ -160,7 +160,7 @@ public class TaskResultFileJobWorker {
try {
taskFileJobService.touchRunning(job.getId());
} catch (Exception ex) {
log.warn("[task-file-job] heartbeat failed jobId={} taskId={} moduleType={} msg={}",
log.warn("[task-file-job] heartbeat failed jobId={} taskId={} moduleType={} msg={} stage=HEARTBEAT",
job.getId(), job.getTaskId(), job.getModuleType(), ex.getMessage());
}
}, interval, interval, TimeUnit.MILLISECONDS);
@@ -194,7 +194,7 @@ public class TaskResultFileJobWorker {
taskDistributedLockService.acquire(job.getModuleType(), job.getTaskId(), TaskDistributedLockService.DEFAULT_WAIT_MILLIS);
if (lockHandle == null) {
taskFileJobService.requeue(job.getId(), "Task is busy, waiting for previous task operation");
log.info("[task-file-job] process requeued because task lock is busy jobId={} taskId={} moduleType={} resultId={}",
log.info("[task-file-job] process requeued because task lock is busy jobId={} taskId={} moduleType={} resultId={} stage=REQUEUE",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId());
return;
}
@@ -209,13 +209,13 @@ public class TaskResultFileJobWorker {
if (!completed) {
if (isOwnerScopedJob(job)) {
taskFileJobService.touchRunning(job.getId());
log.info("[task-file-job] process waiting for async llm result jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
log.info("[task-file-job] process waiting for async llm result jobId={} taskId={} moduleType={} resultId={} elapsedMs={} stage=WAIT_LLM",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt);
return;
}
taskFileJobService.deferRunning(job.getId(), "Waiting for LLM/file assembly to continue");
log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={} stage=DEFER",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt);
return;
@@ -224,7 +224,7 @@ public class TaskResultFileJobWorker {
taskFileJobService.markSuccess(job, resultFileUrl);
cleanupAfterSuccess(job);
finalizeWithdraw = "WITHDRAW".equals(job.getModuleType());
log.info("[task-file-job] process success jobId={} taskId={} moduleType={} resultId={} elapsedMs={} resultFileUrl={}",
log.info("[task-file-job] process success jobId={} taskId={} moduleType={} resultId={} elapsedMs={} resultFileUrl={} stage=SUCCESS",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt, resultFileUrl);
}
@@ -248,13 +248,13 @@ public class TaskResultFileJobWorker {
// 孤儿 job：fileTask/fileResult 已被删除却残留 task_file_job 行，重试 5 次也救不回来，
// 直接拉满 retryCount 让 worker 跳过，避免每 15 秒刷一次 task not found 警告日志。
if (isOrphanJobFailure(ex, message)) {
log.warn("[task-file-job] process aborted because owning task/result no longer exists jobId={} taskId={} moduleType={} resultId={} msg={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message);
log.warn("[task-file-job] process aborted because owning task/result no longer exists jobId={} taskId={} moduleType={} resultId={} msg={} stage=ORPHAN errorType={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message, ex.getClass().getSimpleName());
taskFileJobService.markFailedPermanent(job, message);
return;
}
log.warn("[task-file-job] process failed jobId={} taskId={} moduleType={} resultId={} msg={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message);
log.warn("[task-file-job] process failed jobId={} taskId={} moduleType={} resultId={} msg={} stage=FAILED errorType={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message, ex.getClass().getSimpleName());
taskFileJobService.markFailed(job, message);
if (taskFileJobService.isRetryExhausted(job.getId())) {
finalizeRetryExhausted(job, message);
@@ -266,7 +266,7 @@ public class TaskResultFileJobWorker {
try {
notifyRetryExhausted(job, message);
if (!taskFileJobService.markFailureFinalized(job.getId(), message)) {
log.warn("[task-file-job] exhausted job terminal callback was already finalized or claim was lost jobId={} taskId={} moduleType={}",
log.warn("[task-file-job] exhausted job terminal callback was already finalized or claim was lost jobId={} taskId={} moduleType={} stage=FINALIZE",
job.getId(), job.getTaskId(), job.getModuleType());
}
} catch (Exception ex) {
backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java
+126
View File
@@ -0,0 +1,126 @@
package com.nanri.aiimage.config;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* task-180：TaskResultFileJobWorker 日志字段与规范对齐（spec 11 §2）。
*
* Worker 的 [task-file-job] 日志统一携带 jobId/taskId/moduleType，并补充规范字段 stage
* （阶段字面量）与 errorType（异常类名）；保留既有前缀与语义，不记录敏感内容。
* 源码锚点审计，纯本地可重复。
*/
class TaskResultFileJobWorkerLogAuditTest {
private static final String WORKER = "src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java";
private List<String> lines() throws IOException {
return Files.readAllLines(Path.of(System.getProperty("user.dir"), WORKER));
}
private static boolean hasLine(List<String> lines, String needle) {
return lines.stream().anyMatch(l -> l.contains(needle));
}
@Test
void dispatchLogHasStageAndJobFields() throws IOException {
List<String> lines = lines();
String target = lines.stream()
.filter(l -> l.contains("[task-file-job]") && l.contains("stage=SKIP_OWNER"))
.findFirst().orElseThrow();
assertTrue(target.contains("jobId={}") && target.contains("taskId={}") && target.contains("moduleType={}"),
"dispatch 日志应带 jobId/taskId/moduleType: " + target.trim());
}
@Test
void successLogHasStageAndFields() throws IOException {
List<String> lines = lines();
String target = lines.stream()
.filter(l -> l.contains("process success") && l.contains("[task-file-job]"))
.findFirst().orElseThrow();
assertTrue(target.contains("stage=SUCCESS"), "成功日志应带 stage=SUCCESS");
assertTrue(target.contains("jobId={}") && target.contains("taskId={}") && target.contains("moduleType={}"));
}
@Test
void failureLogHasStageAndErrorType() throws IOException {
List<String> lines = lines();
String target = lines.stream()
.filter(l -> l.contains("process failed") && l.contains("[task-file-job]"))
.findFirst().orElseThrow();
assertTrue(target.contains("stage=FAILED"), "失败日志应带 stage=FAILED");
assertTrue(target.contains("errorType={}"), "失败日志应带 errorType 占位");
assertTrue(hasLine(lines, "message, ex.getClass().getSimpleName());"),
"失败日志应记录异常类名");
}
@Test
void retryAndRecoverLogsHaveStage() throws IOException {
List<String> lines = lines();
assertTrue(hasLine(lines, "stage=REQUEUE"), "锁忙重排应带 stage=REQUEUE");
assertTrue(hasLine(lines, "stage=DEFER"), "延后等待应带 stage=DEFER");
assertTrue(hasLine(lines, "stage=WAIT_LLM"), "等待异步 LLM 应带 stage=WAIT_LLM");
}
@Test
void heartbeatLogHasStageAndFields() throws IOException {
List<String> lines = lines();
String target = lines.stream()
.filter(l -> l.contains("heartbeat failed") && l.contains("[task-file-job]"))
.findFirst().orElseThrow();
assertTrue(target.contains("stage=HEARTBEAT"), "心跳日志应带 stage=HEARTBEAT");
assertTrue(target.contains("jobId={}") && target.contains("taskId={}") && target.contains("moduleType={}"));
}
@Test
void stuckFinalizeLogsHaveStage() throws IOException {
List<String> lines = lines();
assertTrue(hasLine(lines, "stage=FINALIZE"), "重试耗尽终态回调应带 stage=FINALIZE");
assertTrue(hasLine(lines, "stage=ORPHAN"), "孤儿 job 中断应带 stage=ORPHAN");
}
@Test
void fieldNamesConsistentAndPrefixKept() throws IOException {
List<String> lines = lines();
List<String> perJob = lines.stream()
.filter(l -> l.contains("[task-file-job]") && l.contains("log."))
.toList();
assertFalse(perJob.isEmpty());
for (String l : perJob) {
assertFalse(l.contains("job="), "不应使用 job= 漂移字段: " + l.trim());
assertFalse(l.contains("task="), "不应使用 task= 漂移字段: " + l.trim());
assertFalse(l.contains("module="), "不应使用 module= 漂移字段: " + l.trim());
}
// 阶段字面量来自允许集合（防新增阶段拼写漂移）
List<String> allowed = List.of("SKIP_OWNER", "REQUEUE", "WAIT_LLM", "DEFER", "SUCCESS",
"ORPHAN", "FAILED", "FINALIZE", "HEARTBEAT");
for (String l : perJob) {
if (l.contains("stage=")) {
int idx = l.indexOf("stage=");
String rawValue = l.substring(idx + "stage=".length()).split("[^A-Z0-9_]")[0];
assertTrue(allowed.contains(rawValue),
"阶段字面量应来自允许集合: " + rawValue + " @ " + l.trim());
}
}
}
@Test
void noSensitiveFieldsInWorkerLogs() throws IOException {
List<String> lines = lines();
for (String l : lines) {
if (!l.contains("[task-file-job]")) {
continue;
}
assertFalse(l.contains("password=") || l.contains("secret=") || l.contains("apiKey=")
|| l.contains("cookie=") || l.contains("Bearer") || l.contains("authorization"),
"Worker 日志不应记录敏感字段: " + l.trim());
}
}
}