From f24133f5def99de4729d230d201014d61ca74fd4 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 8 Oct 2026 00:23:12 +0800 Subject: [PATCH] fix(server): coalesce platform evict fanout --- server/realtime_dispatcher.go | 93 ++++++++++++++++++++++++------ server/realtime_dispatcher_test.go | 66 +++++++++++++++++++++ server/sessionstore.go | 5 +- 3 files changed, 145 insertions(+), 19 deletions(-) diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go index ec8faf0..cb252ba 100644 --- a/server/realtime_dispatcher.go +++ b/server/realtime_dispatcher.go @@ -5,6 +5,7 @@ import ( "encoding/json" "strconv" "strings" + "sync" "time" "github.com/tinode/chat/server/auth" @@ -27,6 +28,29 @@ const ( realtimeTargetUsers = "users" ) +const platformEvictFanoutDebounce = 750 * time.Millisecond + +type platformEvictFanoutKey struct { + tenantID types.TenantID + uid types.Uid + platf string + reason string +} + +type platformEvictFanoutPending struct { + skipSid string + timer *time.Timer +} + +var platformEvictFanout = struct { + sync.Mutex + pending map[platformEvictFanoutKey]*platformEvictFanoutPending +}{ + pending: make(map[platformEvictFanoutKey]*platformEvictFanoutPending), +} + +var publishRealtimeFanout = store.PublishRealtimeFanout + type realtimeGroupUnreadRecipient struct { User string `json:"user"` Source string `json:"source"` @@ -69,7 +93,7 @@ func publishRealtimeTopicMessage(topic string, msg *ServerComMessage) { return } dedupKey := realtimeMessageDedupKey(topic, msg) - if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetTopic, topic, msg.SkipSid, dedupKey, msg); err != nil { + if err := publishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetTopic, topic, msg.SkipSid, dedupKey, msg); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetTopic, "error") logs.Warn.Printf("redis realtime fanout: publish topic=%s failed: %v", topic, err) return @@ -89,7 +113,7 @@ func publishRealtimeUserMessage(uid types.Uid, msg *ServerComMessage) { return } dedupKey := realtimeMessageDedupKey(uid.UserId(), msg) - if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetUser, uid.UserId(), msg.SkipSid, dedupKey, msg); err != nil { + if err := publishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetUser, uid.UserId(), msg.SkipSid, dedupKey, msg); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "error") logs.Warn.Printf("redis realtime fanout: publish user=%s failed: %v", uid.UserId(), err) return @@ -109,7 +133,7 @@ func publishRealtimeGroupUnread(tenantID types.TenantID, topic, actor string, se Recipients: recipients, } dedupKey := "group-unread:" + topic + ":" + strconv.FormatInt(seqID, 10) - if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindGroupUnread, + if err := publishRealtimeFanout(currentNodeID(), tenantID, realtimeKindGroupUnread, realtimeTargetUsers, topic, "", dedupKey, payload); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindGroupUnread, realtimeTargetUsers, "error") logs.Warn.Printf("redis realtime fanout: publish group unread topic=%s seq=%d failed: %v", topic, seqID, err) @@ -140,7 +164,7 @@ func publishRealtimeCallEvent(msg *ClientComMessage) { Timestamp: msg.Timestamp, } dedupKey := "call:" + msg.RcptTo + ":" + msg.AsUser + ":" + event.Event + ":" + strconv.FormatInt(msg.Note.SeqId, 10) - if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindCallEvent, realtimeTargetTopic, msg.RcptTo, sid, dedupKey, event); err != nil { + if err := publishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindCallEvent, realtimeTargetTopic, msg.RcptTo, sid, dedupKey, event); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindCallEvent, realtimeTargetTopic, "error") logs.Warn.Printf("redis realtime fanout: publish call event topic=%s from=%s event=%s seq=%d failed: %v", msg.RcptTo, msg.AsUser, event.Event, msg.Note.SeqId, err) @@ -162,7 +186,7 @@ func publishRealtimeEvictUser(tenantID types.TenantID, uid types.Uid, reason str msg.Ctrl.Params = map[string]any{"reason": reason} } dedupKey := "evict:user:" + uid.UserId() + ":" + reason - if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetUser, uid.UserId(), skipSid, dedupKey, msg); err != nil { + if err := publishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetUser, uid.UserId(), skipSid, dedupKey, msg); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetUser, "error") logs.Warn.Printf("redis realtime fanout: publish evict user=%s failed: %v", uid.UserId(), err) return @@ -183,7 +207,7 @@ func publishRealtimeEvictDevice(tenantID types.TenantID, uid types.Uid, deviceID } target := uid.UserId() + ":" + strings.TrimSpace(deviceID) dedupKey := "evict:device:" + target + ":" + reason - if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetDevice, target, "", dedupKey, msg); err != nil { + if err := publishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetDevice, target, "", dedupKey, msg); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetDevice, "error") logs.Warn.Printf("redis realtime fanout: publish evict device=%s failed: %v", target, err) return @@ -191,24 +215,52 @@ func publishRealtimeEvictDevice(tenantID types.TenantID, uid types.Uid, deviceID immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetDevice, "ok") } -// publishRealtimeEvictUserPlatform evicts all sessions of the user on the given -// platform across cluster nodes. skipSid (the new login session) is part of the -// dedup key because repeated logins on the same platform within the dedup window -// must each reach the subscriber nodes. func publishRealtimeEvictUserPlatform(tenantID types.TenantID, uid types.Uid, platf, reason, skipSid string) { if !tenantID.IsValid() || uid.IsZero() || strings.TrimSpace(platf) == "" { immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "skip_invalid") return } + key := platformEvictFanoutKey{ + tenantID: tenantID, + uid: uid, + platf: strings.TrimSpace(platf), + reason: reason, + } + platformEvictFanout.Lock() + if pending := platformEvictFanout.pending[key]; pending != nil { + pending.skipSid = skipSid + pending.timer.Reset(platformEvictFanoutDebounce) + platformEvictFanout.Unlock() + immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "coalesced") + return + } + pending := &platformEvictFanoutPending{skipSid: skipSid} + pending.timer = time.AfterFunc(platformEvictFanoutDebounce, func() { + platformEvictFanout.Lock() + current := platformEvictFanout.pending[key] + if current != pending { + platformEvictFanout.Unlock() + return + } + delete(platformEvictFanout.pending, key) + finalSkipSid := pending.skipSid + platformEvictFanout.Unlock() + publishRealtimeEvictUserPlatformNow(key.tenantID, key.uid, key.platf, key.reason, finalSkipSid) + }) + platformEvictFanout.pending[key] = pending + platformEvictFanout.Unlock() +} + +func publishRealtimeEvictUserPlatformNow(tenantID types.TenantID, uid types.Uid, platf, reason, skipSid string) { msg := NoErrEvicted("", "", types.TimeNow()) msg.TenantID = tenantID msg.AsUser = uid.UserId() if reason != "" { msg.Ctrl.Params = map[string]any{"reason": reason} } - target := uid.UserId() + ":" + strings.TrimSpace(platf) + target := uid.UserId() + ":" + platf dedupKey := "evict:platform:" + target + ":" + skipSid - if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetPlatf, target, skipSid, dedupKey, msg); err != nil { + if err := publishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetPlatf, target, skipSid, dedupKey, msg); err != nil { immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "error") logs.Warn.Printf("redis realtime fanout: publish evict platform=%s failed: %v", target, err) return @@ -472,28 +524,33 @@ func deliverRealtimeSessionControl(envelope *store.RealtimeFanoutEnvelope, msg * immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target") return } - globals.sessionStore.EvictUserDevice(types.ParseUserId(uidText), deviceID, fanoutEvictReason(msg)) - immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted") + recordRealtimeSessionControlEvict(envelope, globals.sessionStore.EvictUserDevice(types.ParseUserId(uidText), deviceID, fanoutEvictReason(msg))) case realtimeTargetPlatf: uidText, platf, ok := strings.Cut(envelope.Target, ":") if !ok { immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target") return } - globals.sessionStore.EvictUserPlatform(types.ParseUserId(uidText), platf, envelope.SkipSid, fanoutEvictReason(msg)) - immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted") + recordRealtimeSessionControlEvict(envelope, globals.sessionStore.EvictUserPlatform(types.ParseUserId(uidText), platf, envelope.SkipSid, fanoutEvictReason(msg))) case realtimeTargetUser: uid := types.ParseUserId(envelope.Target) if uid.IsZero() { immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target") return } - globals.sessionStore.EvictUser(uid, envelope.SkipSid, fanoutEvictReason(msg)) - immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted") + recordRealtimeSessionControlEvict(envelope, globals.sessionStore.EvictUser(uid, envelope.SkipSid, fanoutEvictReason(msg))) } _ = msg } +func recordRealtimeSessionControlEvict(envelope *store.RealtimeFanoutEnvelope, evicted int) { + if evicted > 0 { + immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted") + return + } + immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "no_local_session") +} + func recordRealtimeFanoutDeliverResult(kind, targetType string, delivered int) { if delivered > 0 { immetrics.IncRealtimeFanout("deliver", kind, targetType, "local_delivered") diff --git a/server/realtime_dispatcher_test.go b/server/realtime_dispatcher_test.go index 5aaeb7f..fa82f3d 100644 --- a/server/realtime_dispatcher_test.go +++ b/server/realtime_dispatcher_test.go @@ -260,6 +260,61 @@ func TestRealtimeSessionControlStaleEvictDropped(t *testing.T) { } } +func TestRealtimePlatformEvictFanoutCoalescesToLatestSession(t *testing.T) { + oldPublish := publishRealtimeFanout + t.Cleanup(func() { + publishRealtimeFanout = oldPublish + resetPlatformEvictFanoutForTest() + }) + resetPlatformEvictFanoutForTest() + + type call struct { + target string + skipSid string + dedupKey string + } + calls := make(chan call, 3) + publishRealtimeFanout = func(_ string, _ types.TenantID, kind, targetType, target, skipSid, dedupKey string, _ any) error { + if kind == realtimeKindSessionControl && targetType == realtimeTargetPlatf { + calls <- call{target: target, skipSid: skipSid, dedupKey: dedupKey} + } + return nil + } + + tenantID := types.TenantID(1) + uid := types.Uid(1001) + publishRealtimeEvictUserPlatform(tenantID, uid, "web", "replaced_by_new_login", "sid-old") + publishRealtimeEvictUserPlatform(tenantID, uid, "web", "replaced_by_new_login", "sid-new") + publishRealtimeEvictUserPlatform(tenantID, uid, "web", "replaced_by_new_login", "sid-final") + + select { + case got := <-calls: + t.Fatalf("evict fanout published before debounce: %#v", got) + case <-time.After(platformEvictFanoutDebounce / 2): + } + + select { + case got := <-calls: + if got.target != uid.UserId()+":web" { + t.Fatalf("target = %q", got.target) + } + if got.skipSid != "sid-final" { + t.Fatalf("skipSid = %q, want latest", got.skipSid) + } + if got.dedupKey != "evict:platform:"+uid.UserId()+":web:sid-final" { + t.Fatalf("dedupKey = %q", got.dedupKey) + } + case <-time.After(2 * platformEvictFanoutDebounce): + t.Fatal("timed out waiting for coalesced evict fanout") + } + + select { + case got := <-calls: + t.Fatalf("expected one coalesced fanout, got extra %#v", got) + default: + } +} + func TestRealtimeP2PTopicFallbackRewritesVisibleInfoTopic(t *testing.T) { oldStore := globals.sessionStore oldHub := globals.hub @@ -314,6 +369,17 @@ func TestRealtimeP2PTopicFallbackRewritesVisibleInfoTopic(t *testing.T) { } } +func resetPlatformEvictFanoutForTest() { + platformEvictFanout.Lock() + defer platformEvictFanout.Unlock() + for key, pending := range platformEvictFanout.pending { + if pending.timer != nil { + pending.timer.Stop() + } + delete(platformEvictFanout.pending, key) + } +} + func TestRealtimeCallEventRoutesToLocalTopic(t *testing.T) { oldHub := globals.hub defer func() { globals.hub = oldHub }() diff --git a/server/sessionstore.go b/server/sessionstore.go index 5eaf672..0bfe036 100644 --- a/server/sessionstore.go +++ b/server/sessionstore.go @@ -338,7 +338,7 @@ func (ss *SessionStore) Shutdown() { } // EvictUser terminates all sessions of a given user. -func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string, reason ...string) { +func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string, reason ...string) int { ss.lock.Lock() defer ss.lock.Unlock() @@ -348,6 +348,7 @@ func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string, reason ...strin if len(reason) > 0 && reason[0] != "" { evicted.Ctrl.Params = map[string]any{"reason": reason[0]} } + count := 0 for _, s := range ss.sessCache { if s.uid == uid && !s.isMultiplex() && s.sid != skipSid { _, data := s.serialize(evicted) @@ -356,10 +357,12 @@ func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string, reason ...strin if s.proto == LPOLL { ss.lru.Remove(s.lpTracker) } + count++ } } statsSet("LiveSessions", int64(len(ss.sessCache))) + return count } // EvictUserDevice terminates all sessions of a given user connected with a device.