opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(task): gate every worker completion branch with isConversationCanceled

Browse Source
This commit is contained in:
matevip 2026-05-06 17:40:47 +08:00
parent 01c8765c96
commit d58fd0c3fd
3 changed files with 47 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java
View File

@ -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());
}

27
mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java
View File

@ -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",

24
mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java
View File

@ -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());
}