fix(server): coalesce platform evict fanout

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))
+		recordRealtimeSessionControlEvict(envelope, globals.sessionStore.EvictUserDevice(types.ParseUserId(uidText), deviceID, fanoutEvictReason(msg)))
-		immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
 	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))
+		recordRealtimeSessionControlEvict(envelope, globals.sessionStore.EvictUserPlatform(types.ParseUserId(uidText), platf, envelope.SkipSid, fanoutEvictReason(msg)))
-		immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
 	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))
+		recordRealtimeSessionControlEvict(envelope, globals.sessionStore.EvictUser(uid, envelope.SkipSid, fanoutEvictReason(msg)))
-		immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
 	}
 	_ = 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")

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 }()

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.