From fc875cc84e56edb77a658a93394b9f6906bd7cc9 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 27 Sep 2026 22:59:17 +0800 Subject: [PATCH] test: report ambiguous group publish outcomes --- ...IM-group2k-r1-群聊压测问题分析-20260927.md | 107 ++++++++++++++++++ loadtest/im-pressure/merge.go | 17 ++- loadtest/im-pressure/merge_test.go | 8 +- 3 files changed, 129 insertions(+), 3 deletions(-) create mode 100644 docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md diff --git a/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md b/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md new file mode 100644 index 0000000..42d0dbf --- /dev/null +++ b/docs/test/IM-group2k-r1-群聊压测问题分析-20260927.md @@ -0,0 +1,107 @@ +# IM 2000 人群聊压测问题分析 + +日期：2026-09-27 +Run ID：`group2k-r1-20260927-01` +场景：A/B 各 1000 个 WebSocket，各 500 个发送者发 2 条，计划发送 2000 条群消息。 + +## 1. 结论 + +本轮失败，且不能仅解读为 WebSocket 丢包。核心问题是 2000 人群每条消息触发 O(群成员数) 的额外用户 `me` Topic 路由，导致 Topic master、集群通道和 MySQL 连接池连锁积压。服务端处理吞吐下降后，发送端先经历 ACK 超时，随后 Topic 队列写满并返回 503。 + +同时，压测程序对 ACK 超时的口径不完整：超时后服务端仍可能接受、广播并落库，不能把 `ctrl_timeout` 直接当作明确拒绝。 + +## 2. 客户端结果 + +| 指标 | A | B | 合计 | +|---|---:|---:|---:| +| WS READY | 1000 | 1000 | 2000 | +| 发送尝试 | 1000 | 1000 | 2000 | +| 收到 2xx ACK | 16 | 19 | 35 | +| 503 | 224 | 272 | 496 | +| ACK 超时 | 732 | 709 | 1441 | +| socket 错误 | 28 | 0 | 28 | +| 断线 | 21 | 0 | 21 | +| ACK 台账内实时缺失 | 735 | 0 | 735 | +| ACK 台账内历史落库缺失 | 0 | 0 | 0 | +| 未在 ACK 台账中的实时投递 | 368270 | 377000 | 745270 | + +35 条 ACK 消息理论上应产生 70000 次本轮投递，实际共记录 69265 次，其中 A 端少 735 次。但这不是服务端实际接受的全量；额外 745270 次投递来自“未及时收到 ACK，但服务端已处理”的消息。 + +## 3. 服务端证据 + +### 3.1 实际接受和落库 + +- `im_message_publish_stage_duration_seconds_count{kind="group",stage="save",result="ok"}` 增加 488。 +- `im_kafka_message_persist_total{kind="group",result="batch_ok"}` 增加 488。 +- 群消息 Kafka 批持久化平均约 46.6ms，488 条全部 batch success。 + +因此，服务端至少接受并落库 488 条，而不是客户端 ACK 台账中的 35 条。其中至少 453 条属于“已接受但 ACK 未在 20 秒内返回”。 + +### 3.2 队列溢出 + +| Pod | 日志 | 次数 | +|---|---|---:| +| `im-server-2` | `s.publish: sub.broadcast channel full` | 496 | +| `im-server-2` | `hub: topic's broadcast queue is full` | 1655 | +| `im-server-1` | `cluster: proxy channel full` | 47 | + +群 Topic `grp3Ita5tyBIMs` 的 master 在 `im-server-2`。其 `clientMsg` 队列容量只有 192，proxy 响应队列容量为 128。队列满时会立即返回 503 或丢弃 proxy 响应。 + +### 3.3 每条群消息的路由放大 + +`Topic.saveAndBroadcastMessage` 在每条消息中同步执行： + +1. `presSubsOffline("msg", ..., offlineOnly=true)` 遍历 `t.perUser` 的 2000 个成员。 +2. 为每个成员构建一条用户 `me` Topic presence 消息。 +3. `routeUserServerMessage` 同时走 Tinode 原生 Cluster 路由和 Redis user fanout。 +4. 最后才向群 Topic 的会话广播 `{data}`。 + +488 条已接受群消息产生了约 98 万次额外用户路由： + +- `realtime_fanout_events_total{stage="publish",target_type="user"}` 增加 979979。 +- `realtime_fanout_events_total{stage="receive",target_type="user"}` 增加 894360。 +- Redis 接收后真正本地投递只有 666 次，`no_local_session` 为 893694。 +- Cluster route 约增加 109 万次。 + +这证明绝大多数用户 fanout 在当前全员已订阅群 Topic 的场景中完全无效，却阻塞了 Topic master 的串行处理协程。 + +### 3.4 MySQL 连接等待 + +3 个 Pod 的 `meeting_im` 连接池在测试窗口中合计： + +- 等待次数增加 4011。 +- 等待时间增加约 35399 秒。 +- DB `exec ok` 增加 3453，平均耗时约 501ms。 + +连接池每 Pod 最大 24 个连接。这不是本轮第一个堵点，但会进一步拖慢群权限检查、未读数更新和其他业务 SQL。仅提高连接数不会消除 O(成员数) 的放大。 + +## 4. 压测程序口径问题 + +1. `ctrl_timeout` 和 socket 错误属于发布结果未知，不是明确拒绝。 +2. recorder 能够统计 `unexpected_messages`，但 merge 报告没有汇总该字段，也没有将它加入门禁。 +3. 历史回查只检查 ACK 台账，因此只检查了 35 条，无法单独证明其他超时消息的落库情况。 +4. 群聊应增加“已 ACK”、“已明确拒绝”、“结果未知”、“未 ACK 但已观测广播”和“最终落库”五种状态。 + +## 5. 改进顺序 + +### P0 + +1. 停止在原生 Cluster 已开启时对同一条 user message 再做 Redis user fanout，避免双通道重复路由。 +2. 重构群消息的 unread presence：已订阅该群 Topic 的用户不再生成额外 `me` 通知；未订阅但在线的成员使用节点级批量 fanout，不得在 Topic master 协程中逐人跨 Pod 路由。 +3. 为 Topic `clientMsg`、proxy 响应队列和 session send queue 增加 length/capacity/drop Prometheus 指标。 +4. 压测 merge 报告增加结果未知及 unacked-observed 口径，并把 `unexpected_messages` 纳入失败门禁。 + +### P1 + +1. 检查群消息未读数更新和 Push receipt 是否对 2000 人逐人访问 Redis/MySQL，改为批处理或延迟合并。 +2. 为 `CanPost` 的全员禁言和单人禁言结果增加短 TTL 缓存，管理操作时精确失效。 +3. P0 完成后先跑 100、500、1000、2000 人群阶梯，每档检查 ACK P95、Topic 队列水位、Cluster route 数、DB pool wait 和 Kafka lag，不直接重跑全量。 + +## 6. 本轮验收结论 + +- 连接建立：通过，2000/2000 READY。 +- 发布成功率：失败。 +- ACK 延迟：失败，已超过 20 秒超时窗口。 +- 实时广播：失败，存在队列丢弃和断线。 +- Kafka 持久化：已处理的 488 条全部成功，但该结论不代表计划的 2000 条均被接受。 +- 总体：失败，必须完成 P0 后按阶梯重测。 diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index f99f2d6..d1d76d8 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -19,6 +19,8 @@ type CombinedReport struct { PublishAttempted int64 `json:"publish_attempted"` PublishAccepted int64 `json:"publish_accepted"` PublishRejected int64 `json:"publish_rejected"` + PublishExplicitReject int64 `json:"publish_explicitly_rejected"` + PublishOutcomeUnknown int64 `json:"publish_outcome_unknown"` PublishErrors map[string]int64 `json:"publish_errors,omitempty"` PublishSuccessRate float64 `json:"publish_success_rate"` AckP95MS float64 `json:"ack_p95_ms"` @@ -36,6 +38,7 @@ type CombinedReport struct { OutOfOrder int64 `json:"out_of_order"` OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` Disconnects int64 `json:"disconnects"` + UnexpectedMessages int64 `json:"unexpected_messages"` HistoryChecked int64 `json:"history_checked"` HistoryFound int64 `json:"history_found"` PersistenceLost int64 `json:"persistence_lost"` @@ -71,6 +74,9 @@ func runMerge(aPath, bPath, output string) error { for category, count := range summary.PublishErrors { report.PublishErrors[category] += count } + report.PublishOutcomeUnknown += summary.PublishErrors["ctrl_timeout"] + summary.PublishErrors["socket"] + report.PublishExplicitReject += summary.PublishErrors["403"] + summary.PublishErrors["502"] + + summary.PublishErrors["503"] + summary.PublishErrors["other"] report.ExpectedDeliveries += summary.ExpectedPeer + summary.ExpectedGroup report.ReceivedDeliveries += summary.ReceivedPeer + summary.ReceivedGroup report.RealtimeLost += summary.RealtimeLost @@ -79,6 +85,7 @@ func runMerge(aPath, bPath, output string) error { report.OutOfOrder += summary.OutOfOrder report.OrderingViolations = append(report.OrderingViolations, summary.OrderingViolations...) report.Disconnects += summary.Disconnects + report.UnexpectedMessages += summary.UnexpectedMessages report.HistoryChecked += summary.HistoryChecked report.HistoryFound += summary.HistoryFound report.PersistenceLost += summary.PersistenceLost @@ -112,6 +119,14 @@ func runMerge(aPath, bPath, output string) error { report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("publish success rate=%.6f expected>=0.999900", report.PublishSuccessRate)) } + if report.PublishOutcomeUnknown > 0 { + report.FailureReasons = append(report.FailureReasons, + fmt.Sprintf("publish outcome unknown=%d (ACK timeout or socket failure)", report.PublishOutcomeUnknown)) + } + if report.UnexpectedMessages > 0 { + report.FailureReasons = append(report.FailureReasons, + fmt.Sprintf("deliveries observed without ACK ledger=%d", report.UnexpectedMessages)) + } if report.AckP95MS > 500 { report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ack p95=%.3fms expected<=500ms", report.AckP95MS)) } @@ -119,7 +134,7 @@ func runMerge(aPath, bPath, output string) error { report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("delivery p95=%.0fms expected<=1000ms", report.DeliveryP95MS)) } - if report.PublishRejected > 0 || report.RealtimeLost > 0 || report.PersistenceLost > 0 || report.Duplicates > 0 || report.OutOfOrder > 0 || report.Disconnects > 0 { + if report.PublishRejected > 0 || report.RealtimeLost > 0 || report.PersistenceLost > 0 || report.Duplicates > 0 || report.OutOfOrder > 0 || report.Disconnects > 0 || report.UnexpectedMessages > 0 { report.FailureReasons = append(report.FailureReasons, "message correctness gate failed") } report.Passed = len(report.FailureReasons) == 0 diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index 2f1dd81..7878d30 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -12,8 +12,9 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { dir := t.TempDir() a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ReadyConnections: 1000, ExpectedConnections: 1000, - PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9, - RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10} + PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9, UnexpectedMessages: 42, + PublishErrors: map[string]int64{"ctrl_timeout": 3, "socket": 2, "503": 1}, + RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10} b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000, ExpectedConnections: 1000, PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, @@ -39,6 +40,9 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { if report.Passed || report.PersistenceLost != 1 { t.Fatalf("correctness gate did not fail: %+v", report) } + if report.UnexpectedMessages != 42 || report.PublishOutcomeUnknown != 5 || report.PublishExplicitReject != 1 { + t.Fatalf("ambiguous publish outcomes were not merged: %+v", report) + } if len(report.OrderingViolations) != 1 || report.OrderingViolations[0].ReceiverIndex != 7 { t.Fatalf("ordering violations were not merged: %+v", report.OrderingViolations) }