From a01cdae9719de05227169f8007478a144fd668f6 Mon Sep 17 00:00:00 2001
From: Austin <Austin@users.noreply.github.com>
Date: Wed, 16 Sep 2026 13:22:47 +0800
Subject: [PATCH] =?UTF-8?q?fix(realtime):=20evict=20fanout=20=E5=A2=9E?=
 =?UTF-8?q?=E5=8A=A0=E6=97=B6=E6=95=88=E6=A0=A1=E9=AA=8C=EF=BC=8C=E9=98=B2?=
 =?UTF-8?q?=E6=AD=A2=E5=BB=B6=E8=BF=9F=E9=87=8D=E6=94=BE=E8=AF=AF=E8=B8=A2?=
 =?UTF-8?q?=E6=96=B0=E7=99=BB=E5=BD=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

deliverRealtimeSessionControl 分发前校验 Ctrl.Timestamp，超过 10s 的 session-control（evict）消息直接丢弃并记 skip_stale 指标，避免 consumer 重启重放 PEL / stream 积压时旧消息的 skipSid 与新会话不符导致新登录被误踢。无时间戳消息放行以兼容旧发布方。

验证：TestRealtime 3/3、regression_kick_notification 14/14、regression_mobile_kick 21/21。
---
 server/realtime_dispatcher.go      | 23 +++++++++++++
 server/realtime_dispatcher_test.go | 53 ++++++++++++++++++++++++++++++
 2 files changed, 76 insertions(+)

diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go
index 2668c88..e917e68 100644
--- a/server/realtime_dispatcher.go
+++ b/server/realtime_dispatcher.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/tinode/chat/server/logs"
 	immetrics "github.com/tinode/chat/server/metrics"
@@ -223,11 +224,33 @@ func deliverRealtimeTopicMessage(envelope *store.RealtimeFanoutEnvelope, msg *Se
 	recordRealtimeFanoutDeliverResult(envelope.Kind, realtimeTargetTopic, delivered)
 }
 
+// fanoutSessionControlStaleLimit caps how long after publication a
+// session-control (evict) fanout message remains actionable. Delayed replays
+// (consumer backlog, pending-entry redelivery after restart) carry the original
+// skipSid, so a replayed evict would otherwise terminate freshly established
+// sessions whose sid differs from the stale skipSid (新登录被旧消息误踢).
+const fanoutSessionControlStaleLimit = 10 * time.Second
+
+// fanoutSessionControlStale reports whether the session-control fanout message
+// is too old to act on. Messages without a timestamp (unknown origin) are
+// delivered to stay compatible with legacy publishers.
+func fanoutSessionControlStale(msg *ServerComMessage) bool {
+	if msg == nil || msg.Ctrl == nil || msg.Ctrl.Timestamp.IsZero() {
+		return false
+	}
+	return time.Since(msg.Ctrl.Timestamp) > fanoutSessionControlStaleLimit
+}
+
 func deliverRealtimeSessionControl(envelope *store.RealtimeFanoutEnvelope, msg *ServerComMessage) {
 	if globals.sessionStore == nil {
 		immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_no_session_store")
 		return
 	}
+	if fanoutSessionControlStale(msg) {
+		immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_stale")
+		logs.Warn.Printf("redis realtime fanout: dropped stale session-control message target=%s older=%s", envelope.Target, fanoutSessionControlStaleLimit)
+		return
+	}
 	switch envelope.TargetType {
 	case realtimeTargetDevice:
 		uidText, deviceID, ok := strings.Cut(envelope.Target, ":")
diff --git a/server/realtime_dispatcher_test.go b/server/realtime_dispatcher_test.go
index 9404d75..88cce72 100644
--- a/server/realtime_dispatcher_test.go
+++ b/server/realtime_dispatcher_test.go
@@ -103,6 +103,52 @@ func TestRealtimeP2PTopicFallbackRewritesVisibleTopic(t *testing.T) {
 	}
 }
 
+func TestRealtimeSessionControlStaleEvictDropped(t *testing.T) {
+	oldStore := globals.sessionStore
+	defer func() { globals.sessionStore = oldStore }()
+
+	tenantID := types.TenantID(1)
+	uid := types.Uid(1001)
+	globals.sessionStore = realtimeTestSessionStore(
+		realtimeTestPlatformSession("sid-live", tenantID, uid, "web"),
+	)
+
+	envelope := func(ts time.Time) *store.RealtimeFanoutEnvelope {
+		msg := &ServerComMessage{
+			Ctrl: &MsgServerCtrl{
+				Code:      205,
+				Text:      "evicted",
+				Timestamp: ts,
+				Params:    map[string]any{"reason": "replaced_by_new_login"},
+			},
+		}
+		raw, err := json.Marshal(msg)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return &store.RealtimeFanoutEnvelope{
+			TenantID:   tenantID,
+			Kind:       realtimeKindSessionControl,
+			TargetType: realtimeTargetPlatf,
+			Target:     uid.UserId() + ":web",
+			SkipSid:    "sid-new",
+			Message:    raw,
+		}
+	}
+
+	// A stale replay (old skipSid != current session sid) must not evict.
+	deliverRealtimeFanoutEnvelope(envelope(time.Now().Add(-2 * fanoutSessionControlStaleLimit)))
+	if globals.sessionStore.Get("sid-live") == nil {
+		t.Fatal("stale evict must not terminate the session")
+	}
+
+	// A fresh evict must evict as usual.
+	deliverRealtimeFanoutEnvelope(envelope(time.Now()))
+	if globals.sessionStore.Get("sid-live") != nil {
+		t.Fatal("fresh evict must terminate the session")
+	}
+}
+
 func realtimeTestSessionStore(sessions ...*Session) *SessionStore {
 	ss := NewSessionStore(time.Minute)
 	for _, sess := range sessions {
@@ -117,6 +163,7 @@ func realtimeTestSession(sid string, tenantID types.TenantID, uid types.Uid, sub
 		tenantID: tenantID,
 		uid:      uid,
 		send:     make(chan any, 4),
+		stop:     make(chan any, 1),
 		subs:     make(map[string]*Subscription),
 	}
 	if sub != "" {
@@ -124,3 +171,9 @@ func realtimeTestSession(sid string, tenantID types.TenantID, uid types.Uid, sub
 	}
 	return sess
 }
+
+func realtimeTestPlatformSession(sid string, tenantID types.TenantID, uid types.Uid, platf string) *Session {
+	sess := realtimeTestSession(sid, tenantID, uid, "")
+	sess.platf = platf
+	return sess
+}