+ * For the WebChat channel (issue #586), {@code done}/{@code error} is the + * logical end of the stream and downstream integrators reading the SSE + * stream by standard semantics ("read until the server closes") must see + * the connection actually close — otherwise a 5-second answer holds a + * backend connection pool slot for the full 10-minute SseEmitter timeout. + * The in-house web channel does NOT call this (it keeps the emitter open + * for reconnect + buffer replay of late {@code async_task_*} events); the + * close-on-done policy is channel-scoped, not global. + *
+ * Also the shared closing sequence invoked by {@link #cleanupStaleRuns()} + * on eviction so a forcibly-reclaimed run does not leave subscribers + * hanging in silence until their own timeout fires. + *
+ * Idempotent: safe to call when no run exists or subscribers are already + * empty. Each {@code em.complete()} is wrapped so one dead subscriber + * cannot abort the loop before later subscribers are closed. + */ + public void closeSubscribers(String conversationId) { + RunState state = runs.get(conversationId); + if (state == null) return; + synchronized (state.lock) { + for (SseEmitter em : state.subscribers) { + try { + em.complete(); + } catch (Exception ignored) { + // A subscriber that is already closed/errored must not + // prevent the rest from being closed. + } + } + state.subscribers.clear(); + } + } + /** * Force a wedged run to terminate. Used by the admin Live view's * "End it" action when the friendly stop has been observed not to take diff --git a/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java index 58468044..c27a37d7 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java @@ -103,6 +103,20 @@ public class WebChatController { @Value("${mateclaw.jwt.secret:MateClaw-JWT-Secret-Key-2024-Please-Change-In-Production}") private String visitorTokenSecret; + /** + * SseEmitter timeout (minutes) for WebChat SSE streams. Previously + * hardcoded to 10 minutes across three {@code new Utf8SseEmitter(...)} call + * sites; downstream integrators were forced to reason about a constant + * living in someone else's repo (issue #586). Configurable so operators + * have a documented knob, defaulting to the historical 10 minutes. + */ + @Value("${mateclaw.webchat.sse-timeout-minutes:10}") + private int webchatSseTimeoutMinutes; + + private long sseTimeoutMillis() { + return (long) webchatSseTimeoutMinutes * 60_000L; + } + private final ExecutorService sseExecutor = Executors.newCachedThreadPool(); /** @@ -115,7 +129,7 @@ public class WebChatController { @RequestBody WebChatRequest request) { // RFC-058 PR-1: Utf8SseEmitter 显式 charset=UTF-8，防止中文 SSE 乱码 - SseEmitter emitter = new Utf8SseEmitter(10 * 60 * 1000L); + SseEmitter emitter = new Utf8SseEmitter(sseTimeoutMillis()); // 验证 API Key 并获取关联的 Channel 配置 ChannelEntity channel = resolveChannel(apiKey); @@ -282,12 +296,21 @@ public class WebChatController { persistErr.getMessage()); } streamTracker.broadcast(conversationId, "done", "{\"status\":\"completed\"}"); + // WebChat is a pure-backend SSE channel with no re-attach + // endpoint: for third-party integrators reading until the + // server closes, `done` IS the end of the stream. Close the + // subscriber connections so a 5-second answer doesn't hold + // a downstream connection-pool slot for the full SseEmitter + // timeout (issue #586). The in-house web channel does NOT + // do this — it keeps emitters open for reconnect + replay. + streamTracker.closeSubscribers(conversationId); streamTracker.complete(conversationId); }) .doOnError(e -> { log.error("[WebChat] Stream error: {}", e.getMessage()); streamTracker.broadcast(conversationId, "error", "{\"message\":" + escapeJson(e.getMessage()) + "}"); + streamTracker.closeSubscribers(conversationId); streamTracker.complete(conversationId); }) .subscribe(); @@ -1187,7 +1210,7 @@ public class WebChatController { @RequestParam String visitorId, @RequestParam(required = false) String sessionId, @RequestParam String pendingId) { - SseEmitter emitter = new Utf8SseEmitter(10 * 60 * 1000L); + SseEmitter emitter = new Utf8SseEmitter(sseTimeoutMillis()); ChannelEntity channel = resolveChannel(apiKey); if (channel == null) { sendErrorAndComplete(emitter, "Invalid API Key"); @@ -1248,6 +1271,7 @@ public class WebChatController { broadcastApprovalResolved(conversationId, consumed); streamTracker.broadcast(conversationId, "done", "{\"status\":\"already_resolved\"}"); + streamTracker.closeSubscribers(conversationId); return; } @@ -1263,6 +1287,7 @@ public class WebChatController { pendingId); streamTracker.broadcast(conversationId, "done", "{\"status\":\"error\",\"message\":\"No agent bound to approval\"}"); + streamTracker.closeSubscribers(conversationId); return; } @@ -1333,12 +1358,16 @@ public class WebChatController { } streamTracker.broadcast(conversationId, "done", "{\"status\":\"completed\"}"); + // Close the WebChat SSE connection on the logical end of + // the replay stream — same rationale as /stream (issue #586). + streamTracker.closeSubscribers(conversationId); streamTracker.complete(conversationId); }) .doOnError(e -> { log.error("[WebChat] approve replay stream error: {}", e.getMessage()); streamTracker.broadcast(conversationId, "error", "{\"message\":" + escapeJson(e.getMessage()) + "}"); + streamTracker.closeSubscribers(conversationId); streamTracker.complete(conversationId); }) .subscribe(); @@ -1349,6 +1378,7 @@ public class WebChatController { streamTracker.broadcast(conversationId, "error", "{\"message\":" + escapeJson(e.getMessage()) + "}"); } catch (Exception ignored) {} + streamTracker.closeSubscribers(conversationId); streamTracker.complete(conversationId); } }); @@ -1404,7 +1434,7 @@ public class WebChatController { @RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken, @RequestParam String visitorId, @RequestParam(required = false) String sessionId) { - SseEmitter emitter = new Utf8SseEmitter(10 * 60 * 1000L); + SseEmitter emitter = new Utf8SseEmitter(sseTimeoutMillis()); ChannelEntity channel = resolveChannel(apiKey); if (channel == null) { sendErrorAndComplete(emitter, "Invalid API Key"); diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCloseSubscribersTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCloseSubscribersTest.java new file mode 100644 index 00000000..afd2c88a --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCloseSubscribersTest.java @@ -0,0 +1,137 @@ +package vip.mate.channel.web; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Pins {@link ChatStreamTracker#closeSubscribers(String)} and its wiring into + * the eviction path — issue #586. WebChat's {@code done}/{@code error} is the + * logical end of the stream; subscriber SSE connections must actually close so + * backend integrators reading "until the server closes" are not held for the + * full 10-minute SseEmitter timeout. The eviction path must do the same so a + * forcibly-reclaimed run does not leave subscribers in silence. + * + *
Completion is observed by the post-complete send() throwing + * {@code IllegalStateException: ResponseBodyEmitter has already completed} + * (the servlet container's onCompletion callback does not fire in a unit test + * without an async request, so we assert on the emitter's own state instead). + */ +class ChatStreamTrackerCloseSubscribersTest { + + private ChatStreamTracker newTracker() { + return new ChatStreamTracker(new ObjectMapper()); + } + + /** True when the emitter has been completed (a subsequent send() throws). */ + private static boolean isCompleted(SseEmitter emitter) { + try { + emitter.send(SseEmitter.event().data("probe")); + return false; + } catch (Exception e) { + return e.getMessage() != null && e.getMessage().contains("already completed"); + } + } + + @Test + @DisplayName("closeSubscribers completes every attached emitter") + void closesAllSubscribers() { + ChatStreamTracker tracker = newTracker(); + String cid = "close-all"; + tracker.register(cid); + + SseEmitter emA = new SseEmitter(); + SseEmitter emB = new SseEmitter(); + tracker.attach(cid, emA); + tracker.attach(cid, emB); + + tracker.closeSubscribers(cid); + + assertTrue(isCompleted(emA), "subscriber A must be completed"); + assertTrue(isCompleted(emB), "subscriber B must be completed"); + // subscribers list is cleared after close + assertEquals(0, tracker.getAllSnapshot().getFirst().subscriberCount()); + } + + @Test + @DisplayName("closeSubscribers is idempotent / safe when no run or no subscribers") + void closeSubscribersSafeWhenEmpty() { + ChatStreamTracker tracker = newTracker(); + // No run at all — must not throw. + assertDoesNotThrow(() -> tracker.closeSubscribers("never-registered")); + + tracker.register("no-subs"); + tracker.closeSubscribers("no-subs"); // no subscribers — no-op, no throw + // run still alive (closeSubscribers does NOT mark done) + assertEquals(0, tracker.getAllSnapshot().getFirst().subscriberCount()); + } + + @Test + @DisplayName("closeSubscribers does not mark the run done (run stays until complete())") + void closeSubscribersDoesNotMarkDone() { + ChatStreamTracker tracker = newTracker(); + String cid = "close-not-done"; + tracker.register(cid); + tracker.incrementFlux(cid); + SseEmitter em = new SseEmitter(); + tracker.attach(cid, em); + + tracker.closeSubscribers(cid); + + // The run is still registered — closeSubscribers only closes the SSE + // connections, it does not finalize the run lifecycle. That remains + // complete()'s job so the retention window for reconnect still applies. + assertEquals(1, tracker.getAllSnapshot().size()); + assertTrue(tracker.streamExistsOnThisNode(cid)); + } + + @Test + @DisplayName("Eviction closes subscriber emitters (not just disposes the Flux)") + void evictionClosesSubscribers() { + ChatStreamTracker tracker = new ChatStreamTracker(new ObjectMapper()); + tracker.setIdleTimeoutMinutesForTesting(5); + String cid = "evict-close"; + tracker.register(cid); + + SseEmitter em = new SseEmitter(); + tracker.attach(cid, em); + + // Backdate so the idle-eviction path fires. + tracker.backdateLastEventForTesting(cid, System.currentTimeMillis() - 6 * 60_000L); + tracker.cleanupStaleRuns(); + + assertTrue(isCompleted(em), + "eviction must close the subscriber emitter so the client is not " + + "left hanging in silence until its own timeout"); + assertFalse(tracker.hasRunStateForTesting(cid)); + } + + @Test + @DisplayName("One already-dead subscriber does not block the rest from being closed") + void closeSubscribersResilientToDeadEmitter() { + ChatStreamTracker tracker = newTracker(); + String cid = "resilient-close"; + tracker.register(cid); + + SseEmitter dead = new SseEmitter(); + // Force the dead emitter into a completed state so the complete() call + // inside closeSubscribers() throws on it — proving the loop survives. + dead.complete(); + SseEmitter live = new SseEmitter(); + tracker.attach(cid, dead); + tracker.attach(cid, live); + + tracker.closeSubscribers(cid); + + assertTrue(isCompleted(live), + "the live subscriber must still be closed even though a dead " + + "subscriber threw on complete()"); + } +}