perf: batch group unread fanout

server/cluster.go

@@ -743,7 +743,9 @@ func (Cluster) TopicProxy(msg *ClusterResp, unused *bool) error {
 		msg.SrvMsg.uid = types.ParseUserId(msg.SrvMsg.AsUser)
 		select {
 		case t.proxy <- msg:
+			immetrics.ObserveInternalQueueEnqueue("cluster_topic_proxy", len(t.proxy), cap(t.proxy), true)
 		default:
+			immetrics.ObserveInternalQueueEnqueue("cluster_topic_proxy", len(t.proxy), cap(t.proxy), false)
 			logs.Warn.Printf("cluster: proxy channel full, topic %s", msg.RcptTo)
 		}
 	} else {

server/hub.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/tinode/chat/server/auth"
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
@@ -247,7 +248,9 @@ func (h *Hub) run() {
 				if dst.clientMsg != nil {
 					select {
 					case dst.clientMsg <- msg:
+						immetrics.ObserveInternalQueueEnqueue("topic_client", len(dst.clientMsg), cap(dst.clientMsg), true)
 					default:
+						immetrics.ObserveInternalQueueEnqueue("topic_client", len(dst.clientMsg), cap(dst.clientMsg), false)
 						logs.Err.Println("hub: topic's broadcast queue is full", dst.name)
 					}
 				} else {
@@ -275,7 +278,9 @@ func (h *Hub) run() {
 				// Everything is OK, sending packet to known topic
 				select {
 				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)
 				}
 			} else if (strings.HasPrefix(msg.RcptTo, "usr") || strings.HasPrefix(msg.RcptTo, "grp")) &&

server/metrics/metrics.go

@@ -102,6 +102,19 @@ var (
 		Name: "im_cluster_proxy_response_total",
 		Help: "Cluster proxy responses by pod and result.",
 	}, []string{"pod", "result"})
+	InternalQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "im_internal_queue_enqueue_total",
+		Help: "Internal message queue enqueue attempts by pod, queue, and result.",
+	}, []string{"pod", "queue", "result"})
+	InternalQueueDepth = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_internal_queue_depth",
+		Help:    "Internal message queue depth observed after enqueue attempts by pod and queue.",
+		Buckets: []float64{0, 1, 4, 16, 32, 64, 128, 192, 256, 512, 1024, 2048, 4096, 8192},
+	}, []string{"pod", "queue"})
+	InternalQueueCapacity = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "im_internal_queue_capacity",
+		Help: "Configured internal message queue capacity 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.",
@@ -135,7 +148,8 @@ func init() {
 		HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration,
 		RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength,
 		ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration,
-		ClusterRouteTotal, ClusterProxyResponseTotal, MessagePersistDuration, MessagePublishStageDuration,
+		ClusterRouteTotal, ClusterProxyResponseTotal, InternalQueueEnqueueTotal, InternalQueueDepth,
+		InternalQueueCapacity, MessagePersistDuration, MessagePublishStageDuration,
 		KafkaMessagePersistTotal)
 	Registry.MustRegister(KafkaConsumerLag, KafkaBacklogOldestAge, KafkaDeadLettersTotal)
 }
@@ -177,6 +191,18 @@ func ObserveRealtimeFanoutDelivery(kind, targetType string, timestamp time.Time)
 		Observe(time.Since(timestamp).Seconds())
 }
 
