diff --git a/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java b/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java index 77a028f6..54647533 100644 --- a/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java +++ b/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java @@ -6,11 +6,13 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; +import org.springframework.context.event.EventListener; import org.springframework.stereotype.Service; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.task.model.AsyncTaskEntity; import vip.mate.task.model.AsyncTaskInfo; import vip.mate.task.repository.AsyncTaskMapper; +import vip.mate.workspace.conversation.event.ConversationDeletedEvent; import jakarta.annotation.PreDestroy; import java.time.LocalDateTime; @@ -48,6 +50,33 @@ public class AsyncTaskService implements ApplicationRunner { /** 活跃轮询任务，key = taskId */ private final ConcurrentHashMap> activePolls = new ConcurrentHashMap<>(); + /** Reverse mapping taskId → conversationId so a {@link ConversationDeletedEvent} + * listener can cancel every poller belonging to the deleted conversation + * without scanning DB (the {@code mate_async_task} rows are gone by the + * time the after-commit event fires). Populated in {@link #startPolling}; + * cleared in {@link #cancelPolling}. */ + private final ConcurrentHashMap pollTaskToConv = new ConcurrentHashMap<>(); + + /** Conversations whose deletion has fanned out to this service. Workers + * consult {@link #isConversationCanceled} before persisting anything tied + * to a conversation — the music virtual-thread worker, image/video poll + * completion handlers, and any future provider-level callback are + * asynchronous and may finish AFTER the conversation row + attachment + * directory have already been wiped. Without this gate they would + * recreate the directory + a dangling {@code mate_message} row. + *

+ * Value = expiry epoch-ms. Entries older than {@link #CANCEL_RETENTION_MS} + * are reaped on each event and on each lookup so the map cannot grow + * without bound. The retention window is comfortably longer than + * {@link #MAX_POLL_DURATION_MINUTES} and the music worker's ~120s upstream + * HTTP timeout, so any in-flight worker for a deleted conversation will + * still see the cancel flag when it tries to write back. */ + private final ConcurrentHashMap canceledConversations = new ConcurrentHashMap<>(); + + /** 30 minutes — covers MAX_POLL_DURATION_MINUTES (15) + music worker's + * ~2 min upstream blocking call with comfortable headroom. */ + private static final long CANCEL_RETENTION_MS = 30L * 60 * 1000; + /** 每用户最多并行任务数 */ private static final int MAX_ACTIVE_TASKS_PER_USER = 3; @@ -181,6 +210,9 @@ public class AsyncTaskService implements ApplicationRunner { }, 3, POLL_INTERVAL_SECONDS, TimeUnit.SECONDS); activePolls.put(taskId, future); + if (task.getConversationId() != null) { + pollTaskToConv.put(taskId, task.getConversationId()); + } log.info("[AsyncTask] Started polling for task {} (interval={}s, timeout={}min)", taskId, POLL_INTERVAL_SECONDS, MAX_POLL_DURATION_MINUTES); } @@ -190,6 +222,49 @@ public class AsyncTaskService implements ApplicationRunner { if (future != null) { future.cancel(false); } + pollTaskToConv.remove(taskId); + } + + // ==================== Conversation-deleted fan-out ==================== + + /** + * Returns true if this conversation was deleted recently enough that any + * still-running async worker (music virtual-thread, image/video poll + * completion, …) must abort before writing a file or persisting a + * message — see {@link #canceledConversations}. + *

+ * Sweeps stale entries on read so the map stays small. + */ + public boolean isConversationCanceled(String conversationId) { + if (conversationId == null) return false; + sweepCanceled(); + return canceledConversations.containsKey(conversationId); + } + + @EventListener + public void onConversationDeleted(ConversationDeletedEvent event) { + String convId = event.conversationId(); + if (convId == null) return; + + canceledConversations.put(convId, System.currentTimeMillis() + CANCEL_RETENTION_MS); + + int cancelled = 0; + for (Map.Entry entry : pollTaskToConv.entrySet()) { + if (convId.equals(entry.getValue())) { + cancelPolling(entry.getKey()); + cancelled++; + } + } + if (cancelled > 0) { + log.info("[AsyncTask] Cancelled {} active poller(s) for deleted conversation {}", + cancelled, convId); + } + sweepCanceled(); + } + + private void sweepCanceled() { + long now = System.currentTimeMillis(); + canceledConversations.entrySet().removeIf(e -> e.getValue() < now); } // ==================== 状态更新 ==================== 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 44751781..583d3789 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 @@ -231,6 +231,15 @@ 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); 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 98c581f0..4900cf1b 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 @@ -118,6 +118,16 @@ 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); 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 4a282ee2..667761b2 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 @@ -161,6 +161,15 @@ 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);