fix: move group push work off topic loop

server/datamodel.go

@@ -382,6 +382,8 @@ type ClientComMessage struct {
 	MetaWhat int `json:"-"`
 	// Timestamp when this message was received by the server.
 	Timestamp time.Time `json:"-"`
+	// Timestamp when this message entered the current topic client queue.
+	queuedAt time.Time
 	// Tenant context inherited from the originating session.
 	TenantID types.TenantID `json:"-"`
 

server/hub.go

@@ -57,6 +57,12 @@ type userStatusReq struct {
 	state types.ObjState
 }
 
+type deferredPresenceKey struct {
+	tenantID types.TenantID
+	topic    string
+	source   string
+}
+
 // Hub is the core structure which holds topics.
 type Hub struct {
 
@@ -74,6 +80,9 @@ type Hub struct {
 
 	// Channel for routing server-generated messages, buffered at 4096
 	routeSrv chan *ServerComMessage
+	// Presence updates which could not immediately enter a topic queue.
+	deferredPresence     chan *ServerComMessage
+	deferredPresenceStop chan struct{}
 
 	// subscribe session to topic, possibly creating a new topic, buffered at 256
 	join chan *ClientComMessage
@@ -121,14 +130,16 @@ func newHub() *Hub {
 	h := &Hub{
 		topics: &sync.Map{},
 		// TODO: verify if these channels have to be buffered.
-		routeCli:   make(chan *ClientComMessage, 4096),
-		routeSrv:   make(chan *ServerComMessage, 4096),
-		join:       make(chan *ClientComMessage, 256),
-		unreg:      make(chan *topicUnreg, 256),
-		rehash:     make(chan bool),
-		meta:       make(chan *ClientComMessage, 128),
-		userStatus: make(chan *userStatusReq, 128),
-		shutdown:   make(chan chan<- bool),
+		routeCli:             make(chan *ClientComMessage, 4096),
+		routeSrv:             make(chan *ServerComMessage, 4096),
+		deferredPresence:     make(chan *ServerComMessage, 8192),
+		deferredPresenceStop: make(chan struct{}),
+		join:                 make(chan *ClientComMessage, 256),
+		unreg:                make(chan *topicUnreg, 256),
+		rehash:               make(chan bool),
+		meta:                 make(chan *ClientComMessage, 128),
+		userStatus:           make(chan *userStatusReq, 128),
+		shutdown:             make(chan chan<- bool),
 	}
 
 	statsRegisterInt("LiveTopics")
@@ -154,11 +165,70 @@ func newHub() *Hub {
 	statsRegisterHistogram("RequestLatency", requestLatencyDistribution)
 	statsRegisterHistogram("OutgoingMessageSize", outgoingMessageSizeDistribution)
 
+	go h.runDeferredPresence()
 	go h.run()
 
 	return h
 }
 
+func (h *Hub) deferPresence(msg *ServerComMessage) bool {
+	select {
+	case h.deferredPresence <- msg:
+		immetrics.ObserveInternalQueueEnqueue("topic_presence_deferred", len(h.deferredPresence), cap(h.deferredPresence), true)
+		return true
+	default:
+		immetrics.ObserveInternalQueueEnqueue("topic_presence_deferred", len(h.deferredPresence), cap(h.deferredPresence), false)
+		return false
+	}
+}
+
+func (h *Hub) runDeferredPresence() {
+	const maxPending = 16384
+	ticker := time.NewTicker(5 * time.Millisecond)
+	defer ticker.Stop()
+	pending := make(map[deferredPresenceKey]*ServerComMessage)
+
+	for {
+		select {
+		case msg := <-h.deferredPresence:
+			if msg == nil || msg.Pres == nil {
+				continue
+			}
+			key := deferredPresenceKey{tenantID: msg.TenantID, topic: msg.RcptTo, source: msg.Pres.Src}
+			if _, exists := pending[key]; exists || len(pending) < maxPending {
+				pending[key] = msg
+			} else {
+				immetrics.ObserveInternalQueueEnqueue("topic_presence_pending", len(pending), maxPending, false)
+				logs.Err.Printf("hub: deferred presence queue full topic=%s tenant=%d source=%s",
+					msg.RcptTo, msg.TenantID, msg.Pres.Src)
+			}
+			immetrics.SetInternalQueueCurrent("topic_presence_pending", len(pending))
+		case <-ticker.C:
+			budget := 64
+			for key, msg := range pending {
+				dst := h.topicGet(types.TopicKey{TenantID: key.tenantID, Topic: key.topic})
+				if dst == nil || dst.serverMsg == nil {
+					delete(pending, key)
+					continue
+				}
+				select {
+				case dst.serverMsg <- msg:
+					immetrics.ObserveInternalQueueEnqueue("topic_server", len(dst.serverMsg), cap(dst.serverMsg), true)
+					delete(pending, key)
+					budget--
+				default:
+				}
+				if budget == 0 {
+					break
+				}
+			}
+			immetrics.SetInternalQueueCurrent("topic_presence_pending", len(pending))
+		case <-h.deferredPresenceStop:
+			return
+		}
+	}
+}
+
 func (h *Hub) run() {
 	for {
 		select {
@@ -246,6 +316,7 @@ func (h *Hub) run() {
 			if dst := h.topicGet(key); dst != nil {
 				// Everything is OK, sending packet to known topic
 				if dst.clientMsg != nil {
+					msg.queuedAt = time.Now()
 					select {
 					case dst.clientMsg <- msg:
 						immetrics.ObserveInternalQueueEnqueue("topic_client", len(dst.clientMsg), cap(dst.clientMsg), true)
@@ -284,8 +355,10 @@ func (h *Hub) run() {
 				case dst.serverMsg <- msg:
 					immetrics.ObserveInternalQueueEnqueue("topic_server", len(dst.serverMsg), cap(dst.serverMsg), true)
 				default:
-					immetrics.ObserveInternalQueueEnqueue("topic_server", len(dst.serverMsg), cap(dst.serverMsg), false)
-					logs.Err.Println("hub: topic's broadcast queue is full", dst.name)
+					if msg.Pres == nil || !h.deferPresence(msg) {
+						immetrics.ObserveInternalQueueEnqueue("topic_server", len(dst.serverMsg), cap(dst.serverMsg), false)
+						logs.Err.Println("hub: topic's broadcast queue is full", dst.name)
+					}
 				}
 			} else if (strings.HasPrefix(msg.RcptTo, "usr") || strings.HasPrefix(msg.RcptTo, "grp")) &&
 				globals.cluster.isRemoteTopic(key) {
@@ -357,6 +430,9 @@ func (h *Hub) run() {
 			}
 
 			logs.Info.Printf("Hub shutdown completed with %d topics", topicCount)
+			if h.deferredPresenceStop != nil {
+				close(h.deferredPresenceStop)
+			}
 
 			// let the main goroutine know we are done with the cleanup
 			hubdone <- true

server/metrics/metrics.go

@@ -154,6 +154,10 @@ var (
 		Name: "im_internal_queue_capacity",
 		Help: "Configured internal message queue capacity by pod and queue.",
 	}, []string{"pod", "queue"})
+	InternalQueueCurrent = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "im_internal_queue_current",
+		Help: "Current entries in an internal deferred queue by pod and queue.",
+	}, []string{"pod", "queue"})
 	MessagePersistDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Name:    "im_message_persist_duration_seconds",
 		Help:    "Kafka message persistence duration by message kind and result.",
@@ -164,6 +168,34 @@ var (
 		Help:    "Message publish hot-path duration by message kind, stage, and result.",
 		Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
 	}, []string{"kind", "stage", "result"})
+	TopicQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_topic_queue_wait_duration_seconds",
+		Help:    "Time a client publish waited in a topic queue before processing by topic kind.",
+		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"pod", "kind"})
+	PushDispatchQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "im_push_dispatch_queue_enqueue_total",
+		Help: "Asynchronous push dispatch queue enqueue attempts by pod and result.",
+	}, []string{"pod", "result"})
+	PushDispatchQueueDepth = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "im_push_dispatch_queue_depth",
+		Help: "Current asynchronous push dispatch queue depth by pod.",
+	}, []string{"pod"})
+	PushDispatchQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_push_dispatch_queue_wait_duration_seconds",
+		Help:    "Time a push dispatch job waited before a worker started processing it.",
+		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"pod"})
+	PushDispatchDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_push_dispatch_duration_seconds",
+		Help:    "Asynchronous push presence resolution and dispatch duration by pod and result.",
+		Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"pod", "result"})
+	PushRecipientCount = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_push_recipient_count",
+		Help:    "Recipients in an asynchronously dispatched push receipt by pod.",
+		Buckets: []float64{1, 2, 10, 50, 100, 300, 600, 900, 1200, 2000},
+	}, []string{"pod"})
 	KafkaMessagePersistTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "im_kafka_message_persist_total",
 		Help: "Kafka message persistence outcomes by message kind and result.",
