mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(webchat): close SSE streams on done or error
Close WebChat subscriber SSE connections when the logical stream reaches done/error, make the emitter timeout configurable, and ensure stale-run eviction closes subscribers instead of leaving clients waiting.
This commit is contained in:
倪程伟
GitHub
parent
ddb6a837ea
commit
e9c08edb3a
@ -1623,6 +1623,11 @@ public class ChatStreamTracker {
|
||||
}
|
||||
// 先清理资源再移除
|
||||
stopHeartbeat(entry.getKey());
|
||||
// Close subscriber SSE connections so an evicted run does not
|
||||
// leave clients hanging in silence until their own emitter
|
||||
// timeout (issue #586). Aligns the eviction path with the
|
||||
// close-out sequence forceRecycle() uses.
|
||||
closeSubscribers(entry.getKey());
|
||||
Disposable d = state.disposable;
|
||||
if (d != null && !d.isDisposed()) {
|
||||
d.dispose();
|
||||
@ -1781,6 +1786,42 @@ public class ChatStreamTracker {
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Close every live subscriber's SSE connection for this run.
|
||||
* <p>
|
||||
* 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.
|
||||
* <p>
|
||||
* 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.
|
||||
* <p>
|
||||
* 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
|
||||
|
||||
@ -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");
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p>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()");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user