mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-15 20:08:18 +08:00
fix(sse): distinguish stream_not_local vs completed on reconnect
This commit is contained in:
parent
0b55d5a227
commit
3f10553186
@ -100,11 +100,37 @@ public class ChatController {
|
|||||||
|
|
||||||
registerEmitterCallbacks(emitter, conversationId);
|
registerEmitterCallbacks(emitter, conversationId);
|
||||||
|
|
||||||
|
// Issue #17 — distinguish "stream truly completed on this node"
|
||||||
|
// from "stream is running on another node (multi-node deployment
|
||||||
|
// without sticky session)". They look identical from attach()'s
|
||||||
|
// boolean return, but the user-facing remediation is different.
|
||||||
|
boolean existsLocally = streamTracker.streamExistsOnThisNode(conversationId);
|
||||||
boolean attached = streamTracker.attach(conversationId, emitter);
|
boolean attached = streamTracker.attach(conversationId, emitter);
|
||||||
if (!attached) {
|
if (!attached) {
|
||||||
// 没有活跃的流(已完成或服务器重启后丢失),通知前端直接结束
|
|
||||||
try {
|
try {
|
||||||
sendEvent(emitter, "done", Map.of("status", "completed"));
|
if (existsLocally) {
|
||||||
|
// RunState exists here but is done — stream finished normally
|
||||||
|
sendEvent(emitter, "done", Map.of("status", "completed"));
|
||||||
|
} else {
|
||||||
|
// No RunState on this node. Either:
|
||||||
|
// (a) The stream finished long ago and was cleaned up, OR
|
||||||
|
// (b) Multi-node deployment routed this reconnect to a
|
||||||
|
// DIFFERENT node than the originating one. The CE
|
||||||
|
// build assumes single-instance (see ChatStreamTracker
|
||||||
|
// class javadoc and rfc-054 §0); LB must be configured
|
||||||
|
// for sticky session by conversationId.
|
||||||
|
// We can't tell (a) from (b) at this layer, so emit an
|
||||||
|
// explicit code the front-end can surface to operators.
|
||||||
|
log.info("SSE reconnect: no RunState locally for conversationId={} — " +
|
||||||
|
"either completed-and-cleaned or running on another node", conversationId);
|
||||||
|
sendEvent(emitter, "done", Map.of(
|
||||||
|
"status", "stream_not_local",
|
||||||
|
"message", "Stream is not active on this node. " +
|
||||||
|
"If you're running a multi-node deployment, " +
|
||||||
|
"verify the load balancer is configured for sticky " +
|
||||||
|
"session by conversationId. See deploy/multi-node-deployment.md."
|
||||||
|
));
|
||||||
|
}
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
log.warn("SSE reconnect done send error: {}", e.getMessage());
|
log.warn("SSE reconnect done send error: {}", e.getMessage());
|
||||||
}
|
}
|
||||||
|
|||||||
@ -25,6 +25,25 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
|||||||
* 一个后台 Flux 生产者持续产出事件,广播给所有 SseEmitter 订阅者并缓存到 buffer。
|
* 一个后台 Flux 生产者持续产出事件,广播给所有 SseEmitter 订阅者并缓存到 buffer。
|
||||||
* 新连接(重连)到来时,先回放 buffer,再接入实时流。
|
* 新连接(重连)到来时,先回放 buffer,再接入实时流。
|
||||||
*
|
*
|
||||||
|
* <h2>Single-instance assumption</h2>
|
||||||
|
* <p><strong>The {@link #runs} map is process-local memory.</strong> A reconnect
|
||||||
|
* request can only re-attach to a {@code RunState} that lives on the <em>same</em>
|
||||||
|
* JVM that originally created it. In a multi-node deployment behind a load
|
||||||
|
* balancer, the LB MUST be configured for sticky session by {@code conversationId}
|
||||||
|
* (Nginx {@code hash $arg_conversationId consistent;}, K8s Ingress
|
||||||
|
* cookie-based affinity, AWS ALB target-group stickiness, etc.).
|
||||||
|
*
|
||||||
|
* <p>This is an explicit CE constraint — see
|
||||||
|
* {@code rfcs/community/90-appendix/02-tech-debt-inventory.md §4.1} and
|
||||||
|
* {@code rfc-054 §0}. Cross-node SSE relay (Redis Stream / NATS / Kafka) is
|
||||||
|
* tracked under the EE roadmap.
|
||||||
|
*
|
||||||
|
* <p>Operator-facing diagnostics: callers should use
|
||||||
|
* {@link #streamExistsOnThisNode(String)} when distinguishing "stream finished
|
||||||
|
* normally" from "stream is on a different node" — both return {@code false}
|
||||||
|
* from {@link #attach(String, SseEmitter)} but mean very different things to
|
||||||
|
* the user.
|
||||||
|
*
|
||||||
* @author MateClaw Team
|
* @author MateClaw Team
|
||||||
*/
|
*/
|
||||||
@Slf4j
|
@Slf4j
|
||||||
@ -285,6 +304,24 @@ public class ChatStreamTracker {
|
|||||||
broadcast(conversationId, eventName, json);
|
broadcast(conversationId, eventName, json);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Diagnostic helper for the multi-node deployment edge case (issue #17):
|
||||||
|
* tells the caller whether a {@link RunState} for this conversation
|
||||||
|
* exists on <em>this</em> JVM at all (regardless of done state).
|
||||||
|
*
|
||||||
|
* <p>{@link #attach(String, SseEmitter)} returns {@code false} both when
|
||||||
|
* the stream finished normally <em>and</em> when no state exists on this
|
||||||
|
* node. Callers that need to distinguish those two cases (e.g. to send a
|
||||||
|
* different SSE event to the client) should consult this method first.
|
||||||
|
*
|
||||||
|
* @return {@code true} when a RunState exists locally for this
|
||||||
|
* conversationId; {@code false} when it never existed here OR was
|
||||||
|
* already cleaned up after completion
|
||||||
|
*/
|
||||||
|
public boolean streamExistsOnThisNode(String conversationId) {
|
||||||
|
return runs.containsKey(conversationId);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 将 emitter 附着到现有的运行中的流。
|
* 将 emitter 附着到现有的运行中的流。
|
||||||
* 先回放 buffer 中的全部事件,再加入订阅者列表接收后续实时事件。
|
* 先回放 buffer 中的全部事件,再加入订阅者列表接收后续实时事件。
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user