From d58fd0c3fdf7aa803e4131992533c5ffd65e5aba Mon Sep 17 00:00:00 2001 From: matevip Date: Wed, 6 May 2026 17:40:47 +0800 Subject: [PATCH] fix(task): gate every worker completion branch with isConversationCanceled --- .../tool/image/ImageGenerationService.java | 24 ++++++++++------- .../tool/music/MusicGenerationService.java | 27 ++++++++++++------- .../tool/video/VideoGenerationService.java | 24 ++++++++++------- 3 files changed, 47 insertions(+), 28 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java index 583d3789..cbb76cb8 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java @@ -221,6 +221,16 @@ public class ImageGenerationService { * 异步任务完成时的回写逻辑：下载图片 → 保存消息 → 广播 SSE */ private void handleAsyncCompletion(AsyncTaskEntity task, TaskPollResult result) { + // The conversation may have been deleted while the poller was running. + // Gate every post-completion side effect — file write, message save, + // success/failure broadcast — so we never write to a tombstoned + // conversation regardless of which sub-branch we'd take. + if (asyncTaskService.isConversationCanceled(task.getConversationId())) { + log.info("[ImageGen] Task {} (success={}) aborted: conversation {} was deleted", + task.getTaskId(), result.succeeded(), task.getConversationId()); + return; + } + if (result.succeeded()) { try { String imageUrl = result.imageUrl(); @@ -231,15 +241,6 @@ public class ImageGenerationService { return; } - // Conversation deletion can race with the final poll iteration; - // dropping the result avoids resurrecting attachment files and - // a dangling mate_message row. - if (asyncTaskService.isConversationCanceled(task.getConversationId())) { - log.info("[ImageGen] Task {} succeeded but conversation {} was deleted, dropping result", - task.getTaskId(), task.getConversationId()); - return; - } - // 下载图片到本地 Path localPath = fileDownloader.download(imageUrl, task.getConversationId(), task.getTaskId(), 0); String servingUrl = fileDownloader.toServingUrl(task.getConversationId(), localPath); @@ -262,6 +263,11 @@ public class ImageGenerationService { } catch (Exception e) { log.error("[ImageGen] Completion handling failed for task {}: {}", task.getTaskId(), e.getMessage(), e); + if (asyncTaskService.isConversationCanceled(task.getConversationId())) { + log.info("[ImageGen] Skipping failure broadcast for deleted conversation {}", + task.getConversationId()); + return; + } asyncTaskService.broadcastTaskEvent(task, "async_task_completed", false, null, null, "图片下载或保存失败: " + e.getMessage()); } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java index 4900cf1b..66ebc617 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java @@ -110,6 +110,18 @@ public class MusicGenerationService { asyncTaskService.updateStatus(task.getTaskId(), "running", null, null, null); MusicGenerationResult result = generateWithFallback(request, config); + + // The conversation may have been deleted while the provider was + // blocking (~120s). Gate the entire post-provider tail — status + // update, broadcast, persistence — so we never write to a + // tombstoned conversation regardless of whether the provider + // succeeded or failed. + if (asyncTaskService.isConversationCanceled(conversationId)) { + log.info("[Music] Task {} (success={}) aborted: conversation {} was deleted", + task.getTaskId(), result.isSuccess(), conversationId); + return; + } + if (!result.isSuccess()) { asyncTaskService.updateStatus(task.getTaskId(), "failed", null, null, result.getErrorMessage()); @@ -118,16 +130,6 @@ public class MusicGenerationService { return; } - // Conversation may have been deleted while the upstream provider was - // blocking (~120s). Persisting now would recreate the attachment - // directory we just wiped and INSERT a mate_message row pointing at - // a non-existent conversation. Drop the result silently. - if (asyncTaskService.isConversationCanceled(conversationId)) { - log.info("[Music] Task {} succeeded but conversation {} was deleted, dropping result", - task.getTaskId(), conversationId); - return; - } - String audioUrl = persistAudio(conversationId, task.getTaskId(), result); saveAssistantMessage(conversationId, audioUrl, result); @@ -153,6 +155,11 @@ public class MusicGenerationService { log.info("[Music] Task {} succeeded, audio at {}", task.getTaskId(), audioUrl); } catch (Exception e) { log.error("[Music] Task {} worker failed: {}", task.getTaskId(), e.getMessage(), e); + if (asyncTaskService.isConversationCanceled(conversationId)) { + log.info("[Music] Skipping failure status/broadcast for deleted conversation {}", + conversationId); + return; + } asyncTaskService.updateStatus(task.getTaskId(), "failed", null, null, "音乐生成异常: " + e.getMessage()); asyncTaskService.broadcastTaskEventWithData(task, "async_task_completed", diff --git a/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java index 667761b2..4b0397b8 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java @@ -151,6 +151,16 @@ public class VideoGenerationService { * 任务完成时的回写逻辑：下载视频 → 保存消息 → 广播 SSE */ private void handleCompletion(AsyncTaskEntity task, TaskPollResult result) { + // The conversation may have been deleted while the poller was running. + // Gate every post-completion side effect — file write, message save, + // success/failure broadcast — so we never write to a tombstoned + // conversation regardless of which sub-branch we'd take. + if (asyncTaskService.isConversationCanceled(task.getConversationId())) { + log.info("[VideoGen] Task {} (success={}) aborted: conversation {} was deleted", + task.getTaskId(), result.succeeded(), task.getConversationId()); + return; + } + if (result.succeeded()) { try { String videoUrl = result.videoUrl(); @@ -161,15 +171,6 @@ public class VideoGenerationService { return; } - // Conversation deletion can race with the final poll iteration; - // dropping the result avoids resurrecting attachment files and - // a dangling mate_message row. - if (asyncTaskService.isConversationCanceled(task.getConversationId())) { - log.info("[VideoGen] Task {} succeeded but conversation {} was deleted, dropping result", - task.getTaskId(), task.getConversationId()); - return; - } - // 下载视频到本地 Path localPath = fileDownloader.download(videoUrl, task.getConversationId(), task.getTaskId()); String servingUrl = fileDownloader.toServingUrl(task.getConversationId(), localPath); @@ -192,6 +193,11 @@ public class VideoGenerationService { } catch (Exception e) { log.error("[VideoGen] Completion handling failed for task {}: {}", task.getTaskId(), e.getMessage(), e); + if (asyncTaskService.isConversationCanceled(task.getConversationId())) { + log.info("[VideoGen] Skipping failure broadcast for deleted conversation {}", + task.getConversationId()); + return; + } asyncTaskService.broadcastTaskEvent(task, "async_task_completed", false, null, "视频下载或保存失败: " + e.getMessage()); }