+func ObserveInternalQueueEnqueue(queue string, depth, capacity int, accepted bool) {
+	result := "ok"
+	if !accepted {
+		result = "full"
+	}
+	pod := metricPodName()
+	queue = normalizeLabel(queue)
+	InternalQueueEnqueueTotal.WithLabelValues(pod, queue, result).Inc()
+	InternalQueueDepth.WithLabelValues(pod, queue).Observe(float64(depth))
+	InternalQueueCapacity.WithLabelValues(pod, queue).Set(float64(capacity))
+}
+
 func normalizeLabel(value string) string {
 	if value == "" {
 		return "unknown"

server/monitoring_http_test.go

@@ -90,6 +90,7 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 	immetrics.KafkaMessagePersistTotal.WithLabelValues("p2p", "batch_ok").Add(0)
 	immetrics.KafkaConsumerLag.WithLabelValues("test", "topic", "group").Set(0)
 	immetrics.KafkaDeadLettersTotal.WithLabelValues("p2p", "test").Add(0)
+	immetrics.ObserveInternalQueueEnqueue("test", 0, 1, true)
 	wrt := httptest.NewRecorder()
 	serveMetrics(wrt, httptest.NewRequest(http.MethodGet, "/metrics", nil))
 
@@ -102,6 +103,7 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 	}
 	for _, metric := range []string{"im_ws_connections_active", "im_cluster_p2m_queue_length",
 		"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_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/pres.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
@@ -466,12 +467,26 @@ func (t *Topic) presSubsOffline(what string, params *presParams,
 	if offlineOnly {
 		skipTopic = t.name
 	}
+	if t.cat == types.TopicCatGrp && offlineOnly && what == "msg" && store.RealtimeFanoutEnabled() {
+		recipients := make([]realtimeGroupUnreadRecipient, 0, len(t.perUser))
+		for uid, pud := range t.perUser {
+			if pud.deleted || !presOfflineFilter(pud.modeGiven&pud.modeWant, what, filterSource) {
+				continue
+			}
+			recipients = append(recipients, realtimeGroupUnreadRecipient{
+				User:   uid.UserId(),
+				Source: t.original(uid),
+			})
+		}
+		if publishRealtimeGroupUnread(t.tenantID, t.name, params.actor, params.seqID, recipients) {
+			return
+		}
+	}
 
 	for uid, pud := range t.perUser {
 		if pud.deleted || !presOfflineFilter(pud.modeGiven&pud.modeWant, what, filterSource) {
 			continue
 		}
-
 		user := uid.UserId()
 		actor := params.actor
 		target := params.target
@@ -693,10 +708,16 @@ func presSingleUserOfflineOffline(tenantID types.TenantID, uid types.Uid, origin
 }
 
 func routeUserServerMessage(uid types.Uid, msg *ServerComMessage) {
-	if msg == nil || uid.IsZero() {
+	if msg == nil || uid.IsZero() || globals.hub == nil {
 		return
 	}
-	globals.hub.routeSrv <- msg
+	select {
+	case globals.hub.routeSrv <- msg:
+		immetrics.ObserveInternalQueueEnqueue("hub_route_server", len(globals.hub.routeSrv), cap(globals.hub.routeSrv), true)
+	default:
+		immetrics.ObserveInternalQueueEnqueue("hub_route_server", len(globals.hub.routeSrv), cap(globals.hub.routeSrv), false)
+		logs.Err.Printf("route user message: hub routeSrv queue full, user=%s", uid.UserId())
+	}
 	publishRealtimeUserMessage(uid, msg)
 }
 

server/realtime_dispatcher.go

@@ -16,6 +16,7 @@ import (
 
 const (
 	realtimeKindMessage        = "message"
+	realtimeKindGroupUnread    = "group_unread"
 	realtimeKindSessionControl = "session_control"
 	realtimeKindCallEvent      = "call_event"
 
@@ -23,8 +24,21 @@ const (
 	realtimeTargetUser   = "user"
 	realtimeTargetDevice = "device"
 	realtimeTargetPlatf  = "platform"
+	realtimeTargetUsers  = "users"
 )
 
+type realtimeGroupUnreadRecipient struct {
+	User   string `json:"user"`
+	Source string `json:"source"`
+}
+
+type realtimeGroupUnreadMessage struct {
+	Topic      string                         `json:"topic"`
+	Actor      string                         `json:"actor,omitempty"`
+	SeqID      int64                          `json:"seq_id"`
+	Recipients []realtimeGroupUnreadRecipient `json:"recipients"`
+}
+
 type realtimeCallEventMessage struct {
 	Topic     string          `json:"topic"`
 	From      string          `json:"from"`
@@ -68,6 +82,12 @@ func publishRealtimeUserMessage(uid types.Uid, msg *ServerComMessage) {
 		immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "skip_invalid")
 		return
 	}
+	// Native cluster routing already forwards user topics between pods. Publishing
+	// the same message through Redis duplicates work and may deliver it twice.
+	if globals.cluster != nil {
+		immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "skip_native_cluster")
+		return
+	}
 	dedupKey := realtimeMessageDedupKey(uid.UserId(), msg)
 	if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetUser, uid.UserId(), msg.SkipSid, dedupKey, msg); err != nil {
 		immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "error")
@@ -77,6 +97,29 @@ func publishRealtimeUserMessage(uid types.Uid, msg *ServerComMessage) {
 	immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "ok")
 }
 
+func publishRealtimeGroupUnread(tenantID types.TenantID, topic, actor string, seqID int64,
+	recipients []realtimeGroupUnreadRecipient) bool {
+	if !tenantID.IsValid() || topic == "" || seqID <= 0 || len(recipients) == 0 || !store.RealtimeFanoutEnabled() {
+		return false
+	}
+	payload := &realtimeGroupUnreadMessage{
+		Topic:      topic,
+		Actor:      actor,
+		SeqID:      seqID,
+		Recipients: recipients,
+	}
+	dedupKey := "group-unread:" + topic + ":" + strconv.FormatInt(seqID, 10)
+	if err := store.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)
+		return false
+	}
+	immetrics.IncRealtimeFanout("publish", realtimeKindGroupUnread, realtimeTargetUsers, "ok")
+	deliverRealtimeGroupUnread(tenantID, payload)
+	return true
+}
+
 func publishRealtimeCallEvent(msg *ClientComMessage) {
 	if msg == nil || msg.Note == nil || !msg.TenantID.IsValid() ||
 		msg.RcptTo == "" || msg.AsUser == "" || msg.Note.What != "call" ||
@@ -193,6 +236,16 @@ func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) {
 		return
 	}
 	immetrics.IncRealtimeFanout("receive", envelope.Kind, envelope.TargetType, "ok")
+	if envelope.Kind == realtimeKindGroupUnread {
+		var batch realtimeGroupUnreadMessage
+		if err := json.Unmarshal(envelope.Message, &batch); err != nil {
+			immetrics.IncRealtimeFanout("receive", envelope.Kind, envelope.TargetType, "malformed")
+			logs.Warn.Println("redis realtime fanout: malformed group unread batch", err)
+			return
+		}
+		deliverRealtimeGroupUnread(envelope.TenantID, &batch)
+		return
+	}
 	var msg ServerComMessage
 	if err := json.Unmarshal(envelope.Message, &msg); err != nil {
 		immetrics.IncRealtimeFanout("receive", envelope.Kind, envelope.TargetType, "malformed")
@@ -224,6 +277,51 @@ func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) {
 	}
 }
 
+func deliverRealtimeGroupUnread(tenantID types.TenantID, batch *realtimeGroupUnreadMessage) {
+	if globals.sessionStore == nil || batch == nil || batch.Topic == "" || batch.SeqID <= 0 || len(batch.Recipients) == 0 {
+		immetrics.IncRealtimeFanout("deliver", realtimeKindGroupUnread, realtimeTargetUsers, "skip_invalid")
+		return
+	}
+	recipients := make(map[types.Uid]string, len(batch.Recipients))
+	for _, recipient := range batch.Recipients {
+		uid := types.ParseUserId(recipient.User)
+		if !uid.IsZero() {
+			recipients[uid] = recipient.Source
+		}
+	}
+	delivered := 0
+	globals.sessionStore.Range(func(_ string, sess *Session) bool {
+		if sess == nil || sess.isCluster() || sess.isProxy() || sess.tenantID != tenantID {
+			return true
+		}
+		source, ok := recipients[sess.uid]
+		if !ok || sessionHasTopicOrP2PAlias(sess, batch.Topic) {
+			return true
+		}
+		actor := batch.Actor
+		if actor == sess.uid.UserId() {
+			actor = ""
+		}
+		msg := &ServerComMessage{
+			TenantID: tenantID,
+			RcptTo:   sess.uid.UserId(),
+			Pres: &MsgServerPres{
+				Topic:     "me",
+				What:      "msg",
+				Src:       source,
+				AcsActor:  actor,
+				SeqId:     batch.SeqID,
+				SkipTopic: batch.Topic,
+			},
+		}
+		if sess.queueOut(msg) {
+			delivered++
+		}
+		return true
+	})
+	recordRealtimeFanoutDeliverResult(realtimeKindGroupUnread, realtimeTargetUsers, delivered)
+}
+
 func deliverRealtimeCallEvent(envelope *store.RealtimeFanoutEnvelope) {
 	var event realtimeCallEventMessage
 	if err := json.Unmarshal(envelope.Message, &event); err != nil {

server/realtime_dispatcher_test.go

@@ -109,6 +109,60 @@ func TestRealtimeUserPresenceSkipsSessionsAttachedToTopic(t *testing.T) {
 	}
 }
 
+func TestRealtimeGroupUnreadBatchPreservesDetachedMultiDeviceNotification(t *testing.T) {
+	oldStore := globals.sessionStore
+	defer func() { globals.sessionStore = oldStore }()
+
+	tenantID := types.TenantID(1)
+	recipient := types.Uid(1002)
+	other := types.Uid(1003)
+	topic := "grpUnreadBatchTest"
+	globals.sessionStore = realtimeTestSessionStore(
+		realtimeTestSession("sid-attached", tenantID, recipient, topic),
+		realtimeTestSession("sid-detached-1", tenantID, recipient, ""),
+		realtimeTestSession("sid-detached-2", tenantID, recipient, ""),
+		realtimeTestSession("sid-other", tenantID, other, ""),
+	)
+	payload := realtimeGroupUnreadMessage{
+		Topic: topic,
+		Actor: types.Uid(1001).UserId(),
+		SeqID: 51,
+		Recipients: []realtimeGroupUnreadRecipient{
+			{User: recipient.UserId(), Source: topic},
+		},
+	}
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
+		TenantID:   tenantID,
+		Kind:       realtimeKindGroupUnread,
+		TargetType: realtimeTargetUsers,
+		Target:     topic,
+		Message:    raw,
+	})
+
+	if len(globals.sessionStore.Get("sid-attached").send) != 0 {
+		t.Fatal("attached session must rely on the topic data broadcast")
+	}
+	for _, sid := range []string{"sid-detached-1", "sid-detached-2"} {
+		select {
+		case got := <-globals.sessionStore.Get(sid).send:
+			pres := got.(*ServerComMessage).Pres
+			if pres == nil || pres.What != "msg" || pres.Src != topic || pres.SeqId != 51 || pres.SkipTopic != topic {
+				t.Fatalf("%s got unexpected presence: %#v", sid, pres)
+			}
+		default:
+			t.Fatalf("detached session %s must receive unread presence", sid)
+		}
+	}
+	if len(globals.sessionStore.Get("sid-other").send) != 0 {
+		t.Fatal("non-recipient session received unread presence")
+	}
+}
+
 func TestRealtimeP2PTopicFallbackRewritesVisibleTopic(t *testing.T) {
 	oldStore := globals.sessionStore
 	oldHub := globals.hub

server/session.go

@@ -25,6 +25,7 @@ import (
 	"github.com/tinode/chat/pbx"
 	"github.com/tinode/chat/server/auth"
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 
@@ -860,7 +861,9 @@ func (s *Session) publish(msg *ClientComMessage) {
 		// This is a post to a subscribed topic. The message is sent to the topic only
 		select {
 		case sub.broadcast <- msg:
+			immetrics.ObserveInternalQueueEnqueue("session_topic_publish", len(sub.broadcast), cap(sub.broadcast), true)
 		default:
+			immetrics.ObserveInternalQueueEnqueue("session_topic_publish", len(sub.broadcast), cap(sub.broadcast), false)
 			// Reply with a 503 to the user.
 			s.queueOut(ErrServiceUnavailableReply(msg, msg.Timestamp))
 			logs.Err.Println("s.publish: sub.broadcast channel full, topic ", msg.RcptTo, s.sid)

server/store/group_message_runtime.go

@@ -392,6 +392,13 @@ func StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler f
 	return groupMsgRuntime.StartRealtimeFanoutSubscriber(ctx, nodeID, handler)
 }
 
+// RealtimeFanoutEnabled reports whether Redis Pub/Sub fanout is configured.
+// Callers use it to retain the native routing fallback when Redis is disabled.
+func RealtimeFanoutEnabled() bool {
+	runtime, ok := groupMsgRuntime.(*redisGroupMessageRuntime)
+	return ok && runtime != nil && runtime.cfg.GroupFanout
+}
+
 func CacheObject(tenantID types.TenantID, namespace, id string, value any) error {
 	if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || value == nil {
 		return nil