diff --git a/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md b/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md index 42d0dbf..c73e889 100644 --- a/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md +++ b/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md @@ -105,3 +105,41 @@ Run ID：`group2k-r1-20260927-01` - 实时广播：失败，存在队列丢弃和断线。 - Kafka 持久化：已处理的 488 条全部成功，但该结论不代表计划的 2000 条均被接受。 - 总体：失败，必须完成 P0 后按阶梯重测。 + +## 7. P0 修复实现 + +本轮分析后的代码修复如下： + +1. 群消息的 `{pres what=msg}` 不再由 Topic master 对每个成员逐条跨 Pod 路由。master 将收件人清单合并为一条 `group_unread/users` Redis Pub/Sub 消息；来源 Pod 本地投递一次，其他 Pod 各消费一次。 +2. 每个 Pod 只扫描一次本地 WebSocket：已订阅该群 Topic 的连接依靠正常 `{data}` 广播，不重复发送 `me` presence；同一用户未订阅该群的其他设备仍收到 `me` presence，保留多开语义。 +3. Redis fanout 未启用或批量发布失败时，自动回退到原来的逐用户 Topic 路由，避免静默漏掉未读通知。 +4. 原生 Tinode Cluster 开启时，普通 user message 不再重复发布 Redis user fanout；session control 仍保留 Redis 跨 Pod 广播。 +5. `routeSrv` 改为有界非阻塞写入，队列满时记录错误和指标，不再无限阻塞 Topic 处理链路。 +6. 新增 `im_internal_queue_enqueue_total`、`im_internal_queue_depth`、`im_internal_queue_capacity`，覆盖 `session_topic_publish`、`hub_route_server`、`topic_client`、`topic_server`、`cluster_topic_proxy`。 + +修复后的复杂度从“每条消息约 2000 次 Cluster route + 2000 次 Redis publish”降为“一次 Redis publish + 每 Pod 一次本地 session 扫描”。2000 人都订阅群 Topic 时，批量 unread 在本地扫描后不会产生重复 WebSocket 帧。 + +## 8. 复测观测项 + +复测继续使用 `group2k-r1` 相同发送模型，并重点检查： + +```promql +sum by (queue, result) (rate(im_internal_queue_enqueue_total[1m])) + +histogram_quantile( + 0.95, + sum by (le, queue) (rate(im_internal_queue_depth_bucket[1m])) +) + +sum by (stage, kind, target_type, status) ( + increase(realtime_fanout_events_total[5m]) +) +``` + +验收要求： + +- `session_topic_publish`、`topic_client`、`topic_server`、`cluster_topic_proxy` 的 `result="full"` 均为 0。 +- `group_unread/users` 的 publish 数应接近服务端实际接受的群消息数，不再乘以群成员数。 +- 群压测窗口内普通 `message/user` Redis publish 不再随群成员数增长。 +- `im_cluster_route_total` 不应再出现每条群消息约 2000 次的线性放大。 +- 2000 条计划消息需分别核对 ACK、未知结果、实时观测和最终落库，不能只按 ACK 台账判断落库率。 diff --git a/server/cluster.go b/server/cluster.go index 5611436..1f5541e 100644 --- a/server/cluster.go +++ b/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 { diff --git a/server/hub.go b/server/hub.go index f18615c..02f1330 100644 --- a/server/hub.go +++ b/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")) && diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 1e2b714..718a71e 100644 --- a/server/metrics/metrics.go +++ b/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" diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index 1fb5bd1..9bc2523 100644 --- a/server/monitoring_http_test.go +++ b/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) diff --git a/server/pres.go b/server/pres.go index 3f2a9e9..afe7b3c 100644 --- a/server/pres.go +++ b/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) } diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go index 25d1482..ec8faf0 100644 --- a/server/realtime_dispatcher.go +++ b/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 { diff --git a/server/realtime_dispatcher_test.go b/server/realtime_dispatcher_test.go index 91e0a22..5aaeb7f 100644 --- a/server/realtime_dispatcher_test.go +++ b/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 diff --git a/server/session.go b/server/session.go index f066875..d8dc7e3 100644 --- a/server/session.go +++ b/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) diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index 1314e42..3d620fc 100644 --- a/server/store/group_message_runtime.go +++ b/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