diff --git a/docs/test/IM-group2k-r4-群聊复测分析-20260928.md b/docs/test/IM-group2k-r4-群聊复测分析-20260928.md index c5ed03c..6691888 100644 --- a/docs/test/IM-group2k-r4-群聊复测分析-20260928.md +++ b/docs/test/IM-group2k-r4-群聊复测分析-20260928.md @@ -128,7 +128,7 @@ loadtest/im-pressure/results/group2k-r4-20260928-01/ 1. agent 分开输出 `publish_rejected` 和 `publish_outcome_unknown`;前者只包含服务端明确拒绝,后者包含 ACK 超时和 socket 失败。 2. A/B 在 drain 后向 coordinator 提交“实时已观测但未 ACK”的消息台账,coordinator 去重合并后将其加入历史落库核验。 3. 报告新增 `history_unacked_observed_checked` 和 `history_unacked_observed_found`,可直接确认结果未知消息是否最终落库。 -4. 历史回查默认调整为 1 个 worker、每页 50 条、全局每 100ms 最多发起一次查询,避免核验程序自身制造 Topic/Proxy 洪峰。 +4. r5 复测后,历史回查进一步调整为 1 个 worker、每页 20 条、全局每 500ms 最多发起一次查询,避免核验程序自身制造 Topic/Proxy 洪峰。 5. 新增 `backend-debug-metrics-rbac.yaml`,为测试环境的 `argocd/backend-debug` ServiceAccount 提供 `metrics.k8s.io/pods` 的 namespace 只读权限。 上述改造只解决统计口径、历史查询压力和资源指标权限。机器 A 的公网连接异常仍需通过下一轮主机与网络观测定位。 diff --git a/docs/test/IM-group2k-r5-群聊复测分析-20260928.md b/docs/test/IM-group2k-r5-群聊复测分析-20260928.md new file mode 100644 index 0000000..673483b --- /dev/null +++ b/docs/test/IM-group2k-r5-群聊复测分析-20260928.md @@ -0,0 +1,92 @@ +# IM 2000 人群聊 r5 复测分析 + +日期:2026-09-28 + +Run ID:`group2k-r5-20260928-01` + +## 1. 结论 + +本轮再次证明服务端对稳定客户端的群消息可靠性没有丢失,失败集中在机器 A 的本地代理/TUN 链路: + +- B:1000 次发送全部 ACK,1426000 次 ACK 台账内投递全部到达,断线、重复、乱序和实时缺失均为 0。 +- A:574 次发布结果未知、657 次统计窗口内断线、464596 次 ACK 台账内实时缺失。 +- A 的 broken pipe 全部写向 `198.18.0.81:443`;本机 DNS 也将 `api.118im.com` 解析为 `198.18.0.81`,路由经 `utun19827 -> 172.19.0.1`。`198.18.0.0/15` 是代理/TUN 常用 Fake-IP 网段,不是 API 的真实公网地址。 +- 机器 A 压测进程 CPU 平均占用不高、内存峰值约 150MB,B 在同一服务端负载下完全稳定,不支持“服务端全局过载”解释。 +- 新增观测台账生效:574 条结果未知中,241 条被实时观测并加入历史核验,241 条全部找到。 +- 历史总核验 1667/1667,持久化丢失为 0。 +- 历史回查仍产生 1197 次 `topic broadcast queue full`,但全部发生在历史阶段,不在实时发送阶段。 + +总体结论:**群消息落库通过,B 侧实时可靠性通过;A 侧 TUN/Fake-IP 链路失败;历史查询限速仍需继续降低。** + +## 2. 汇总结果 + +| 指标 | A | B | 合计 | +|---|---:|---:|---:| +| READY | 1000 | 1000 | 2000 | +| 发送尝试 | 1000 | 1000 | 2000 | +| ACK | 426 | 1000 | 1426 | +| 明确拒绝 | 0 | 0 | 0 | +| ACK 超时 | 444 | 0 | 444 | +| socket 结果未知 | 130 | 0 | 130 | +| 断线 | 657 | 0 | 657 | +| ACK 台账内应收 | 1426000 | 1426000 | 2852000 | +| ACK 台账内实收 | 961404 | 1426000 | 2387404 | +| ACK 台账内实时缺失 | 464596 | 0 | 464596 | +| ACK 台账外额外投递 | 185738 | 241000 | 426738 | +| 历史核验 | 1667/1667 | 0/0 | 1667/1667 | +| 未 ACK 观测消息核验 | 241/241 | 0/0 | 241/241 | + +服务端实际处理消息至少为 `1426 + 241 = 1667` 条。其余 333 条结果未知既没有 ACK,也没有任何实时观测证据,不能证明服务端接受。 + +## 3. A 机器故障证据 + +### 3.1 网络路径 + +测试结束后的本机检查结果: + +```text +api.118im.com -> 198.18.0.81 +route: 198.18.0.81 -> gateway 172.19.0.1, interface utun19827 +broken pipe remote: 198.18.0.81:443 +``` + +A 的错误构成为: + +- `ctrl_timeout=444` +- `websocket_closed=103` +- `broken_pipe=27` +- 其他 1 + +连接从 `17:05:12 UTC` 开始下降,随后从 1000 持续降至约 333;B 同期保持 1000 连接。下一轮前必须关闭该代理/TUN,或至少将 `api.118im.com` 设置为 DIRECT,并确认 DNS 不再返回 `198.18.0.0/15` Fake-IP。 + +### 3.2 本机资源 + +- 进程 RSS 峰值约 150MB。 +- 压测期间累计 CPU 增量约 75 秒,对约 625 秒窗口而言不构成 CPU 饱和。 +- open FDs 在连接大量断开后仍约为 1011。`WSClient.readLoop` 在读失败返回时没有立即关闭底层连接,虽然进程结束时会统一关闭,但应修复运行期 FD 滞留。 + +## 4. 服务端和历史核验 + +实时发送和 drain 阶段没有新增以下错误: + +- `sub.broadcast channel full` +- `cluster: proxy channel full` +- `topic's broadcast queue is full` +- Redis fanout publish failure + +历史核验约在 `17:13:15 UTC` 开始。`im-server-2` 从 `17:13:22` 到 `17:14:14` 记录 1197 次群 Topic broadcast queue full。当前 1 worker、每页 50 条、100ms 间隔仍会让单次历史回放快速灌入容量为 64 的 Topic 队列。 + +尽管重试后 1667 条全部找到,历史核验不应持续制造队列丢弃。建议调整为: + +- `history_page_size=20` +- `history_query_interval=500ms` +- `history_workers=1` +- 历史超时保持 3 分钟;1667 条约需 42 秒完成首轮查询。 + +## 5. 下一步 + +1. 修复 `WSClient.readLoop` 退出时未关闭底层连接的问题。 +2. 将历史分页调整为 20 条、查询间隔 500ms。 +3. 在 A 机关闭 TUN/代理,或配置 `api.118im.com` DIRECT;起压前检查 DNS 和路由不再进入 `198.18.0.0/15`、`utun*`。 +4. 测试环境尚未应用 metrics RBAC,`kubectl auth can-i list pods.metrics.k8s.io` 仍返回 `no`,因此本轮没有 Pod CPU/内存数据。需要将 `backend-debug-metrics-rbac.yaml` 加入实际 ArgoCD overlay。 +5. 完成上述三项后再运行同规模 r6。若 A/B 都稳定,再根据 B 本轮 ACK P95 约 1.14s、同机投递 P95 位于 2s 桶的结果继续做服务端延迟优化。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index a501668..a56da17 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -52,7 +52,7 @@ - 群消息设置 `noecho=false`,每条服务端接受的群消息期望 2000 个连接各收到一次。 - 实时 drain 结束后,两机先汇总 ACK 台账和“已实时观测但未收到 ACK”的消息台账;等待 `history_settle`(默认 5 分钟)让 Kafka 自然排空,再通过 `{get what=data}` 按 Topic/Seq 回查两类消息。 - 每轮建立 WS 后、正式计时前,以 32 个 worker 对本机 1000 个账号做好友关系、P2P Topic 和发布权限预检;任一异常会终止本轮。 -- 历史回查默认每机使用 1 个 worker、每页 50 条、查询间隔 100ms、总超时 3 分钟,并在超时内持续重试缺失区间。该限速用于避免正确性核验本身制造 Topic/Proxy 洪峰;历史读取吞吐应使用独立场景测试。 +- 历史回查默认每机使用 1 个 worker、每页 20 条、查询间隔 500ms、总超时 3 分钟,并在超时内持续重试缺失区间。该限速用于避免正确性核验本身制造 Topic/Proxy 洪峰;历史读取吞吐应使用独立场景测试。 - publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`。`publish_rejected` 只统计明确拒绝,超时和 socket 错误单独计入 `publish_outcome_unknown`。 - `delivery_local_p95_ms` 使用发送方本机收到 echo 的同机时钟样本,是投递延迟门禁口径;兼容字段 `delivery_p95_ms` 与它保持一致。 - `delivery_cross_machine_p95_ms` 是跨机器单向观测值,只有两机时钟同步达标时才有意义;`cross_machine_negative_latency_samples > 0` 明确表示存在时钟偏差。 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 3c7ba01..e31d7a0 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -390,7 +390,7 @@ func (r *agentRun) issueHistoryQueries(ctx context.Context, queries []historyQue return } jobs := make(chan historyQuery) - interval := 100 * time.Millisecond + interval := 500 * time.Millisecond if r.cfg != nil && r.cfg.HistoryInterval != "" { interval = mustDuration(r.cfg.HistoryInterval) } @@ -430,7 +430,7 @@ func (r *agentRun) historyPageSize() int { if r.cfg != nil && r.cfg.HistoryPageSize > 0 { return r.cfg.HistoryPageSize } - return 50 + return 20 } func missingHistory(recorder *deliveryRecorder, expected map[string]AcceptedMessage) []AcceptedMessage { diff --git a/loadtest/im-pressure/config.example.json b/loadtest/im-pressure/config.example.json index fe8b265..649f25d 100644 --- a/loadtest/im-pressure/config.example.json +++ b/loadtest/im-pressure/config.example.json @@ -18,8 +18,8 @@ "preflight_workers": 32, "skip_p2p_preflight": false, "history_workers": 1, - "history_page_size": 50, - "history_query_interval": "100ms", + "history_page_size": 20, + "history_query_interval": "500ms", "history_settle": "5m", "history_timeout": "3m", "p2p_messages_per_user": 10, diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index da4aadc..cc73a89 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -90,13 +90,13 @@ func loadConfig(path string) (*Config, error) { return nil, errors.New("history_workers must not exceed 128") } if cfg.HistoryPageSize <= 0 { - cfg.HistoryPageSize = 50 + cfg.HistoryPageSize = 20 } if cfg.HistoryPageSize > 200 { return nil, errors.New("history_page_size must not exceed 200") } if cfg.HistoryInterval == "" { - cfg.HistoryInterval = "100ms" + cfg.HistoryInterval = "500ms" } if _, err = configDuration(cfg.HistoryInterval); err != nil { return nil, fmt.Errorf("history_query_interval: %w", err) diff --git a/loadtest/im-pressure/model_test.go b/loadtest/im-pressure/model_test.go index 51668ab..8897e45 100644 --- a/loadtest/im-pressure/model_test.go +++ b/loadtest/im-pressure/model_test.go @@ -16,7 +16,7 @@ func TestLoadConfigAppliesBoundedHistoryDefaults(t *testing.T) { if err != nil { t.Fatal(err) } - if cfg.HistoryWorkers != 1 || cfg.HistoryPageSize != 50 || cfg.HistoryInterval != "100ms" || + if cfg.HistoryWorkers != 1 || cfg.HistoryPageSize != 20 || cfg.HistoryInterval != "500ms" || cfg.HistorySettle != "5m" || cfg.HistoryTimeout != "3m" || cfg.PreflightWorkers != 32 || cfg.WSPingInterval != "20s" { t.Fatalf("unexpected defaults: %+v", cfg) diff --git a/loadtest/im-pressure/wsclient.go b/loadtest/im-pressure/wsclient.go index 8508523..9bbc454 100644 --- a/loadtest/im-pressure/wsclient.go +++ b/loadtest/im-pressure/wsclient.go @@ -344,7 +344,10 @@ func (c *WSClient) removeWait(id string) { } func (c *WSClient) readLoop() { - defer close(c.done) + defer func() { + _ = c.conn.Close() + close(c.done) + }() for { var envelope map[string]json.RawMessage if err := c.conn.ReadJSON(&envelope); err != nil { diff --git a/loadtest/im-pressure/wsclient_test.go b/loadtest/im-pressure/wsclient_test.go index d86ee68..6e98984 100644 --- a/loadtest/im-pressure/wsclient_test.go +++ b/loadtest/im-pressure/wsclient_test.go @@ -48,3 +48,39 @@ func TestWSClientKeepaliveSendsPing(t *testing.T) { t.Fatal("websocket keepalive ping was not received") } } + +func TestWSClientReadLoopClosesConnectionAfterDecodeError(t *testing.T) { + serverSawClose := make(chan struct{}, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + if err = conn.WriteMessage(websocket.TextMessage, []byte("{")); err != nil { + return + } + if _, _, err = conn.ReadMessage(); err != nil { + serverSawClose <- struct{}{} + } + })) + defer server.Close() + + conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil) + if err != nil { + t.Fatal(err) + } + client := &WSClient{conn: conn, waits: make(map[string]chan ctrlPacket), done: make(chan struct{})} + go client.readLoop() + + select { + case <-client.done: + case <-time.After(time.Second): + t.Fatal("read loop did not exit after malformed frame") + } + select { + case <-serverSawClose: + case <-time.After(time.Second): + t.Fatal("read loop exited without closing the underlying connection") + } +}