From 39fb23bd16adf8f99fa25847fa87f6f6166c63fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Fri, 4 Sep 2026 23:44:39 +0800 Subject: [PATCH] =?UTF-8?q?task-180:=20Worker=20=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E5=AF=B9=E9=BD=90=E8=A7=84=E8=8C=83=EF=BC=88?= =?UTF-8?q?TaskResultFileJobWorker=20=E5=85=B3=E9=94=AE=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=E8=A1=A5=20stage=3D/errorType=3D=EF=BC=8C=E4=BF=9D=E7=95=99=20?= =?UTF-8?q?[task-file-job]=20=E5=89=8D=E7=BC=80=EF=BC=89+=208=20=E6=9D=A1?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - dispatch/success/failure/orphan/heartbeat/requeue/defer/waittxllm/finalize 日志追加规范 stage 字面量 - 失败/孤儿分支追加 errorType={}（exception 类名）；保留 jobId/taskId/moduleType 与既有语义，仅消息文本追加、无逻辑改动 - 审计测试锁定字段名一致、阶段字面量白名单、无敏感字段 --- .../task/service/TaskResultFileJobWorker.java | 22 +-- .../TaskResultFileJobWorkerLogAuditTest.java | 126 ++++++++++++++++++ 2 files changed, 137 insertions(+), 11 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java index 7fef078d..f88e1961 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java @@ -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) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java b/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java new file mode 100644 index 00000000..dcaa8b0b --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java @@ -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 lines() throws IOException { + return Files.readAllLines(Path.of(System.getProperty("user.dir"), WORKER)); + } + + private static boolean hasLine(List lines, String needle) { + return lines.stream().anyMatch(l -> l.contains(needle)); + } + + @Test + void dispatchLogHasStageAndJobFields() throws IOException { + List 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 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 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 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 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 lines = lines(); + assertTrue(hasLine(lines, "stage=FINALIZE"), "重试耗尽终态回调应带 stage=FINALIZE"); + assertTrue(hasLine(lines, "stage=ORPHAN"), "孤儿 job 中断应带 stage=ORPHAN"); + } + + @Test + void fieldNamesConsistentAndPrefixKept() throws IOException { + List lines = lines(); + List 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 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 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()); + } + } +}