fix(approval): clear in-memory pendingMap on conversation delete via domain event

This commit is contained in:
matevip 2026-05-06 16:45:23 +08:00
parent 73d69c1dfe
commit bf8a533b19
4 changed files with 82 additions and 2 deletions

View File

@ -92,6 +92,34 @@ public class ApprovalService {
pendingMap.remove(pendingId); pendingMap.remove(pendingId);
} }
/**
* INTERNAL drop every map entry tied to the given conversation in one pass.
* Used by {@link ApprovalWorkflowService}'s {@code ConversationDeletedEvent}
* listener to clear residue once the {@code mate_tool_approval} rows for the
* conversation have already been deleted by the cascade. Without this, a
* still-PENDING entry (or any not yet GC'd resolved entry) would survive in
* the map until TTL eviction, and {@code findPendingByConversation} would
* keep handing out a ghost approval that points at a non-existent
* conversation row.
* <p>
* Only {@code ApprovalWorkflowService} should call this.
*
* @return number of entries removed
*/
int removeAllByConversation(String conversationId) {
if (conversationId == null) return 0;
int removed = 0;
var iter = pendingMap.entrySet().iterator();
while (iter.hasNext()) {
var entry = iter.next();
if (conversationId.equals(entry.getValue().getConversationId())) {
iter.remove();
removed++;
}
}
return removed;
}
/** /**
* INTERNAL register a {@link PendingApproval} reconstructed from DB during JVM startup. * INTERNAL register a {@link PendingApproval} reconstructed from DB during JVM startup.
* Bypasses id generation and pre-existing-entry checks; the snapshot's {@code pendingId} * Bypasses id generation and pre-existing-entry checks; the snapshot's {@code pendingId}

View File

@ -10,6 +10,7 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner; import org.springframework.boot.ApplicationRunner;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order; import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
@ -22,6 +23,7 @@ import vip.mate.approval.repository.ToolApprovalMapper;
import vip.mate.tool.guard.model.GuardEvaluation; import vip.mate.tool.guard.model.GuardEvaluation;
import vip.mate.tool.guard.model.GuardFinding; import vip.mate.tool.guard.model.GuardFinding;
import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
import java.time.Instant; import java.time.Instant;
import java.time.LocalDateTime; import java.time.LocalDateTime;
@ -82,6 +84,26 @@ public class ApprovalWorkflowService implements ApplicationRunner {
} }
} }
/**
* Drop in-memory approval state for a deleted conversation. The cascade in
* {@link ConversationService#deleteConversation} already removed the
* {@code mate_tool_approval} rows; this listener clears the parallel
* {@code pendingMap} entries so {@code findPendingByConversation} cannot
* keep returning a ghost approval that points at a non-existent
* conversation row.
* <p>
* Runs after the DB cascade commits see
* {@link ConversationDeletedEvent}.
*/
@EventListener
public void onConversationDeleted(ConversationDeletedEvent event) {
int removed = approvalService.removeAllByConversation(event.conversationId());
if (removed > 0) {
log.info("[ApprovalWorkflow] Dropped {} in-memory pending entries for deleted conversation {}",
removed, event.conversationId());
}
}
/** /**
* Reconstruct in-memory pending approvals from DB at startup, preserving the * Reconstruct in-memory pending approvals from DB at startup, preserving the
* original {@code pendingId} and {@code createdAt} so subsequent resolve / GC * original {@code pendingId} and {@code createdAt} so subsequent resolve / GC

View File

@ -6,6 +6,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronization;
@ -20,6 +21,7 @@ import vip.mate.channel.model.ChannelSessionEntity;
import vip.mate.channel.repository.ChannelSessionMapper; import vip.mate.channel.repository.ChannelSessionMapper;
import vip.mate.task.model.AsyncTaskEntity; import vip.mate.task.model.AsyncTaskEntity;
import vip.mate.task.repository.AsyncTaskMapper; import vip.mate.task.repository.AsyncTaskMapper;
import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
import vip.mate.workspace.conversation.model.ConversationEntity; import vip.mate.workspace.conversation.model.ConversationEntity;
import vip.mate.workspace.conversation.model.MessageContentPart; import vip.mate.workspace.conversation.model.MessageContentPart;
import vip.mate.workspace.conversation.model.MessageEntity; import vip.mate.workspace.conversation.model.MessageEntity;
@ -61,6 +63,7 @@ public class ConversationService {
private final ToolApprovalMapper toolApprovalMapper; private final ToolApprovalMapper toolApprovalMapper;
private final AsyncTaskMapper asyncTaskMapper; private final AsyncTaskMapper asyncTaskMapper;
private final ChannelSessionMapper channelSessionMapper; private final ChannelSessionMapper channelSessionMapper;
private final ApplicationEventPublisher eventPublisher;
/** /**
* 获取用户的会话列表返回 VO包含 agentName/agentIcon/status * 获取用户的会话列表返回 VO包含 agentName/agentIcon/status
@ -481,19 +484,29 @@ public class ConversationService {
conversationId, messages, approvals, asyncTasks, conversationId, messages, approvals, asyncTasks,
channelSessions, childrenUnlinked, conversations); channelSessions, childrenUnlinked, conversations);
registerAttachmentCleanupAfterCommit(conversationId); registerPostCommitCleanup(conversationId);
} }
private void registerAttachmentCleanupAfterCommit(String conversationId) { /**
* After-commit cleanup: file IO and the {@link ConversationDeletedEvent}
* fan-out both run only if the cascade actually persists, and an IO
* failure cannot roll back the DB cascade. The event lets approval and
* async-task modules drop their in-memory state (pendingMap, active
* pollers, canceled-conv set) so workers cannot resurrect orphan rows
* after the conversation row is gone.
*/
private void registerPostCommitCleanup(String conversationId) {
if (TransactionSynchronizationManager.isSynchronizationActive()) { if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override @Override
public void afterCommit() { public void afterCommit() {
cleanAttachmentFiles(conversationId); cleanAttachmentFiles(conversationId);
eventPublisher.publishEvent(new ConversationDeletedEvent(conversationId));
} }
}); });
} else { } else {
cleanAttachmentFiles(conversationId); cleanAttachmentFiles(conversationId);
eventPublisher.publishEvent(new ConversationDeletedEvent(conversationId));
} }
} }

View File

@ -0,0 +1,17 @@
package vip.mate.workspace.conversation.event;
/**
* Fired AFTER {@link vip.mate.workspace.conversation.ConversationService#deleteConversation}
* commits its DB cascade.
* <p>
* Subscribers must use this to clean up any in-memory or scheduled state keyed
* on the deleted conversation e.g. the approval pending map, async-task
* pollers, SSE buffers, anything that survives independently of the DB row.
* <p>
* Published from a {@code TransactionSynchronization.afterCommit} hook so that
* a listener observing this event can safely assume the conversation row,
* its messages, and every cascaded associate row are gone. If the transaction
* rolls back, the event is never published.
*/
public record ConversationDeletedEvent(String conversationId) {
}