@@ -191,7 +223,9 @@ func init() {
 		RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength,
 		ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration,
 		ClusterRouteTotal, ClusterProxyResponseTotal, InternalQueueEnqueueTotal, InternalQueueDepth,
-		InternalQueueCapacity, MessagePersistDuration, MessagePublishStageDuration,
+		InternalQueueCapacity, InternalQueueCurrent, MessagePersistDuration, MessagePublishStageDuration, TopicQueueWaitDuration,
+		PushDispatchQueueEnqueueTotal, PushDispatchQueueDepth, PushDispatchQueueWaitDuration,
+		PushDispatchDuration, PushRecipientCount,
 		KafkaMessagePersistTotal)
 	Registry.MustRegister(KafkaConsumerLag, KafkaBacklogOldestAge, KafkaDeadLettersTotal)
 }
@@ -220,6 +254,39 @@ func ObserveMessagePublishStage(kind, stage string, started time.Time, err error
 		Observe(time.Since(started).Seconds())
 }
 
+func ObserveTopicQueueWait(kind string, queuedAt time.Time) {
+	if queuedAt.IsZero() {
+		return
+	}
+	TopicQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(kind)).Observe(time.Since(queuedAt).Seconds())
+}
+
+func ObservePushDispatchEnqueue(depth int, accepted bool) {
+	result := "ok"
+	if !accepted {
+		result = "full"
+	}
+	PushDispatchQueueEnqueueTotal.WithLabelValues(metricPodName(), result).Inc()
+	PushDispatchQueueDepth.WithLabelValues(metricPodName()).Set(float64(depth))
+}
+
+func SetPushDispatchQueueDepth(depth int) {
+	PushDispatchQueueDepth.WithLabelValues(metricPodName()).Set(float64(depth))
+}
+
+func ObservePushDispatch(queuedAt time.Time, recipients int, started time.Time, err error) {
+	pod := metricPodName()
+	if !queuedAt.IsZero() {
+		PushDispatchQueueWaitDuration.WithLabelValues(pod).Observe(started.Sub(queuedAt).Seconds())
+	}
+	result := "ok"
+	if err != nil {
+		result = "error"
+	}
+	PushDispatchDuration.WithLabelValues(pod, result).Observe(time.Since(started).Seconds())
+	PushRecipientCount.WithLabelValues(pod).Observe(float64(recipients))
+}
+
 func IncRealtimeFanout(stage, kind, targetType, status string) {
 	RealtimeFanoutEventsTotal.WithLabelValues(normalizeLabel(stage), normalizeLabel(kind),
 		normalizeLabel(targetType), normalizeLabel(status)).Inc()
@@ -245,6 +312,10 @@ func ObserveInternalQueueEnqueue(queue string, depth, capacity int, accepted boo
 	InternalQueueCapacity.WithLabelValues(pod, queue).Set(float64(capacity))
 }
 
+func SetInternalQueueCurrent(queue string, depth int) {
+	InternalQueueCurrent.WithLabelValues(metricPodName(), normalizeLabel(queue)).Set(float64(depth))
+}
+
 func ObserveWSWrite(frame, result, reason string, started time.Time, payloadBytes int) {
 	pod := metricPodName()
 	frame = normalizeLabel(frame)

server/monitoring_http_test.go

@@ -91,10 +91,14 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 	immetrics.KafkaConsumerLag.WithLabelValues("test", "topic", "group").Set(0)
 	immetrics.KafkaDeadLettersTotal.WithLabelValues("p2p", "test").Add(0)
 	immetrics.ObserveInternalQueueEnqueue("test", 0, 1, true)
+	immetrics.SetInternalQueueCurrent("test", 0)
 	immetrics.ObserveWSWrite("text", "ok", "none", time.Now(), 10)
 	immetrics.IncWSDisconnect("close_normal")
 	immetrics.ObserveWSSendQueueEnqueue(1, 160, true)
 	immetrics.ObserveWSSendQueueDequeue(time.Now(), false)
+	immetrics.ObserveTopicQueueWait("group", time.Now())
+	immetrics.ObservePushDispatchEnqueue(1, true)
+	immetrics.ObservePushDispatch(time.Now(), 1, time.Now(), nil)
 	wrt := httptest.NewRecorder()
 	serveMetrics(wrt, httptest.NewRequest(http.MethodGet, "/metrics", nil))
 
@@ -110,7 +114,10 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 		"im_ws_disconnect_total", "im_ws_send_queue_current", "im_ws_send_queue_capacity",
 		"im_ws_send_queue_enqueue_total", "im_ws_send_queue_depth", "im_ws_send_queue_wait_duration_seconds",
 		"im_cluster_p2m_enqueue_total", "im_cluster_rpc_duration_seconds", "im_cluster_route_total",
-		"im_internal_queue_enqueue_total", "im_internal_queue_depth", "im_internal_queue_capacity",
+		"im_internal_queue_enqueue_total", "im_internal_queue_depth", "im_internal_queue_capacity", "im_internal_queue_current",
+		"im_topic_queue_wait_duration_seconds", "im_push_dispatch_queue_enqueue_total",
+		"im_push_dispatch_queue_depth", "im_push_dispatch_queue_wait_duration_seconds",
+		"im_push_dispatch_duration_seconds", "im_push_recipient_count",
 		"im_kafka_message_persist_total", "im_kafka_consumer_lag", "im_kafka_dead_letters_total"} {
 		if !strings.Contains(body, "# HELP "+metric+" ") {
 			t.Fatalf("metric %s was not exported", metric)

server/push.go

@@ -8,14 +8,109 @@
 package main
 
 import (
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/push"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
 
+const (
+	pushDispatchQueueCapacity = 4096
+	pushDispatchWorkers       = 4
+)
+
+type pushDispatchJob struct {
+	rcpt            *push.Receipt
+	resolvePresence bool
+	queuedAt        time.Time
+}
+
+var asyncPushDispatcher struct {
+	once   sync.Once
+	queues [pushDispatchWorkers]chan pushDispatchJob
+	depth  atomic.Int64
+}
+
+func initAsyncPushDispatcher() {
+	asyncPushDispatcher.once.Do(func() {
+		for i := range pushDispatchWorkers {
+			asyncPushDispatcher.queues[i] = make(chan pushDispatchJob, pushDispatchQueueCapacity/pushDispatchWorkers)
+			go func(queue <-chan pushDispatchJob) {
+				for job := range queue {
+					depth := asyncPushDispatcher.depth.Add(-1)
+					immetrics.SetPushDispatchQueueDepth(int(depth))
+					processPushDispatch(job)
+				}
+			}(asyncPushDispatcher.queues[i])
+		}
+	})
+}
+
+func pushDispatchShard(rcpt *push.Receipt) int {
+	hash := uint32(2166136261)
+	for _, value := range []byte(rcpt.Payload.Topic) {
+		hash ^= uint32(value)
+		hash *= 16777619
+	}
+	hash ^= uint32(rcpt.TenantID)
+	hash *= 16777619
+	return int(hash % pushDispatchWorkers)
+}
+
+func dispatchPushAsync(rcpt *push.Receipt, resolvePresence bool) bool {
+	if rcpt == nil || globals.usersUpdate == nil {
+		return false
+	}
+	initAsyncPushDispatcher()
+	queue := asyncPushDispatcher.queues[pushDispatchShard(rcpt)]
+	job := pushDispatchJob{rcpt: rcpt, resolvePresence: resolvePresence, queuedAt: time.Now()}
+	depth := asyncPushDispatcher.depth.Add(1)
+	select {
+	case queue <- job:
+		immetrics.ObservePushDispatchEnqueue(int(asyncPushDispatcher.depth.Load()), true)
+		return true
+	default:
+		depth = asyncPushDispatcher.depth.Add(-1)
+		immetrics.ObservePushDispatchEnqueue(int(depth), false)
+		logs.Err.Printf("push dispatch queue full: tenant=%d topic=%s recipients=%d shard_len=%d shard_cap=%d total_depth=%d",
+			rcpt.TenantID, rcpt.Payload.Topic, len(rcpt.To), len(queue), cap(queue), depth)
+		return false
+	}
+}
+
+func processPushDispatch(job pushDispatchJob) {
+	started := time.Now()
+	var dispatchErr error
+	if job.resolvePresence {
+		unresolved := make([]types.Uid, 0, len(job.rcpt.To))
+		for uid, recipient := range job.rcpt.To {
+			if recipient.Delivered == 0 {
+				unresolved = append(unresolved, uid)
+			}
+		}
+		presences, err := store.GetUsersPresence(job.rcpt.TenantID, unresolved)
+		if err != nil {
+			dispatchErr = err
+			logs.Warn.Printf("push dispatch: failed to read presence topic=%s recipients=%d err=%v",
+				job.rcpt.Payload.Topic, len(unresolved), err)
+		}
+		for uid, presence := range presences {
+			if presence != nil && presence.Online && presence.SessionCount > 0 {
+				recipient := job.rcpt.To[uid]
+				recipient.Delivered = presence.SessionCount
+				job.rcpt.To[uid] = recipient
+			}
+		}
+	}
+	sendPush(job.rcpt)
+	immetrics.ObservePushDispatch(job.queuedAt, len(job.rcpt.To), started, dispatchErr)
+}
+
 // Subscribe or unsubscribe user to/from FCM topic (channel).
 func (t *Topic) channelSubUnsub(uid types.Uid, sub bool) {
 	push.ChannelSub(&push.ChannelReq{
@@ -75,15 +170,6 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR
 			// Make sure the sender's devices receive a silent push.
 			online = 1
 		}
-		if online == 0 {
-			presence, err := store.GetUserPresence(t.tenantID, uid)
-			if err != nil {
-				logs.Warn.Printf("topic[%s]: failed to read user presence for push: uid=%s err=%v", t.name, uid.UserId(), err)
-			} else if presence != nil && presence.Online && presence.SessionCount > 0 {
-				online = presence.SessionCount
-			}
-		}
-
 		// Send only to those who have notifications enabled.
 		mode := pud.modeWant & pud.modeGiven
 		if mode.IsPresencer() && mode.IsReader() && !pud.deleted && !pud.isChan {

server/push_async_test.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/tinode/chat/server/push"
+	"github.com/tinode/chat/server/store/types"
+)
+
+func TestDispatchPushAsyncDeliversToUserUpdater(t *testing.T) {
+	oldUsersUpdate := globals.usersUpdate
+	globals.usersUpdate = make(chan *UserCacheReq, 1)
+	t.Cleanup(func() { globals.usersUpdate = oldUsersUpdate })
+
+	uid := types.Uid(101)
+	rcpt := &push.Receipt{
+		TenantID: 1,
+		To:       map[types.Uid]push.Recipient{uid: {Delivered: 1}},
+		Payload:  push.Payload{What: push.ActMsg, Topic: "grpAsync"},
+	}
+	if !dispatchPushAsync(rcpt, false) {
+		t.Fatal("push dispatch was not enqueued")
+	}
+
+	select {
+	case req := <-globals.usersUpdate:
+		if req.PushRcpt != rcpt {
+			t.Fatal("user updater received the wrong receipt")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for asynchronous push dispatch")
+	}
+}
+
+func TestDeferredPresenceCoalescesAndRetries(t *testing.T) {
+	tenantID := types.TenantID(1)
+	topicName := "grpDeferred"
+	topic := &Topic{tenantID: tenantID, name: topicName, serverMsg: make(chan *ServerComMessage, 1)}
+	topic.serverMsg <- &ServerComMessage{TenantID: tenantID, RcptTo: topicName}
+
+	hub := &Hub{
+		topics:               &sync.Map{},
+		deferredPresence:     make(chan *ServerComMessage, 4),
+		deferredPresenceStop: make(chan struct{}),
+	}
+	hub.topicPut(types.TopicKey{TenantID: tenantID, Topic: topicName}, topic)
+	go hub.runDeferredPresence()
+	t.Cleanup(func() { close(hub.deferredPresenceStop) })
+
+	off := &ServerComMessage{TenantID: tenantID, RcptTo: topicName,
+		Pres: &MsgServerPres{What: "off", Src: "usrA"}}
+	on := &ServerComMessage{TenantID: tenantID, RcptTo: topicName,
+		Pres: &MsgServerPres{What: "on", Src: "usrA"}}
+	if !hub.deferPresence(off) || !hub.deferPresence(on) {
+		t.Fatal("failed to enqueue deferred presence")
+	}
+	time.Sleep(20 * time.Millisecond)
+	<-topic.serverMsg
+
+	select {
+	case got := <-topic.serverMsg:
+		if got.Pres == nil || got.Pres.What != "on" {
+			t.Fatalf("expected latest coalesced presence, got %#v", got.Pres)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for deferred presence retry")
+	}
+}

server/session.go

@@ -904,6 +904,7 @@ func (s *Session) publish(msg *ClientComMessage) {
 
 	if sub := s.getSub(msg.RcptTo); sub != nil {
 		// This is a post to a subscribed topic. The message is sent to the topic only
+		msg.queuedAt = time.Now()
 		select {
 		case sub.broadcast <- msg:
 			immetrics.ObserveInternalQueueEnqueue("session_topic_publish", len(sub.broadcast), cap(sub.broadcast), true)
@@ -1722,6 +1723,7 @@ func (s *Session) note(msg *ClientComMessage) {
 
 	if sub := s.getSub(msg.RcptTo); sub != nil {
 		// Pings can be sent to subscribed topics only
+		msg.queuedAt = time.Now()
 		select {
 		case sub.broadcast <- msg:
 		default:

server/store/group_message_runtime.go

@@ -111,6 +111,7 @@ type groupMessageRuntime interface {
 	RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error
 	MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error)
 	GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error)
+	GetUsersPresence(tenantID types.TenantID, uids []types.Uid) (map[types.Uid]*UserPresence, error)
 	MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error
 	MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error
 	GetTopicNodeOnlineMembers(tenantID types.TenantID, topic, nodeID string) ([]types.Uid, error)
@@ -339,6 +340,14 @@ func GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, err
 	return groupMsgRuntime.GetUserPresence(tenantID, uid)
 }
 
+func GetUsersPresence(tenantID types.TenantID, uids []types.Uid) (map[types.Uid]*UserPresence, error) {
+	result := make(map[types.Uid]*UserPresence, len(uids))
+	if groupMsgRuntime == nil || !tenantID.IsValid() || len(uids) == 0 {
+		return result, nil
+	}
+	return groupMsgRuntime.GetUsersPresence(tenantID, uids)
+}
+
 func UserPresenceEnabled() bool {
 	return groupMsgRuntime != nil
 }
@@ -1027,6 +1036,40 @@ func (r *redisGroupMessageRuntime) GetUserPresence(tenantID types.TenantID, uid
 	return presence, nil
 }
 
+func (r *redisGroupMessageRuntime) GetUsersPresence(tenantID types.TenantID, uids []types.Uid) (map[types.Uid]*UserPresence, error) {
+	result := make(map[types.Uid]*UserPresence, len(uids))
+	if len(uids) == 0 {
+		return result, nil
+	}
+	ctx := context.Background()
+	pipe := r.client.Pipeline()
+	commands := make(map[types.Uid]*redis.MapStringStringCmd, len(uids))
+	for _, uid := range uids {
+		if !uid.IsZero() {
+			commands[uid] = pipe.HGetAll(ctx, r.presenceUserKey(tenantID, uid))
+		}
+	}
+	if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil {
+		return nil, err
+	}
+	for uid, command := range commands {
+		raw, err := command.Result()
+		if err != nil && err != redis.Nil {
+			return nil, err
+		}
+		presence := &UserPresence{}
+		if len(raw) > 0 {
+			presence.Online = raw["online"] == "1"
+			presence.NodeID = raw["node_id"]
+			presence.SessionCount, _ = strconv.Atoi(raw["session_count"])
+			presence.LastActiveAt, _ = time.Parse(time.RFC3339Nano, raw["last_active_at"])
+			presence.LastSeenAt, _ = time.Parse(time.RFC3339Nano, raw["last_seen_at"])
+		}
+		result[uid] = presence
+	}
+	return result, nil
+}
+
 func (r *redisGroupMessageRuntime) MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error {
 	if nodeID == "" {
 		nodeID = "local"

server/store/group_message_runtime_test.go

@@ -127,6 +127,10 @@ func (r *testGroupMessageRuntime) GetUserPresence(types.TenantID, types.Uid) (*U
 	return nil, nil
 }
 
+func (r *testGroupMessageRuntime) GetUsersPresence(types.TenantID, []types.Uid) (map[types.Uid]*UserPresence, error) {
+	return map[types.Uid]*UserPresence{}, nil
+}
+
 func (r *testGroupMessageRuntime) MarkTopicMemberOnline(types.TenantID, string, types.Uid, string) error {
 	return nil
 }

server/topic.go

@@ -650,6 +650,14 @@ func (t *Topic) runLocal(hub *Hub) {
 
 // handleClientMsg is the top-level handler of messages received by the topic from sessions.
 func (t *Topic) handleClientMsg(msg *ClientComMessage) {
+	kind := "other"
+	if t.cat == types.TopicCatP2P {
+		kind = "p2p"
+	} else if t.cat == types.TopicCatGrp {
+		kind = "group"
+	}
+	immetrics.ObserveTopicQueueWait(kind, msg.queuedAt)
+	msg.queuedAt = time.Time{}
 	if msg.TenantID.IsZero() {
 		msg.TenantID = t.tenantID
 	} else if msg.TenantID != t.tenantID {
@@ -1464,9 +1472,13 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid,
 	publishRealtimeTopicMessage(t.name, data)
 	immetrics.ObserveMessagePublishStage(kind, "realtime_fanout", fanoutStarted, nil)
 
-	// sendPush will update unread message count and send push notification.
-	if pushRcpt := t.pushForData(asUid, data.Data, markedReadBySender); pushRcpt != nil {
-		sendPush(pushRcpt)
+	// Push preparation copies immutable topic state. Presence resolution, cluster
+	// recipient partitioning and unread updates run outside the topic loop.
+	pushStarted := time.Now()
+	pushRcpt := t.pushForData(asUid, data.Data, markedReadBySender)
+	immetrics.ObserveMessagePublishStage(kind, "push_prepare", pushStarted, nil)
+	if pushRcpt != nil {
+		dispatchPushAsync(pushRcpt, true)
 	}
 	return nil
 }