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 +}