diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index 58063b7..2c0e393 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -146,3 +146,61 @@ Run ID：`s60-group-r2-0929`。 - 群聊 R2：`loadtest/im-pressure/results/s60-group-r2-0929/` 结果目录包含 coordinator、A/B summary、合并 report 和 agent 指标 JSONL。 + +## 12. 精确成员规模诊断复测 + +为区分“在线连接数”和“群成员总数”的影响，新增 300、600 人独立群复测。每个测试群在起压前均按用户 ID 集合核验，要求实际成员与目标账号完全相等。 + +| 场景 | Run ID | 在线/群成员 | ACK | 实时投递 | 历史 | DB 精确命中 | ACK P95 | 投递 P95 | +|---|---|---:|---:|---:|---:|---:|---:|---:| +| 原 1200 人群、仅 300 人在线 | `s300-group-r1b-0929` | 300/1200 | 45/300 | 13500/13500（按 ACK 台账） | 206/206 | 45 ACK + 210 结果未知全部落库 | 18994ms | 89489ms | +| 精确 300 人群 | `s300m-group-r1-0930` | 300/300 | 300/300 | 90000/90000 | 300/300 | 300/300 | 235.177ms | 500ms 桶内 | +| 精确 600 人群 | `s600m-group-r1-0930` | 600/600 | 600/600 | 360000/360000 | 600/600 | 600/600 | 235.177ms | 500ms 桶内 | +| 精确 900 人群 | `s900m-group-r1-0930` | 900/900 | 890/900 | 800555/801000（按 ACK 台账） | 899/899 | 890 ACK + 9 结果未知落库 | >10s，最大19.272s | >10s，最大55.279s | + +三轮均未出现 WS 发送队列满、WS 写失败或 Kafka lag。精确 300、600 人群在实时阶段也没有新增 `session_topic_publish`、`topic_client`、`topic_server` 满队列，说明 WS 写链路在 600 在线规模内不是瓶颈。 + +原 1200 人群仅 300 人在线时有约 900 个离线成员。群消息处理会在 Topic 串行路径中为每个离线成员逐个调用 Redis presence 查询；上一条消息 ACK 后的推送准备持续占用 Topic goroutine，使后续消息长时间等待，最终表现为 ACK 超时和分钟级投递延迟。精确成员群消除这部分离线成员后，同等 300 在线规模恢复正常，验证了该路径是主要根因。 + +600 人实时发送通过后，240 秒错峰下线仍使群主 Pod 的 `topic_server full` 从 36 增至 128，净增 92；其他两类内部队列无增量。该问题来自最后会话离线时的 presence 扩散，与消息保存、Kafka 消费和 WS 实时投递无关，仍需在服务端合并或限流离线状态广播。 + +精确 900 人群出现容量拐点：900 次发送中 890 次 ACK，9 次 ACK 超时、1 次 socket 失败；7 条连接在实时阶段超时断开并自动恢复。按 ACK 台账计算的 445 次实时缺口全部可从历史补回，890 条 ACK 全部精确落库；9 条结果未知实际落库，1 条 socket 失败未落库。因此本轮没有持久化丢失，但实时正确性和延迟门禁未通过。 + +900 人实时阶段三个内部队列的 full 计数均无增量，Kafka lag 为 0。成功处理的 899 条消息中，`permission_check` P95 不超过 10ms、`save` P95 不超过 50ms、`kafka_produce` P95 不超过 25ms、`realtime_fanout` P95 不超过 5ms；抽样覆盖后半段 575468 次 WS 投递，发送队列等待 P95 不超过 10ms、socket write P95 不超过 0.5ms，且没有发送队列满或 text write 失败。 + +现有 `realtime_fanout` 指标在 `pushForData` 之前结束。`pushForData` 位于 Topic 串行处理路径中，每条群消息遍历全部成员、构造推送收件人并调用 `sendPush`；即使成员全部在线，900 人规模下这段未计时工作仍可让后续发布在 `topic_client` 中排队。队列未达到容量上限并不代表没有排队，这与 ACK 延迟逐步增长到 19 秒、投递延迟增长到 55 秒的现象一致。下一步应先补充 `push_prepare`、`push_dispatch` 和 Topic client queue wait 指标，再将群推送准备移出 Topic 串行热路径或改为异步批处理。 + +900 人在 240 秒错峰下线后，群主 Pod 的 `topic_server full` 从 0 增至 379，说明 presence 离线扩散问题随群规模明显放大。压测机两个 agent 的峰值 CPU 分别约 39.5% 和 38.5% 单核，峰值 RSS 各约 76MB、FD 各 460，不是本轮瓶颈。 + +新增产物： + +- 300 人精确群：`loadtest/im-pressure/results/s300-group-members-setup-0929/` +- 300 人复测：`loadtest/im-pressure/results/s300m-group-r1-0930/` +- 600 人精确群：`loadtest/im-pressure/results/s600-group-members-setup-0930/` +- 600 人复测：`loadtest/im-pressure/results/s600m-group-r1-0930/` +- 900 人精确群：`loadtest/im-pressure/results/s900-group-members-setup-0930/` +- 900 人复测：`loadtest/im-pressure/results/s900m-group-r1-0930/` + +## 13. 容量拐点修复（待发版复测） + +针对精确 900 人群暴露的 Topic 串行积压和批量下线 presence 风暴，已完成以下代码改造： + +1. `pushForData` 在 Topic goroutine 内只复制成员推送快照，不再逐成员同步查询 Redis，也不再执行跨 Pod 收件人拆分和 user updater 投递。 +2. 新增总容量 4096 的有界异步推送队列和 4 个固定分片 worker。同一 `tenant + topic` 固定进入同一分片，保证同一群的推送和未读更新顺序；不同群可并行处理。 +3. 离线成员 presence 使用一次 Redis pipeline 批量读取，单条消息的 Redis 网络往返由成员数 N 次降为 1 次。 +4. 群离线 presence 遇到 Topic server queue 满时不再直接丢弃。服务端按 `tenant + topic + source user` 合并为最新状态，并在 Topic 队列恢复后重投；待处理集合上限 16384。 +5. 新增 Topic client queue wait、push prepare、push dispatch queue、push dispatch duration、push recipient count、deferred presence 当前深度等指标。 + +新增重点指标： + +- `im_topic_queue_wait_duration_seconds` +- `im_message_publish_stage_duration_seconds{stage="push_prepare"}` +- `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_internal_queue_current{queue="topic_presence_pending"}` +- `im_internal_queue_enqueue_total{queue="topic_presence_deferred"}` + +发版后先复测精确 600 人群，确认基线无回归；再复测精确 900 人群。900 人验收要求：900/900 ACK、810000/810000 实时投递、0 断连、0 持久化缺失，`topic_client/topic_server/push_dispatch` 无 full，Topic queue wait P95 不超过 500ms，ACK P95 不超过 500ms，投递 P95 不超过 1s。渐进下线结束后 deferred presence 必须回落到 0，且不得出现 `topic_server full`。 diff --git a/server/datamodel.go b/server/datamodel.go index 01bb8d5..63d1950 100644 --- a/server/datamodel.go +++ b/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:"-"` diff --git a/server/hub.go b/server/hub.go index 2e59200..14a2443 100644 --- a/server/hub.go +++ b/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 diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 972466a..289b08d 100644 --- a/server/metrics/metrics.go +++ b/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) diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index 93e4f02..6cc86de 100644 --- a/server/monitoring_http_test.go +++ b/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) diff --git a/server/push.go b/server/push.go index 50529eb..db6e408 100644 --- a/server/push.go +++ b/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 { diff --git a/server/push_async_test.go b/server/push_async_test.go new file mode 100644 index 0000000..c61529d --- /dev/null +++ b/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") + } +} diff --git a/server/session.go b/server/session.go index d5b9e06..79c4ff9 100644 --- a/server/session.go +++ b/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: diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index 3d620fc..cf4e479 100644 --- a/server/store/group_message_runtime.go +++ b/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" diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 76a24ad..d16d1f3 100644 --- a/server/store/group_message_runtime_test.go +++ b/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 } diff --git a/server/topic.go b/server/topic.go index f1ee8bb..417fa79 100644 --- a/server/topic.go +++ b/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 }