From 3f1055318684b32a1f1cb3ef074d55cc9835f9b8 Mon Sep 17 00:00:00 2001
From: matevip <matevip@mate.vip>
Date: Sun, 26 Apr 2026 08:32:44 +0800
Subject: [PATCH] fix(sse): distinguish stream_not_local vs completed on
 reconnect

---
 .../vip/mate/channel/web/ChatController.java  | 30 ++++++++++++++-
 .../mate/channel/web/ChatStreamTracker.java   | 37 +++++++++++++++++++
 2 files changed, 65 insertions(+), 2 deletions(-)

diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
index da6ec97f..7b4af081 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
@@ -100,11 +100,37 @@ public class ChatController {
 
             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);
             if (!attached) {
-                // 没有活跃的流（已完成或服务器重启后丢失），通知前端直接结束
                 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) {
                     log.warn("SSE reconnect done send error: {}", e.getMessage());
                 }
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java
index b109db13..c6e53236 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java
@@ -25,6 +25,25 @@ import java.util.concurrent.atomic.AtomicBoolean;
  * 一个后台 Flux 生产者持续产出事件，广播给所有 SseEmitter 订阅者并缓存到 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
  */
 @Slf4j
@@ -285,6 +304,24 @@ public class ChatStreamTracker {
         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 附着到现有的运行中的流。
      * 先回放 buffer 中的全部事件，再加入订阅者列表接收后续实时事件。