From 73a594146f356704dcaff695b4fb54a7cf604903 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 1 Oct 2026 14:44:41 +0800 Subject: [PATCH] fix: trace delayed websocket acknowledgements --- docs/test/IM-单机60%容量压力测试方案.md | 7 +- .../IM-单机60%容量压力测试结果-20260929.md | 255 +++++++++++++++++- loadtest/im-pressure/agent.go | 156 +++++++++-- loadtest/im-pressure/agent_metrics.go | 4 + loadtest/im-pressure/callback_queue_test.go | 17 ++ loadtest/im-pressure/config.example.json | 4 +- .../config.single-machine.example.json | 10 +- loadtest/im-pressure/merge.go | 106 ++++---- loadtest/im-pressure/merge_test.go | 17 +- loadtest/im-pressure/model.go | 126 ++++++--- loadtest/im-pressure/model_test.go | 28 ++ loadtest/im-pressure/wsclient.go | 126 +++++++-- loadtest/im-pressure/wsclient_test.go | 50 +++- server/cluster.go | 1 + server/cluster_gob_test.go | 1 + server/hdl_websock.go | 6 + server/metrics/metrics.go | 14 +- server/monitoring_http_test.go | 2 + server/session.go | 7 + 19 files changed, 788 insertions(+), 149 deletions(-) diff --git a/docs/test/IM-单机60%容量压力测试方案.md b/docs/test/IM-单机60%容量压力测试方案.md index ec40eff..64c477b 100644 --- a/docs/test/IM-单机60%容量压力测试方案.md +++ b/docs/test/IM-单机60%容量压力测试方案.md @@ -129,6 +129,8 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}" 两个 agent 都报告 `600/600 READY` 并稳定通过 30 秒屏障后才会开始。任一 agent 未满连接时 coordinator 会撤销开始时间。 +固定 drain 结束后，agent 还会等待入站投递连续静默后再生成实时 summary。默认 `delivery_quiet_period=10s`、`delivery_quiet_timeout=3m`；接收计数增长或回调队列不为空都会重置静默窗口，避免公网晚到帧被误判为实时丢失。 + ## 6. 指标采集 终端 4 在起压前至少 2 分钟启动： @@ -148,6 +150,8 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}" 发版后必须在服务端 `/metrics` 确认 `im_ws_heartbeat_seconds{setting="ping_period"}=15`、`im_ws_heartbeat_seconds{setting="pong_wait"}=90`，并采集 `im_ws_heartbeat_total{event="pong_received"}` 和 `im_ws_disconnect_total{reason="timeout"}` 的分 Pod 增量。 +混合场景还必须采集 `im_ctrl_ack_delivery_duration_seconds{stage="enqueue"}` 和 `im_ctrl_ack_delivery_duration_seconds{stage="write"}`。agent 端同时采集 `im_load_agent_publish_ctrl_observed_total`、`im_load_agent_publish_ctrl_over_20s_total`、`im_load_agent_publish_ctrl_late_total` 和 `im_load_agent_publish_ctrl_unresolved`，用于区分及时 ACK、超过 20 秒的 ACK、超过命令超时后晚到的 ACK 与最终未到 ACK。 + ## 7. 报告和独立落库核验 ```bash @@ -170,7 +174,8 @@ export TENANT='006' # 必须与 config.single-machine.json 中的 tenant 一致 - READY 必须是 `1200/1200`，统计窗口内断连为 0。 - publish 明确拒绝和结果未知均为 0，成功率至少 99.99%。 -- ACK P95 不高于 500ms，同机投递 P95 不高于 1000ms。 +- `publish_ctrl_unresolved` 必须为 0；`publish_ctrl_over_20s` 和 `publish_ctrl_late` 作为定位指标单独报告，不与消息丢失混为一项。 +- ACK P95 和投递 P95 保留记录用于趋势对比，不作为当前公网单机容量测试的阻断门禁。 - 实时缺失、重复、乱序、未知本轮消息均为 0。 - ACK 台账和无 ACK 但已观测台账的历史回查命中率必须 100%，独立 MySQL 核验丢失、错配、重复均为 0。 - 实时发送、历史回查和渐进关闭三个阶段分别统计 Topic/Proxy/Session 队列满，不能只看整轮总数。 diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index 35e063e..7ddd2e1 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -6,7 +6,7 @@ 本轮使用一台物理压测机同时运行逻辑 agent A/B，每侧 600 个 WebSocket，合计 1200 连接。60% 容量的正确性和性能总验收未通过。 -- 单聊实时和持久化正确性通过，但 ACK P95 和投递 P95 未达标。 +- 单聊实时和持久化正确性通过；ACK P95 和投递 P95 仅记录，不再作为当前公网单机压测的阻断条件。 - 群聊在 1200 条群消息、144 万设计投递量下出现 ACK 超时、连接重置和分钟级投递延迟。 - 混合场景明显越过容量拐点，737 个连接断开，仅 44.83% 发送得到 ACK。 - 独立 MySQL 核验确认：三个场景的所有 ACK 消息均精确落库，0 错配、0 重复；所有“无 ACK 但 WS 已观测”消息也均已落库。 @@ -30,7 +30,7 @@ | 场景 | Run ID | ACK | 实时投递 | 断连 | ACK P95 | 投递 P95 | 独立 DB 核验 | 结果 | |---|---|---:|---:|---:|---:|---:|---:|---| -| 单聊 | `s60-p2p-r2-0929` | 12000/12000 | 12000/12000 | 0 | 1341.555ms | 2000ms | 12000 ACK 全部精确命中 | 性能未通过 | +| 单聊 | `s60-p2p-r2-0929` | 12000/12000 | 12000/12000 | 0 | 1341.555ms | 2000ms | 12000 ACK 全部精确命中 | 正确性通过 | | 群聊 | `s60-group-r1-0929` | 1145/1200 | 1350508/1374000 ACK 台账 | 47 | 11207.859ms | 94621.882ms | 1145 ACK + 39 无 ACK 落库 | 未通过 | | 混合 | `s60-mixed-r1-0929` | 1076/2400 | 234829/263657 ACK 台账 | 737 | 17491.634ms | 101226.392ms | 1076 ACK + 1017 无 ACK 落库 | 未通过 | @@ -45,7 +45,7 @@ R1 的历史查询将 `limit=10` 错误地按目标消息数计算，没有计 - 12000 条全部 ACK，无结果未知。 - 12000 次对端投递全部到达，无重复、乱序、断连。 - 12000 条历史回查全部命中，持久化缺失 0。 -- ACK P95 1341.555ms，超过 500ms 门禁；同机投递 P95 落在 2000ms 桶，超过 1000ms 门禁。 +- ACK P95 1341.555ms，同机投递 P95 落在 2000ms 桶；两项作为公网链路观测数据，不影响正确性验收。 ### 4.2 群聊 @@ -396,3 +396,252 @@ Run ID：`s900m-heart-r1-0930`。三个 Pod 发版后均为 Ready、零重启， - `im_ws_heartbeat_total{event="ping_received"}` 有持续增量，服务端 timeout 和客户端断连进一步下降。 - `im_db_invalid_connection_retry_total` 若出现，结果应为 `discard` 或 `recovered`，客户端不再收到对应 500。 - `callback_queue.current` 和 `callback_queue.processing` 在结算时均为 0；结合 `maximum`、`wait_seconds_total / completed` 和 `wait_seconds_max` 区分压测端排队与服务端投递延迟。 + +## 20. 第二轮修复后精确 900 人复测 + +Run ID：`s900m-activity-r1-0930`，部署提交 `36129e3`。复用精确 900 人群 `grpQ5_hPdAz3cE`，A/B 各 450 条 WebSocket 连接。两个 agent 均通过 30 秒满连接屏障，并一直运行至实时投递、5 分钟沉淀、历史回查、MySQL 校验和 120 秒渐进关闭全部完成。 + +| 指标 | 结果 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 发送尝试 / 收到 ACK | 900 / 899 | +| 结果未知 | 1 条 ACK 超时，socket 错误 0 | +| 服务端实际接收并落库 | 900 / 900 | +| 实际实时投递 | 810000 / 810000 | +| 实时到达率 | 100% | +| 重复 / 乱序 | 0 / 0 | +| 压测期非预期断连 | 0 | +| 服务端 WS timeout | 0 | +| 历史回查 | 900 / 900 | +| MySQL 精确命中 | 900 / 900 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| ACK P95 | 4748.423ms | +| 投递 P95 | 10000ms 桶内 | + +唯一 ACK 超时为 `s900m-activity-r1-0930-g-b-0121-01`。该消息虽然 20 秒内没有收到 ctrl ACK，但 A/B 均实时观察到，服务端分配 seq `3714`，历史可查且 MySQL 精确命中；因此是 ACK 返回链路超时，不是消息未入服或丢失。报告中的 `unexpected_messages=900` 正是该条没有进入 ACK ledger、但被 900 个连接实际收到的消息。 + +本轮连接活跃改造生效。三 Pod 均持续记录 `ping_received` 和 `pong_received`，负载阶段没有 WS timeout；两个 agent 全程 450/450 在线且断连为 0。与上一轮 11 次断连相比，客户端 Ping 和业务帧续期后已消除本轮心跳误断。 + +压测端回调队列指标证明 recorder 不再构成秒级延迟： + +| agent | 入队 / 完成 | 峰值等待数 | 最大等待 | 平均等待 | 结算时等待 / 处理中 | +|---|---:|---:|---:|---:|---:| +| A | 405000 / 405000 | 106 | 11.353ms | 约 0.022ms | 0 / 0 | +| B | 405000 / 405000 | 98 | 5.075ms | 约 0.020ms | 0 / 0 | + +服务端内部处理也没有出现容量积压： + +- Topic queue wait P95 不超过 50ms，全部 900 条不超过 250ms。 +- permission check P95 不超过 25ms，消息保存 P95 不超过 50ms，realtime fanout P95 不超过 5ms，push prepare P95 不超过 10ms。 +- WS send queue wait P95 不超过 25ms，socket text write P95 不超过 0.5ms。 +- push dispatch queue 无 full，Kafka consumer lag 和 oldest backlog age 均为 0。 +- 没有 `invalid connection`、消息保存失败、内部队列 full、panic 或 Pod 重启；关闭后 push 队列和 `topic_presence_pending` 均回落为 0。 + +因此本轮消息正确性和持久化门禁已经通过，但端到端延迟门禁仍未通过。测试结束后的三入口公网 RTT 为 153–189ms，HTTPS TTFB 为 0.50–1.04s；三条路由均经 `en1` 和网关 `192.168.1.1` 直连。服务端从 Topic 排队到 socket write 的 P95 远低于端到端 ACK/投递 P95，压测回调队列最大等待也只有 11ms，剩余延迟主要位于服务端 socket 写入之后的 Ingress/LB、公网 TCP、NAT 或压测机网络路径。后续服务端容量验收应在集群同地域压测机执行，并同时采集 Ingress/LB 出站带宽、TCP 重传、reset 和 upstream 指标。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-activity-r1-0930/report.json` +- `loadtest/im-pressure/results/s900m-activity-r1-0930/db-verification.json` +- `loadtest/im-pressure/results/s900m-activity-r1-0930/metrics.jsonl` +- `loadtest/im-pressure/results/s900m-activity-r1-0930/server-metrics/` + +## 21. 精确 900 人混合场景测试 + +混合场景使用 900 条 WebSocket、135 对单聊用户和 224 个群发送者。单聊双方各发 5 条，共 1350 条；群发送者各发 2 条，共 448 条；总发送尝试 1798 条。每个 agent 理论应观察 201600 次群投递和 1350 次单聊发送方回显/接收方投递，共 202950 次唯一回调。 + +### 21.1 R1：发现结算窗口过早 + +Run ID：`s900m-mixed-r1-1001`。 + +- 1798 条消息全部进入服务端，历史 1798/1798，MySQL 1798/1798，缺失、错配、重复和额外记录均为 0。 +- 收到 ACK 1411 条，387 条 ctrl ACK 超时；ACK P95 为 15756.928ms。 +- A/B 负载期均为 450 条连接且断连为 0。 +- 固定 60 秒 drain 结束时，A 的 summary 记录 202867/202950，误报 83 次实时缺口；summary 生成后指标继续增长至 202950/202950，说明这些帧是公网链路中的晚到帧，并未丢失。 +- B 为 202950/202950。回调队列最大等待分别约 98ms/92ms，结算时等待和处理中均为 0。 +- 发现 1 次单聊顺序回退：接收方 A-27 在同一 P2P Topic 上先收到 seq 245、后收到 seq 244。 + +本轮证明只等待“回调队列瞬时为空”不够：帧可能仍位于服务端 socket、Ingress/LB、TCP 或客户端 socket 缓冲区。压测程序已增加 `delivery_quiet_period` 和 `delivery_quiet_timeout`。固定 drain 后，接收计数和回调队列必须连续静默 10 秒才允许生成实时 summary；任意晚到帧都会重置静默窗口，最长等待 3 分钟。 + +同时修正实时统计口径：无 ACK 但已被任一 WebSocket 观察到的消息会与 ACK ledger 合并后参与全部收件人的实时到达率和“实时缺失但已落库”计算，不再被记为 `unexpected_messages`。该口径从后续轮次开始生效；R1/R2 原始报告仍保留当时输出，以上结论已按全量唯一回调和历史/MySQL 结果校正。 + +### 21.2 R2：静默结算修复验证 + +Run ID：`s900m-mixed-r2-1001`。本轮使用修复后的压测程序，静默等待在 A/B 均正常执行并完成后才生成 summary。 + +| 指标 | 结果 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 发送尝试 / ACK | 1798 / 1760 | +| ACK 超时 / socket 错误 | 37 / 1 | +| 压测期断连 | 14，全部为 timeout，A/B 各 7 | +| A 唯一实时回调 | 202535 / 202950 | +| B 唯一实时回调 | 202787 / 202950 | +| 全量唯一实时回调 | 405322 / 405900，约 99.8576% | +| ACK 台账实时缺口 | 566，全部已落库 | +| 重复 / 乱序 | 0 / 0 | +| 历史回查 | 1798 / 1798 | +| MySQL 精确命中 | 1798 / 1798 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| ACK P95 | 14317.159ms | +| 投递 P95 | 46622.373ms | + +R2 的 A/B 各 7 条连接在负载期超时断开，随后全部自动恢复至 450/450。实时缺口与断连窗口一致，静默窗口结束后接收计数不再增长，因此不是 R1 那种过早结算。所有实时缺口均可由历史补回，MySQL 独立校验证明服务端持久化零丢失。 + +两个 agent 的回调队列最大等待约 10ms，结算时 current/processing 均为 0，压测 recorder 不是本轮瓶颈。服务端 1350 条 P2P 消息全部处理完成，P2P Topic queue P95 位于 250–500ms，relationship check P95 位于 100–250ms；Kafka lag 为 0，未出现 MySQL invalid connection、内部队列 full 或 Pod 重启。关闭后 push 队列和 deferred presence 均回落为 0。 + +混合场景在当前公网压测机上未通过实时和延迟门禁，但持久化门禁通过。下一步不应直接扩大到 1200 人，应先在集群同地域压测机重复 900 人混合场景；若同地域仍出现乱序或实时缺口，再分别处理跨 Pod P2P 顺序和 WS 投递路径。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-mixed-r1-1001/` +- `loadtest/im-pressure/results/s900m-mixed-r2-1001/` + +### 21.3 R3：零断连、零实时丢失，但仍有 ctrl ACK 超时 + +Run ID：`s900m-mixed-r3-1001`。本轮沿用 900 条 WebSocket、135 对单聊用户和 224 个群发送者，共发送 1350 条单聊和 448 条群消息。端到端 P95 延迟按最新口径仅作观测，不参与通过判定。 + +| 指标 | R3 | R2 | +|---|---:|---:| +| 就绪连接 | 900 / 900 | 900 / 900 | +| 发送尝试 / ACK | 1798 / 1751 | 1798 / 1760 | +| ctrl ACK 超时 / socket 错误 | 47 / 0 | 37 / 1 | +| 压测期非预期断连 | 0 | 14 | +| 两端唯一 WS 回调 | 405900 / 405900 | 405322 / 405900 | +| 实时缺口 | 0 | 578 次底层回调，ACK 台账折算缺口 566 | +| 重复 / 乱序 | 0 / 0 | 0 / 0 | +| 历史回查 | 1798 / 1798 | 1798 / 1798 | +| 无 ACK 但已观察消息的历史命中 | 47 / 47 | 38 / 38 | +| MySQL 精确命中 | 1798 / 1798 | 1798 / 1798 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | 0 / 0 / 0 / 0 | +| ACK P95（仅观测） | 14020.142ms | 14317.159ms | +| 投递 P95（仅观测） | 52784.346ms | 46622.373ms | + +R3 的消息正确性明显好于 R2：A/B 各收到 202950/202950 个唯一回调，包括每端 201600 次群投递、675 次单聊对端投递和 675 次发送方回显。固定 drain 后又经过连续 10 秒静默窗口，计数未再变化，本轮没有过早结算。 + +47 条 ctrl ACK 超时中，单聊 26 条、群聊 21 条，A/B 分别为 22/25 条。它们全部被 WS 实时观察，47/47 历史可查，并被 MySQL 独立校验纳入 1798/1798 精确命中，所以不是消息未入服或持久化丢失。但客户端在 20 秒内未收到 ctrl ACK，仍是独立的发送协议问题，不属于已取消的 500ms 延迟门禁。因此本轮应判定为： + +- 实时到达、重复、顺序、断连、历史和落库正确性通过。 +- 混合场景总验收仍未通过，唯一阻断为 47 条 `command_timeout=20s` 的 ctrl ACK 超时。 +- ACK P95 和投递 P95 保留在报告中，不影响上述判定。 + +服务端负载期没有 Topic/WS 发送队列 full，WS send queue 结算时均为 0，所有 Kafka consumer lag 为 0，三个 Pod 均为 Ready 且零重启。连接渐进关闭后，三 Pod 连接数从 518/373/550 回落至 7/2/7，与起压前 8/3/6 基本一致。Pod 2 增加了 1 次 `Cluster.Route` 和 1 次 `Cluster.UserCacheUpdate` 瞬时 error，均不足 1ms，最近日志无对应错误，且未影响本轮投递与落库。 + +压测端回调队列在结算时均为 0；A/B 队列峰值为 355/174，最大等待仅为 37.054ms/25.036ms，不是 20 秒 ACK 超时或 52 秒投递 P95 的原因。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-mixed-r3-1001/report.json` +- `loadtest/im-pressure/results/s900m-mixed-r3-1001/db-verification.json` +- `loadtest/im-pressure/results/s900m-mixed-r3-1001/metrics.jsonl` +- `loadtest/im-pressure/results/s900m-mixed-r3-1001/server-metrics/` + +### 21.4 R3 ctrl ACK 超时原因分析 + +#### 21.4.1 代码路径 + +服务端的消息发布顺序为：Topic 队列出队、权限/关系校验、持久化、生成 `NoErrAccepted` ctrl ACK 并调用 `queueOut`，然后才做当前消息的 Topic 广播、跨 Pod 实时 fanout 和离线推送准备。因此，当前消息的大规模 fanout 不会阻塞同一条消息的 ACK 生成。 + +但所有 ctrl 和 data 最终共用用户的同一个 WebSocket/TCP 有序字节流。ACK 可以在当前消息的 data 之前入队，但无法超过已经写入 socket 或正在 Ingress/LB/公网 TCP 中传输的早先 data 帧。混合场景中发送者同时也是 900 人群的收件人，因此每条发送连接在等待自己 ACK 时，前面可能已有数百条群 data 帧。 + +#### 21.4.2 服务端分段证据 + +按起压前与负载结束 Prometheus 快照计算本轮增量： + +| 阶段 | 增量样本 | P95 | +|---|---:|---:| +| P2P Topic 队列等待 | 1350 | 100ms 桶内 | +| P2P 关系校验 | 1350 | 100ms 桶内 | +| P2P 保存 | 1350 | 50ms 桶内 | +| 群 Topic 队列等待 | 448 | 25ms 桶内 | +| 群权限校验 | 448 | 25ms 桶内 | +| 群保存 | 448 | 50ms 桶内 | +| Cluster TopicMaster RPC | 2495 | 3至5ms 桶内 | +| Cluster proxy 队列深度 | 7988 | P95=1 | +| WS 发送队列深度 | 817924 | P95=0 | +| WS 发送队列等待 | 817924 | 10ms 桶内 | +| WS text frame 写入 | 817946 | 0.5ms 桶内 | + +本轮三 Pod 共成功写出约 81.8 万个文本帧、193.6MB。服务端应用层队列、保存、Cluster RPC 和 socket write 均远低于 20 秒；压测时间窗内无 queue full、write timeout、数据库错误、Pod 重启或 panic。因此没有证据支持“业务处理或服务端应用队列阻塞 20 秒”。 + +#### 21.4.3 超时分布证据 + +- 47 条超时全部发生在起压后 20至40 秒，正好是每条连接已经接收约 200至400 条群广播帧的区间。 +- 40/47 是同一用户的第 2 条消息，5 条是第 3 条，只有 2 条是第 1 条；超时不是随机散布在整个测试周期。 +- 三个入口 IP 分别有 16/16/15 条超时，排除单一 Ingress IP 热点。 +- 压测端回调队列最大等待只有 37ms，数据回调已与 socket 读循环解耦；超时不是 recorder 锁竞争或回调队列堵塞。 +- 测试后三个入口的当前 ICMP RTT 约为 167至197ms，HTTPS TTFB 约为 0.49至0.61 秒；路由均经 `en1` 直连，未经 VPN/TUN。 + +综合上述证据，当前根因判断为：**高置信度的 WebSocket/TCP 单流队头阻塞，延迟主要发生在服务端完成 socket write 之后的 Ingress/LB、公网 TCP、NAT 和压测机接收路径。**这个结论可以解释“服务端已快速保存且客户端最终收到全部 data，但部分 ctrl 在 20 秒窗口内未被读到”。 + +尚未完全证明的环节是：当前压测程序在 ctrl 超时后删除 wait entry，晚到 ctrl 会被静默忽略；服务端 WS 写指标也没有区分 ctrl 和 data。因此现有数据无法逐条证明 47 个 ACK 全部在 20 秒后到达，只能证明对应消息全部已被服务端接受、投递和持久化。 + +#### 21.4.4 处理建议 + +1. 压测端增加 late ctrl 台账：ctrl 超时后保留 ID 和发送时间 10 分钟，记录晚到 ACK 数、最终延迟和完全未到数。 +2. 服务端将 WS 出站指标按 `ctrl/data/pres/info` 分类，新增从客户端请求时间到 ctrl ACK 入队、实际 socket write 完成的直方图；跨 Pod 时再记录 master-to-proxy ctrl 延迟。 +3. 下一轮保持完全相同的 900 人混合负载，将 `command_timeout` 改为 90 秒，但单独统计超过 20 秒的 ACK。若 47 类消息都在 90 秒内到达，且服务端 ctrl write 延迟仍在毫秒级，则队头阻塞结论得到直接验证。 +4. 当前业务不对端到端延迟设阻断门禁，因此不建议现在改造独立控制连接或破坏 WebSocket 有序性。应先补齐上述观测，并将“消息已实时观测且历史/MySQL 命中”与“20 秒内收到 ctrl ACK”分开报告。 +5. 另有一项与 ACK 超时无关的独立缺陷：负载结束后 Pod 2 记录 `gob: type not registered for interface: []map[string]interface {}`，影响另一个群 Topic 的 Cluster Route。当前 cluster gob 已注册 `[]any` 和 `map[string]any`，但未注册 `[]map[string]any`，应单独补注册和编解码回归测试。 + +#### 21.4.5 观测改造完成情况与 R4 口径 + +上述观测缺口已补齐： + +- 压测端会统计 `publish_ctrl_observed`、`publish_ctrl_over_20s`、`publish_ctrl_late`、`publish_ctrl_unresolved` 和 `publish_ctrl_p95_ms`，并保留最多 1000 条超过 20 秒或晚到的样本。 +- agent Prometheus 新增 `im_load_agent_publish_ctrl_observed_total`、`im_load_agent_publish_ctrl_over_20s_total`、`im_load_agent_publish_ctrl_late_total` 和 `im_load_agent_publish_ctrl_unresolved`。 +- 服务端 Prometheus 新增 `im_ctrl_ack_delivery_duration_seconds{stage="enqueue|write"}`，按 Pod 记录从请求进入到 ACK 进入 WS 发送队列、以及完成 socket write 的时间。 +- Cluster gob 已注册 `[]map[string]any`，并用真实系统消息结构补了编解码回归测试。 + +R4 保持 R3 的 900 连接混合负载，仅将 `command_timeout` 从 `20s` 改为 `90s`。验证时必须同时满足：`publish_ctrl_unresolved=0`、消息实时与历史/MySQL 零丢失，并对比服务端 `enqueue`/`write` P95 与客户端 ctrl P95。如服务端两阶段仍为毫秒级，但 `publish_ctrl_over_20s>0`，则可将超过 20 秒的部分定位在 socket write 之后的网络/TCP 接收路径。 + +## 22. 精确 900 人全量单聊测试 + +Run ID：`s900m-p2p-r1-1001`。900 条 WebSocket 连接组成 450 对好友，A/B 各承担 450 个用户。每个用户以 1 秒间隔发送 10 条消息，共尝试发送 9000 条。两个 agent 均完整运行至实时静默判定、5 分钟沉淀、历史回查、MySQL 校验和 120 秒渐进关闭结束，没有提前终止。 + +| 指标 | 结果 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 发送尝试 / ACK | 9000 / 9000 | +| 明确拒绝 / 结果未知 | 0 / 0 | +| 对端实时投递 | 9000 / 9000 | +| 发送方回显 | 9000 / 9000 | +| WS 唯一回调总数 | 18000 / 18000 | +| 实时到达率 / 丢失率 | 100% / 0% | +| 重复 / 乱序 / 非预期消息 | 0 / 0 / 0 | +| 压测期非预期断连 | 0 | +| 历史回查 | 9000 / 9000 | +| MySQL 精确命中 | 9000 / 9000 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| ACK P95 / 最大值 | 854.385ms / 1324.17ms | +| 实时投递 P95 / 最大值 | 1000ms 桶内 / 1324.255ms | +| 服务端 Pod 重启 | 0 | + +报告的 `expected_deliveries=9000` 只统计对端投递，不包含发送方自身回显。A/B 各收到 4500 次对端投递和 4500 次本地回显，因此底层唯一 WS 数据回调合计为 18000 次，不是重复消息。 + +本轮单聊正确性和持久化门禁完整通过：9000 条发送全部收到 ACK，所有收发双方均实时收到应有消息，历史接口与 `im_008_messages` 都精确命中 9000 条。没有发生 ACK 超时、socket 错误、重复、乱序、实时丢失或持久化丢失。 + +压测端的回调处理没有构成瓶颈： + +| agent | 回调入队 / 完成 | 队列峰值 | 最大等待 | 结算时等待 / 处理中 | +|---|---:|---:|---:|---:| +| A | 9000 / 9000 | 7 | 16.317ms | 0 / 0 | +| B | 9000 / 9000 | 9 | 13.004ms | 0 / 0 | + +服务端运行状态稳定： + +- 三个 Pod 均为 Ready、0 重启，单 Pod 资源配置为 `requests=1 CPU/512Mi`、`limits=2 CPU/1Gi`。 +- 负载结束时 WS 连接分布为 527/373/544；压测连接渐进关闭后回落至 1/0/2，剩余为环境常驻连接，未再出现压测 agent FD 不释放。 +- 三个 Pod 的 WS send queue 结算时均为 0；Kafka 的消息持久化、群消息持久化、管理命令、配置事件、缓存失效、死信和工作台 consumer lag 均为 0。 +- 关闭过程中的少量 `ping closed` 累计指标来自客户端主动渐退，本轮 agent 记录的负载期非预期断连为 0。 + +按最新验收口径，ACK P95 和投递 P95 仅用于趋势对比，不作为当前公网单机容量测试的阻断条件。重新合并后本轮 `passed=true`：单聊消息正确性、实时到达、顺序性、连接稳定性和落库均通过。ACK P95 854.385ms 和投递 P95 1000ms 仍保留在报告中，但不影响本轮通过结论。 + +全量单聊已通过，无需为 500ms 延迟门禁单独重复 R2。下一步可进入更高风险的群聊或混合场景复测，仍以零丢失、零重复、零乱序、零断连和全量落库为通过条件。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-p2p-r1-1001/report.json` +- `loadtest/im-pressure/results/s900m-p2p-r1-1001/db-verification.json` +- `loadtest/im-pressure/results/s900m-p2p-r1-1001/metrics.jsonl` +- `loadtest/im-pressure/results/s900m-p2p-r1-1001/server-metrics/` diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 6306fc2..0a94f05 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -19,27 +19,32 @@ import ( var runIDPattern = regexp.MustCompile(`^[a-zA-Z0-9_-]{1,24}$`) type agentRun struct { - cfg *Config - machine string - runID string - scenario string - clients map[int]*WSClient - clientsMu sync.RWMutex - recorder *deliveryRecorder - callbackQueue *callbackQueueMetrics - ack *DurationHistogram - attempted atomic.Int64 - ready atomic.Int64 - publish403 atomic.Int64 - publish502 atomic.Int64 - publish503 atomic.Int64 - publishTimeout atomic.Int64 - publishSocket atomic.Int64 - publishOther atomic.Int64 + cfg *Config + machine string + runID string + scenario string + clients map[int]*WSClient + clientsMu sync.RWMutex + recorder *deliveryRecorder + callbackQueue *callbackQueueMetrics + ack *DurationHistogram + ctrlLatency *DurationHistogram + attempted atomic.Int64 + ready atomic.Int64 + publish403 atomic.Int64 + publish502 atomic.Int64 + publish503 atomic.Int64 + publishTimeout atomic.Int64 + publishSocket atomic.Int64 + publishOther atomic.Int64 + publishCtrlObserved atomic.Int64 + publishCtrlOver20 atomic.Int64 + publishLateCtrl atomic.Int64 - mu sync.Mutex - accepted []AcceptedMessage - errors []string + mu sync.Mutex + accepted []AcceptedMessage + errors []string + ctrlSamples []CtrlObservationSample } func validScenario(scenario string) bool { @@ -127,6 +132,7 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo coord := newCoordinatorClient(coordinatorURL) placeholderRunID := "pending" run := &agentRun{cfg: cfg, machine: machine, clients: make(map[int]*WSClient), ack: NewDurationHistogram(), + ctrlLatency: NewDurationHistogram(), callbackQueue: &callbackQueueMetrics{}} run.recorder = newDeliveryRecorder(placeholderRunID, machine, len(accounts)) stopMetrics, err := startAgentMetrics(metricsListen, run) @@ -211,6 +217,12 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo if err = waitForCallbackQueue(ctx, run.callbackQueue, 30*time.Second); err != nil { return err } + quietPeriod, _ := configDuration(cfg.DeliveryQuiet) + quietTimeout, _ := configDuration(cfg.DeliveryTimeout) + log.Printf("agent=%s waiting for %s of delivery quiescence", machine, quietPeriod) + if err = waitForDeliveryQuiescence(ctx, run.recorder, run.callbackQueue, quietPeriod, quietTimeout); err != nil { + return err + } observedLocal := run.recorder.unackedObservedMessages(ledger, start.Scenario) if err = coord.submitObserved(ctx, machine, observedLocal); err != nil { return err @@ -233,7 +245,10 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo run.mu.Lock() base.Errors = append(base.Errors, run.errors...) run.mu.Unlock() - summary := run.recorder.summarize(base, ledger) + // Include messages observed over WebSocket without a timely ctrl ACK. They + // are confirmed server-side publishes and must participate in realtime + // delivery accounting instead of being reported as unexpected traffic. + summary := run.recorder.summarize(base, verificationLedger) historySettle, _ := configDuration(cfg.HistorySettle) log.Printf("agent=%s waiting %s before history verification", machine, historySettle) if err = waitContext(ctx, historySettle); err != nil { @@ -256,7 +271,7 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo summary.HistoryUnackedFound = int64(history.unackedFound) persistenceMissing := history.missing summary.PersistenceLost = int64(len(persistenceMissing)) - summary.RealtimeLostStored = run.recorder.countRealtimeLostStored(ledger, storedMessageIDs) + summary.RealtimeLostStored = run.recorder.countRealtimeLostStored(verificationLedger, storedMessageIDs) if len(persistenceMissing) > 10000 { summary.PersistenceMissing = append([]string(nil), persistenceMissing[:10000]...) summary.Errors = append(summary.Errors, "persistence missing details truncated at 10000") @@ -264,6 +279,7 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo summary.PersistenceMissing = persistenceMissing } summary.FinishedAt = time.Now().UTC() + run.applyCtrlObservations(&summary) if err = writeAgentSummary(output, summary); err != nil { return err } @@ -293,6 +309,51 @@ func waitForCallbackQueue(ctx context.Context, queue *callbackQueueMetrics, time } return nil } +func waitForDeliveryQuiescence(ctx context.Context, recorder *deliveryRecorder, queue *callbackQueueMetrics, + quietPeriod, timeout time.Duration) error { + deadline := time.NewTimer(timeout) + defer deadline.Stop() + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + lastReceived := recorder.received.Load() + stableSince := time.Now() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-deadline.C: + return fmt.Errorf("deliveries did not become quiescent within %s: received=%d queued=%d processing=%d", + timeout, recorder.received.Load(), queue.current.Load(), queue.processing.Load()) + case now := <-ticker.C: + received := recorder.received.Load() + if received != lastReceived || !queue.isDrained() { + lastReceived = received + stableSince = now + continue + } + if now.Sub(stableSince) >= quietPeriod { + return nil + } + } + } +} + +func assignAgentEndpoints(cfg *Config, machine string, accounts []AccountRecord) { + endpoints := cfg.Endpoints[machine] + dialIPs := cfg.DialIPs[machine] + machineOffset := 0 + if machine == "b" { + machineOffset = 1 + } + for i := range accounts { + accounts[i].Endpoint = endpoints[(accounts[i].Index-1)%len(endpoints)] + if len(dialIPs) > 0 { + slot := (accounts[i].Index-1)*2 + machineOffset + accounts[i].DialIP = strings.TrimSpace(dialIPs[slot%len(dialIPs)]) + } + } +} + type historyQuery struct { client *WSClient topic string @@ -690,7 +751,8 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe func (r *agentRun) connectAccountWithRetry(ctx context.Context, account AccountRecord, groupTopic, apiKey, password string) (*WSClient, error) { var lastErr error for attempt := 1; attempt <= 6; attempt++ { - client, err := connectWSClient(ctx, r.cfg, account, apiKey, password, r.recorder.onData, r.recorder.onDisconnect, r.callbackQueue) + client, err := connectWSClient(ctx, r.cfg, account, apiKey, password, r.recorder.onData, r.recorder.onDisconnect, + r.observeCtrl, r.callbackQueue) if err == nil { if err = client.Subscribe(ctx, account.P2PTopic); err == nil { err = client.Subscribe(ctx, groupTopic) @@ -898,7 +960,53 @@ func (r *agentRun) explicitRejected() int64 { } func (r *agentRun) outcomeUnknown() int64 { - return r.publishTimeout.Load() + r.publishSocket.Load() + return r.unresolvedCtrl() + r.publishSocket.Load() +} + +func (r *agentRun) unresolvedCtrl() int64 { + unresolved := r.publishTimeout.Load() - r.publishLateCtrl.Load() + if unresolved < 0 { + return 0 + } + return unresolved +} + +func (r *agentRun) observeCtrl(account AccountRecord, observation ctrlObservation) { + if observation.Kind != "pub" { + return + } + r.publishCtrlObserved.Add(1) + r.ctrlLatency.Observe(observation.Latency) + if observation.Latency > 20*time.Second { + r.publishCtrlOver20.Add(1) + } + if observation.Late { + r.publishLateCtrl.Add(1) + } + if observation.Late || observation.Latency > 20*time.Second { + r.mu.Lock() + if len(r.ctrlSamples) < 1000 { + r.ctrlSamples = append(r.ctrlSamples, CtrlObservationSample{ + Machine: account.Machine, Index: account.Index, ID: observation.ID, Code: observation.Code, + SentAt: observation.SentAt, ReceivedAt: observation.ReceivedAt, + LatencyMS: float64(observation.Latency) / float64(time.Millisecond), Late: observation.Late, + }) + } + r.mu.Unlock() + } +} + +func (r *agentRun) applyCtrlObservations(summary *AgentSummary) { + summary.CtrlTrackingEnabled = true + summary.PublishCtrlObserved = r.publishCtrlObserved.Load() + summary.PublishCtrlOver20 = r.publishCtrlOver20.Load() + summary.PublishLateCtrl = r.publishLateCtrl.Load() + summary.PublishCtrlUnresolved = r.unresolvedCtrl() + summary.PublishOutcomeUnknown = r.outcomeUnknown() + summary.PublishCtrlLatency = r.ctrlLatency.Snapshot() + r.mu.Lock() + summary.CtrlObservationSamples = append([]CtrlObservationSample(nil), r.ctrlSamples...) + r.mu.Unlock() } func (r *agentRun) addAccepted(message AcceptedMessage) { diff --git a/loadtest/im-pressure/agent_metrics.go b/loadtest/im-pressure/agent_metrics.go index 9dd69de..42f33ec 100644 --- a/loadtest/im-pressure/agent_metrics.go +++ b/loadtest/im-pressure/agent_metrics.go @@ -46,6 +46,10 @@ func startAgentMetrics(address string, run *agentRun) (func(), error) { prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_502_total", Help: "Publish commands rejected with 502.", ConstLabels: labels}, func() float64 { return float64(run.publish502.Load()) }), prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_503_total", Help: "Publish commands rejected with 503.", ConstLabels: labels}, func() float64 { return float64(run.publish503.Load()) }), prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_ctrl_timeout_total", Help: "Publish commands with control-response timeout.", ConstLabels: labels}, func() float64 { return float64(run.publishTimeout.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_ctrl_observed_total", Help: "Publish control responses observed, including responses received after command timeout.", ConstLabels: labels}, func() float64 { return float64(run.publishCtrlObserved.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_ctrl_over_20s_total", Help: "Publish control responses taking longer than 20 seconds.", ConstLabels: labels}, func() float64 { return float64(run.publishCtrlOver20.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_ctrl_late_total", Help: "Publish control responses received after command timeout.", ConstLabels: labels}, func() float64 { return float64(run.publishLateCtrl.Load()) }), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_publish_ctrl_unresolved", Help: "Publish control responses still unresolved after accounting for late responses.", ConstLabels: labels}, func() float64 { return float64(run.unresolvedCtrl()) }), prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_socket_error_total", Help: "Publish commands with socket errors.", ConstLabels: labels}, func() float64 { return float64(run.publishSocket.Load()) }), prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_received_total", Help: "Unique matching data deliveries received.", ConstLabels: labels}, func() float64 { received, _, _, _ := run.recorder.metricCounts() diff --git a/loadtest/im-pressure/callback_queue_test.go b/loadtest/im-pressure/callback_queue_test.go index e7015d6..4129045 100644 --- a/loadtest/im-pressure/callback_queue_test.go +++ b/loadtest/im-pressure/callback_queue_test.go @@ -45,3 +45,20 @@ func TestWaitForCallbackQueueWaitsForProcessingCallback(t *testing.T) { t.Fatal("wait returned before the callback finished") } } + +func TestWaitForDeliveryQuiescenceResetsForLateDelivery(t *testing.T) { + recorder := newDeliveryRecorder("run", "a", 1) + queue := &callbackQueueMetrics{} + go func() { + time.Sleep(20 * time.Millisecond) + recorder.received.Add(1) + }() + + started := time.Now() + if err := waitForDeliveryQuiescence(context.Background(), recorder, queue, 150*time.Millisecond, time.Second); err != nil { + t.Fatal(err) + } + if elapsed := time.Since(started); elapsed < 250*time.Millisecond { + t.Fatalf("quiet period was not reset after late delivery: elapsed=%s", elapsed) + } +} diff --git a/loadtest/im-pressure/config.example.json b/loadtest/im-pressure/config.example.json index 4029512..74d372e 100644 --- a/loadtest/im-pressure/config.example.json +++ b/loadtest/im-pressure/config.example.json @@ -27,5 +27,7 @@ "p2p_messages_per_user": 10, "p2p_interval": "1s", "drain_p2p": "30s", - "drain_group": "60s" + "drain_group": "60s", + "delivery_quiet_period": "10s", + "delivery_quiet_timeout": "3m" } diff --git a/loadtest/im-pressure/config.single-machine.example.json b/loadtest/im-pressure/config.single-machine.example.json index 63aff5e..330ac37 100644 --- a/loadtest/im-pressure/config.single-machine.example.json +++ b/loadtest/im-pressure/config.single-machine.example.json @@ -10,10 +10,14 @@ "a": ["wss://im-api.example.com/v0/channels"], "b": ["wss://im-api.example.com/v0/channels"] }, + "dial_ips": { + "a": [], + "b": [] + }, "insecure_tls": false, "prepare_workers": 20, "connect_ramp": "30s", - "command_timeout": "20s", + "command_timeout": "90s", "ws_ping_interval": "20s", "preflight_workers": 32, "skip_p2p_preflight": false, @@ -27,5 +31,7 @@ "p2p_messages_per_user": 10, "p2p_interval": "1s", "drain_p2p": "30s", - "drain_group": "60s" + "drain_group": "60s", + "delivery_quiet_period": "10s", + "delivery_quiet_timeout": "3m" } diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index 17f5fba..f054290 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -10,45 +10,51 @@ import ( ) type CombinedReport struct { - RunID string `json:"run_id"` - Scenario string `json:"scenario"` - GeneratedAt time.Time `json:"generated_at"` - Agents []AgentSummary `json:"agents"` - ExpectedConnections int `json:"expected_connections"` - ReadyConnections int `json:"ready_connections"` - 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"` - DeliveryP95MS float64 `json:"delivery_p95_ms"` - DeliveryLocalP95MS float64 `json:"delivery_local_p95_ms"` - DeliveryCrossP95MS float64 `json:"delivery_cross_machine_p95_ms"` - CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` - ExpectedDeliveries int64 `json:"expected_deliveries"` - ReceivedDeliveries int64 `json:"received_deliveries"` - RealtimeLost int64 `json:"realtime_lost"` - RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` - ArrivalRate float64 `json:"arrival_rate"` - LossRate float64 `json:"loss_rate"` - Duplicates int64 `json:"duplicates"` - OutOfOrder int64 `json:"out_of_order"` - OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` - Disconnects int64 `json:"disconnects"` - DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` - DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` - UnexpectedMessages int64 `json:"unexpected_messages"` - HistoryChecked int64 `json:"history_checked"` - HistoryFound int64 `json:"history_found"` - HistoryUnackedChecked int64 `json:"history_unacked_observed_checked"` - HistoryUnackedFound int64 `json:"history_unacked_observed_found"` - PersistenceLost int64 `json:"persistence_lost"` - Missing []MissingDelivery `json:"missing,omitempty"` - Passed bool `json:"passed"` - FailureReasons []string `json:"failure_reasons,omitempty"` + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + GeneratedAt time.Time `json:"generated_at"` + Agents []AgentSummary `json:"agents"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + 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"` + PublishCtrlObserved int64 `json:"publish_ctrl_observed"` + PublishCtrlOver20 int64 `json:"publish_ctrl_over_20s"` + PublishLateCtrl int64 `json:"publish_ctrl_late"` + PublishCtrlUnresolved int64 `json:"publish_ctrl_unresolved"` + PublishCtrlP95MS float64 `json:"publish_ctrl_p95_ms"` + CtrlObservationSamples []CtrlObservationSample `json:"ctrl_observation_samples,omitempty"` + PublishSuccessRate float64 `json:"publish_success_rate"` + AckP95MS float64 `json:"ack_p95_ms"` + DeliveryP95MS float64 `json:"delivery_p95_ms"` + DeliveryLocalP95MS float64 `json:"delivery_local_p95_ms"` + DeliveryCrossP95MS float64 `json:"delivery_cross_machine_p95_ms"` + CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` + ExpectedDeliveries int64 `json:"expected_deliveries"` + ReceivedDeliveries int64 `json:"received_deliveries"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + ArrivalRate float64 `json:"arrival_rate"` + LossRate float64 `json:"loss_rate"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` + Disconnects int64 `json:"disconnects"` + DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` + DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` + UnexpectedMessages int64 `json:"unexpected_messages"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + HistoryUnackedChecked int64 `json:"history_unacked_observed_checked"` + HistoryUnackedFound int64 `json:"history_unacked_observed_found"` + PersistenceLost int64 `json:"persistence_lost"` + Missing []MissingDelivery `json:"missing,omitempty"` + Passed bool `json:"passed"` + FailureReasons []string `json:"failure_reasons,omitempty"` } func runMerge(aPath, bPath, output string) error { @@ -78,12 +84,24 @@ func runMerge(aPath, bPath, output string) error { report.PublishErrors[category] += count } unknown := summary.PublishOutcomeUnknown - if unknown == 0 { + if summary.CtrlTrackingEnabled { + unknown = summary.PublishCtrlUnresolved + summary.PublishErrors["socket"] + } else if unknown == 0 { unknown = summary.PublishErrors["ctrl_timeout"] + summary.PublishErrors["socket"] } explicit := summary.PublishErrors["403"] + summary.PublishErrors["502"] + summary.PublishErrors["503"] + summary.PublishErrors["other"] report.PublishOutcomeUnknown += unknown + report.PublishCtrlObserved += summary.PublishCtrlObserved + report.PublishCtrlOver20 += summary.PublishCtrlOver20 + report.PublishLateCtrl += summary.PublishLateCtrl + report.PublishCtrlUnresolved += summary.PublishCtrlUnresolved + if remaining := 1000 - len(report.CtrlObservationSamples); remaining > 0 { + if remaining > len(summary.CtrlObservationSamples) { + remaining = len(summary.CtrlObservationSamples) + } + report.CtrlObservationSamples = append(report.CtrlObservationSamples, summary.CtrlObservationSamples[:remaining]...) + } report.PublishExplicitReject += explicit report.PublishRejected += explicit report.ExpectedDeliveries += summary.ExpectedPeer + summary.ExpectedGroup @@ -119,6 +137,7 @@ func runMerge(aPath, bPath, output string) error { report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted) } report.AckP95MS = mergedAckP95(report.Agents) + report.PublishCtrlP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.PublishCtrlLatency }) report.DeliveryLocalP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatencyLocal }) report.DeliveryCrossP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatencyCross }) // delivery_p95_ms remains the compatibility field, but now uses only same-clock @@ -147,13 +166,6 @@ func runMerge(aPath, bPath, output string) error { 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)) - } - if report.DeliveryP95MS > 1000 { - 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 || report.UnexpectedMessages > 0 { report.FailureReasons = append(report.FailureReasons, "message correctness gate failed") } diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index e961208..f3ec12b 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -13,14 +13,17 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ReadyConnections: 1000, ExpectedConnections: 1000, 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, Disconnects: 1, + CtrlTrackingEnabled: true, PublishCtrlObserved: 9, PublishCtrlOver20: 2, PublishLateCtrl: 2, + PublishCtrlUnresolved: 1, + PublishErrors: map[string]int64{"ctrl_timeout": 3, "socket": 2, "503": 1}, + RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10, Disconnects: 1, DisconnectReasons: map[string]int64{"timeout": 1}, DisconnectSamples: []DisconnectSample{{ReceiverMachine: "a", ReceiverIndex: 3, Category: "timeout"}}} b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000, ExpectedConnections: 1000, PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1, OutOfOrder: 1, + CtrlTrackingEnabled: true, PublishCtrlObserved: 10, Disconnects: 2, DisconnectReasons: map[string]int64{"timeout": 1, "connection_reset": 1}, OrderingViolations: []OrderingViolation{{ReceiverMachine: "b", ReceiverIndex: 7, PreviousSeq: 12, CurrentSeq: 11}}} aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json") @@ -43,10 +46,14 @@ 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 || + if report.UnexpectedMessages != 42 || report.PublishOutcomeUnknown != 3 || report.PublishExplicitReject != 1 || report.PublishRejected != 1 { t.Fatalf("ambiguous publish outcomes were not merged: %+v", report) } + if report.PublishCtrlObserved != 19 || report.PublishCtrlOver20 != 2 || report.PublishLateCtrl != 2 || + report.PublishCtrlUnresolved != 1 { + t.Fatalf("ctrl observations 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) } @@ -93,7 +100,7 @@ func TestMergedAckP95FallsBackWhenRawSamplesAreIncomplete(t *testing.T) { } } -func TestRunMergeUsesSameClockDeliveryLatencyForGate(t *testing.T) { +func TestRunMergeReportsSameClockDeliveryLatency(t *testing.T) { dir := t.TempDir() local := HistogramSnapshot{Count: 10, MaxMS: 200, Buckets: map[string]int64{"200ms": 10}} cross := HistogramSnapshot{Count: 10, MaxMS: 5000, Buckets: map[string]int64{"5s": 10}} @@ -120,7 +127,7 @@ func TestRunMergeUsesSameClockDeliveryLatencyForGate(t *testing.T) { t.Fatalf("unexpected delivery percentiles: %+v", report) } if report.CrossClockSkewSamples != 6 || !report.Passed { - t.Fatalf("cross-machine clock skew must not fail the same-clock latency gate: %+v", report) + t.Fatalf("cross-machine clock skew must not fail correctness acceptance: %+v", report) } } diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index a354870..93af4d8 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "net" "os" "sort" "strings" @@ -21,6 +22,7 @@ type Config struct { PasswordEnv string `json:"password_env"` UsersPerMachine int `json:"users_per_machine"` Endpoints map[string][]string `json:"endpoints"` + DialIPs map[string][]string `json:"dial_ips,omitempty"` InsecureTLS bool `json:"insecure_tls"` PrepareWorkers int `json:"prepare_workers"` ConnectRamp string `json:"connect_ramp"` @@ -30,6 +32,8 @@ type Config struct { P2PInterval string `json:"p2p_interval"` DrainP2P string `json:"drain_p2p"` DrainGroup string `json:"drain_group"` + DeliveryQuiet string `json:"delivery_quiet_period"` + DeliveryTimeout string `json:"delivery_quiet_timeout"` PhonePrefix string `json:"registration_phone_prefix"` VerifyCodeEnv string `json:"verification_code_env"` HistoryWorkers int `json:"history_workers"` @@ -85,6 +89,18 @@ func loadConfig(path string) (*Config, error) { if cfg.DrainGroup == "" { cfg.DrainGroup = "60s" } + if cfg.DeliveryQuiet == "" { + cfg.DeliveryQuiet = "10s" + } + if _, err = configDuration(cfg.DeliveryQuiet); err != nil { + return nil, fmt.Errorf("delivery_quiet_period: %w", err) + } + if cfg.DeliveryTimeout == "" { + cfg.DeliveryTimeout = "3m" + } + if _, err = configDuration(cfg.DeliveryTimeout); err != nil { + return nil, fmt.Errorf("delivery_quiet_timeout: %w", err) + } if cfg.HistoryWorkers <= 0 { cfg.HistoryWorkers = 12 } @@ -140,6 +156,11 @@ func loadConfig(path string) (*Config, error) { if len(cfg.Endpoints[machine]) == 0 { return nil, fmt.Errorf("at least one endpoint is required for machine %s", machine) } + for _, ip := range cfg.DialIPs[machine] { + if net.ParseIP(strings.TrimSpace(ip)) == nil { + return nil, fmt.Errorf("invalid dial IP %q for machine %s", ip, machine) + } + } } if cfg.PhonePrefix != "" { if len(cfg.PhonePrefix) != 7 || strings.Trim(cfg.PhonePrefix, "0123456789") != "" { @@ -214,6 +235,7 @@ type AccountRecord struct { TokenExpires time.Time `json:"token_expires,omitempty"` P2PTopic string `json:"p2p_topic,omitempty"` Endpoint string `json:"endpoint,omitempty"` + DialIP string `json:"-"` } func newManifest(cfg *Config) *Manifest { @@ -385,49 +407,67 @@ type AcceptedMessage struct { } type AgentSummary struct { - RunID string `json:"run_id"` - Scenario string `json:"scenario"` - Machine string `json:"machine"` - StartedAt time.Time `json:"started_at"` - LoadEndedAt time.Time `json:"load_ended_at"` - FinishedAt time.Time `json:"finished_at"` - ExpectedConnections int `json:"expected_connections"` - ReadyConnections int `json:"ready_connections"` - PublishAttempted int64 `json:"publish_attempted"` - PublishAccepted int64 `json:"publish_accepted"` - PublishRejected int64 `json:"publish_rejected"` - PublishOutcomeUnknown int64 `json:"publish_outcome_unknown"` - PublishErrors map[string]int64 `json:"publish_errors,omitempty"` - ExpectedPeer int64 `json:"expected_peer"` - ReceivedPeer int64 `json:"received_peer"` - ExpectedEcho int64 `json:"expected_echo"` - ReceivedEcho int64 `json:"received_echo"` - ExpectedGroup int64 `json:"expected_group"` - ReceivedGroup int64 `json:"received_group"` - RealtimeLost int64 `json:"realtime_lost"` - RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` - Duplicates int64 `json:"duplicates"` - OutOfOrder int64 `json:"out_of_order"` - OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` - Disconnects int64 `json:"disconnects"` - DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` - DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` - UnexpectedMessages int64 `json:"unexpected_messages"` - HistoryChecked int64 `json:"history_checked"` - HistoryFound int64 `json:"history_found"` - HistoryUnackedChecked int64 `json:"history_unacked_observed_checked"` - HistoryUnackedFound int64 `json:"history_unacked_observed_found"` - PersistenceLost int64 `json:"persistence_lost"` - AckLatency HistogramSnapshot `json:"ack_latency"` - DeliveryLatency HistogramSnapshot `json:"delivery_latency"` - DeliveryLatencyLocal HistogramSnapshot `json:"delivery_latency_local"` - DeliveryLatencyCross HistogramSnapshot `json:"delivery_latency_cross_machine"` - CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` - CallbackQueue CallbackQueueSnapshot `json:"callback_queue"` - Errors []string `json:"errors,omitempty"` - Missing []MissingDelivery `json:"missing,omitempty"` - PersistenceMissing []string `json:"persistence_missing,omitempty"` - AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + Machine string `json:"machine"` + StartedAt time.Time `json:"started_at"` + LoadEndedAt time.Time `json:"load_ended_at"` + FinishedAt time.Time `json:"finished_at"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + PublishOutcomeUnknown int64 `json:"publish_outcome_unknown"` + PublishErrors map[string]int64 `json:"publish_errors,omitempty"` + CtrlTrackingEnabled bool `json:"ctrl_tracking_enabled"` + PublishCtrlObserved int64 `json:"publish_ctrl_observed"` + PublishCtrlOver20 int64 `json:"publish_ctrl_over_20s"` + PublishLateCtrl int64 `json:"publish_ctrl_late"` + PublishCtrlUnresolved int64 `json:"publish_ctrl_unresolved"` + PublishCtrlLatency HistogramSnapshot `json:"publish_ctrl_latency"` + CtrlObservationSamples []CtrlObservationSample `json:"ctrl_observation_samples,omitempty"` + ExpectedPeer int64 `json:"expected_peer"` + ReceivedPeer int64 `json:"received_peer"` + ExpectedEcho int64 `json:"expected_echo"` + ReceivedEcho int64 `json:"received_echo"` + ExpectedGroup int64 `json:"expected_group"` + ReceivedGroup int64 `json:"received_group"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` + Disconnects int64 `json:"disconnects"` + DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` + DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` + UnexpectedMessages int64 `json:"unexpected_messages"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + HistoryUnackedChecked int64 `json:"history_unacked_observed_checked"` + HistoryUnackedFound int64 `json:"history_unacked_observed_found"` + PersistenceLost int64 `json:"persistence_lost"` + AckLatency HistogramSnapshot `json:"ack_latency"` + DeliveryLatency HistogramSnapshot `json:"delivery_latency"` + DeliveryLatencyLocal HistogramSnapshot `json:"delivery_latency_local"` + DeliveryLatencyCross HistogramSnapshot `json:"delivery_latency_cross_machine"` + CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` + CallbackQueue CallbackQueueSnapshot `json:"callback_queue"` + Errors []string `json:"errors,omitempty"` + Missing []MissingDelivery `json:"missing,omitempty"` + PersistenceMissing []string `json:"persistence_missing,omitempty"` + AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` +} + +type CtrlObservationSample struct { + Machine string `json:"machine"` + Index int `json:"index"` + ID string `json:"id"` + Code int `json:"code"` + SentAt time.Time `json:"sent_at"` + ReceivedAt time.Time `json:"received_at"` + LatencyMS float64 `json:"latency_ms"` + Late bool `json:"late"` } type CallbackQueueSnapshot struct { diff --git a/loadtest/im-pressure/model_test.go b/loadtest/im-pressure/model_test.go index 62532ca..7f6f93b 100644 --- a/loadtest/im-pressure/model_test.go +++ b/loadtest/im-pressure/model_test.go @@ -3,6 +3,7 @@ package main import ( "os" "path/filepath" + "reflect" "testing" ) @@ -24,6 +25,33 @@ func TestLoadConfigAppliesBoundedHistoryDefaults(t *testing.T) { } } +func TestAssignAgentEndpointsDistributesDialIPsAcrossMachines(t *testing.T) { + cfg := &Config{ + Endpoints: map[string][]string{"a": {"wss://api.example.test/ws"}, "b": {"wss://api.example.test/ws"}}, + DialIPs: map[string][]string{"a": {"192.0.2.1", "192.0.2.2", "192.0.2.3"}, "b": {"192.0.2.1", "192.0.2.2", "192.0.2.3"}}, + } + a := []AccountRecord{{Index: 1}, {Index: 2}, {Index: 3}} + b := []AccountRecord{{Index: 1}, {Index: 2}, {Index: 3}} + assignAgentEndpoints(cfg, "a", a) + assignAgentEndpoints(cfg, "b", b) + got := []string{a[0].DialIP, b[0].DialIP, a[1].DialIP, b[1].DialIP, a[2].DialIP, b[2].DialIP} + want := []string{"192.0.2.1", "192.0.2.2", "192.0.2.3", "192.0.2.1", "192.0.2.2", "192.0.2.3"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("unexpected dial IP distribution: got=%v want=%v", got, want) + } +} + +func TestLoadConfigRejectsInvalidDialIP(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.json") + raw := []byte(`{"http_base":"https://example.test","endpoints":{"a":["wss://a"],"b":["wss://b"]},"dial_ips":{"a":["not-an-ip"]}}`) + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatal(err) + } + if _, err := loadConfig(path); err == nil { + t.Fatal("invalid dial IP was accepted") + } +} + func TestSingleMachineExampleUsesSixtyPercentCapacity(t *testing.T) { cfg, err := loadConfig("config.single-machine.example.json") if err != nil { diff --git a/loadtest/im-pressure/wsclient.go b/loadtest/im-pressure/wsclient.go index 728d536..819ec76 100644 --- a/loadtest/im-pressure/wsclient.go +++ b/loadtest/im-pressure/wsclient.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "net" "net/http" "net/url" "strconv" @@ -40,12 +41,36 @@ type queuedDataPacket struct { queuedAt time.Time } +type ctrlWait struct { + ch chan ctrlPacket + kind string + sentAt time.Time +} + +type timedOutCtrl struct { + kind string + sentAt time.Time +} + +type ctrlObservation struct { + ID string + Kind string + Code int + SentAt time.Time + ReceivedAt time.Time + Latency time.Duration + Late bool +} + type ctrlError struct { Code int Text string } -const dataCallbackQueueSize = 512 +const ( + dataCallbackQueueSize = 512 + lateCtrlRetention = 10 * time.Minute +) func (e *ctrlError) Error() string { return fmt.Sprintf("ctrl code=%d text=%s", e.Code, e.Text) @@ -57,19 +82,22 @@ type WSClient struct { commandTimeout time.Duration onData func(AccountRecord, dataPacket) onDisconnect func(AccountRecord, error) + onCtrlObserved func(AccountRecord, ctrlObservation) callbackQueue *callbackQueueMetrics - sendMu sync.Mutex - mu sync.Mutex - seq uint64 - waits map[string]chan ctrlPacket - done chan struct{} - data chan queuedDataPacket - closed atomic.Bool + sendMu sync.Mutex + mu sync.Mutex + seq uint64 + waits map[string]ctrlWait + timedOutCtrls map[string]timedOutCtrl + done chan struct{} + data chan queuedDataPacket + closed atomic.Bool } func connectWSClient(ctx context.Context, cfg *Config, account AccountRecord, apiKey, password string, - onData func(AccountRecord, dataPacket), onDisconnect func(AccountRecord, error), callbackQueue *callbackQueueMetrics) (*WSClient, error) { + onData func(AccountRecord, dataPacket), onDisconnect func(AccountRecord, error), + onCtrlObserved func(AccountRecord, ctrlObservation), callbackQueue *callbackQueueMetrics) (*WSClient, error) { timeout, err := configDuration(cfg.CommandTimeout) if err != nil { return nil, err @@ -79,6 +107,17 @@ func connectWSClient(ctx context.Context, cfg *Config, account AccountRecord, ap return nil, err } dialer := *websocket.DefaultDialer + if account.DialIP != "" { + targetHost := endpointHostname(endpoint) + var netDialer net.Dialer + dialer.NetDialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + host, port, splitErr := net.SplitHostPort(address) + if splitErr == nil && strings.EqualFold(host, targetHost) { + address = net.JoinHostPort(account.DialIP, port) + } + return netDialer.DialContext(ctx, network, address) + } + } if cfg.InsecureTLS { dialer.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} // #nosec G402: explicit load-test option. } @@ -91,8 +130,9 @@ func connectWSClient(ctx context.Context, cfg *Config, account AccountRecord, ap } client := &WSClient{ account: account, conn: conn, commandTimeout: timeout, onData: onData, onDisconnect: onDisconnect, - callbackQueue: callbackQueue, - waits: make(map[string]chan ctrlPacket), done: make(chan struct{}), + onCtrlObserved: onCtrlObserved, + callbackQueue: callbackQueue, + waits: make(map[string]ctrlWait), timedOutCtrls: make(map[string]timedOutCtrl), done: make(chan struct{}), } client.startDataProcessor() go client.readLoop() @@ -136,6 +176,14 @@ func connectWSClient(ctx context.Context, cfg *Config, account AccountRecord, ap return client, nil } +func endpointHostname(raw string) string { + u, err := url.Parse(raw) + if err != nil { + return "" + } + return u.Hostname() +} + func connectForPrepare(ctx context.Context, cfg *Config, account AccountRecord, api *apiClient, password, verificationCode string) (*WSClient, bool, error) { timeout, err := configDuration(cfg.CommandTimeout) if err != nil { @@ -153,7 +201,8 @@ func connectForPrepare(ctx context.Context, cfg *Config, account AccountRecord, if err != nil { return nil, false, err } - client := &WSClient{account: account, conn: conn, commandTimeout: timeout, waits: make(map[string]chan ctrlPacket), done: make(chan struct{})} + client := &WSClient{account: account, conn: conn, commandTimeout: timeout, + waits: make(map[string]ctrlWait), timedOutCtrls: make(map[string]timedOutCtrl), done: make(chan struct{})} go client.readLoop() if _, err = client.command(ctx, "hi", map[string]any{"hi": map[string]any{ "id": client.nextID("hi"), "ver": "0.25", "ua": "im-pressure-prepare/1.0", "tenant": cfg.Tenant, @@ -296,14 +345,15 @@ func ctrlInt64(raw json.RawMessage) (int64, error) { return strconv.ParseInt(text, 10, 64) } -func (c *WSClient) command(ctx context.Context, _ string, message map[string]any) (ctrlPacket, error) { +func (c *WSClient) command(ctx context.Context, kind string, message map[string]any) (ctrlPacket, error) { id := nestedMessageID(message) if id == "" { return ctrlPacket{}, errors.New("command id is required") } wait := make(chan ctrlPacket, 1) + sentAt := time.Now() c.mu.Lock() - c.waits[id] = wait + c.waits[id] = ctrlWait{ch: wait, kind: kind, sentAt: sentAt} c.mu.Unlock() c.sendMu.Lock() err := c.conn.WriteJSON(message) @@ -321,7 +371,33 @@ func (c *WSClient) command(ctx context.Context, _ string, message map[string]any } return ctrl, nil case <-timer.C: - c.removeWait(id) + c.mu.Lock() + entry, pending := c.waits[id] + if pending { + delete(c.waits, id) + if entry.kind == "pub" { + c.timedOutCtrls[id] = timedOutCtrl{kind: entry.kind, sentAt: entry.sentAt} + } + } + c.mu.Unlock() + if !pending { + // The read loop won the timer race and placed the response in the + // buffered channel before removing the waiter. + ctrl := <-wait + if ctrl.Code < 200 || ctrl.Code >= 300 { + return ctrl, &ctrlError{Code: ctrl.Code, Text: ctrl.Text} + } + return ctrl, nil + } + if entry.kind == "pub" { + time.AfterFunc(lateCtrlRetention, func() { + c.mu.Lock() + if current, ok := c.timedOutCtrls[id]; ok && current.sentAt.Equal(sentAt) { + delete(c.timedOutCtrls, id) + } + c.mu.Unlock() + }) + } return ctrlPacket{}, fmt.Errorf("ctrl timeout id=%s", id) case <-ctx.Done(): c.removeWait(id) @@ -374,11 +450,23 @@ func (c *WSClient) readLoop() { var ctrl ctrlPacket if json.Unmarshal(raw, &ctrl) == nil && ctrl.ID != "" { c.mu.Lock() - wait := c.waits[ctrl.ID] - delete(c.waits, ctrl.ID) + wait, pending := c.waits[ctrl.ID] + late, timedOut := c.timedOutCtrls[ctrl.ID] + receivedAt := time.Now() + if pending { + wait.ch <- ctrl + delete(c.waits, ctrl.ID) + } else if timedOut { + delete(c.timedOutCtrls, ctrl.ID) + } c.mu.Unlock() - if wait != nil { - wait <- ctrl + if c.onCtrlObserved != nil && (pending && wait.kind == "pub" || timedOut && late.kind == "pub") { + kind, sentAt, isLate := wait.kind, wait.sentAt, false + if timedOut { + kind, sentAt, isLate = late.kind, late.sentAt, true + } + c.onCtrlObserved(c.account, ctrlObservation{ID: ctrl.ID, Kind: kind, Code: ctrl.Code, + SentAt: sentAt, ReceivedAt: receivedAt, Latency: receivedAt.Sub(sentAt), Late: isLate}) } } continue diff --git a/loadtest/im-pressure/wsclient_test.go b/loadtest/im-pressure/wsclient_test.go index 5082ac1..15906ad 100644 --- a/loadtest/im-pressure/wsclient_test.go +++ b/loadtest/im-pressure/wsclient_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "net/http" "net/http/httptest" "strings" @@ -10,6 +11,49 @@ import ( "github.com/gorilla/websocket" ) +func TestWSClientObservesCtrlAfterCommandTimeout(t *testing.T) { + 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() + var command map[string]any + if err = conn.ReadJSON(&command); err != nil { + return + } + time.Sleep(40 * time.Millisecond) + _ = conn.WriteJSON(map[string]any{"ctrl": map[string]any{"id": "pub-a-0001", "code": 202, "text": "accepted"}}) + })) + defer server.Close() + + conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil) + if err != nil { + t.Fatal(err) + } + observed := make(chan ctrlObservation, 1) + client := &WSClient{ + account: AccountRecord{Machine: "a", Index: 1}, conn: conn, commandTimeout: 10 * time.Millisecond, + waits: make(map[string]ctrlWait), timedOutCtrls: make(map[string]timedOutCtrl), done: make(chan struct{}), + onCtrlObserved: func(_ AccountRecord, observation ctrlObservation) { observed <- observation }, + } + go client.readLoop() + defer client.Close() + + _, err = client.command(context.Background(), "pub", map[string]any{"pub": map[string]any{"id": "pub-a-0001"}}) + if err == nil || !strings.Contains(err.Error(), "ctrl timeout") { + t.Fatalf("command error=%v, want ctrl timeout", err) + } + select { + case observation := <-observed: + if !observation.Late || observation.Kind != "pub" || observation.Code != 202 || observation.Latency < 10*time.Millisecond { + t.Fatalf("unexpected observation: %+v", observation) + } + case <-time.After(time.Second): + t.Fatal("late ctrl response was not observed") + } +} + func TestWSClientKeepaliveSendsPing(t *testing.T) { ping := make(chan struct{}, 1) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -37,7 +81,7 @@ func TestWSClientKeepaliveSendsPing(t *testing.T) { if err != nil { t.Fatal(err) } - client := &WSClient{conn: conn, waits: make(map[string]chan ctrlPacket), done: make(chan struct{})} + client := &WSClient{conn: conn, waits: make(map[string]ctrlWait), timedOutCtrls: make(map[string]timedOutCtrl), done: make(chan struct{})} go client.readLoop() client.startKeepalive(10 * time.Millisecond) defer client.Close() @@ -70,7 +114,7 @@ func TestWSClientReadLoopClosesConnectionAfterDecodeError(t *testing.T) { if err != nil { t.Fatal(err) } - client := &WSClient{conn: conn, waits: make(map[string]chan ctrlPacket), done: make(chan struct{})} + client := &WSClient{conn: conn, waits: make(map[string]ctrlWait), timedOutCtrls: make(map[string]timedOutCtrl), done: make(chan struct{})} go client.readLoop() select { @@ -123,7 +167,7 @@ func TestWSClientDataCallbackDoesNotBlockPong(t *testing.T) { } blocked := make(chan struct{}) client := &WSClient{ - conn: conn, waits: make(map[string]chan ctrlPacket), done: make(chan struct{}), + conn: conn, waits: make(map[string]ctrlWait), timedOutCtrls: make(map[string]timedOutCtrl), done: make(chan struct{}), onData: func(AccountRecord, dataPacket) { <-blocked }, } client.startDataProcessor() diff --git a/server/cluster.go b/server/cluster.go index 1f5541e..56fffd2 100644 --- a/server/cluster.go +++ b/server/cluster.go @@ -1215,6 +1215,7 @@ func clusterInit(configString json.RawMessage, self *string) int { func registerClusterGobTypes() { gob.Register([]any{}) + gob.Register([]map[string]any{}) gob.Register(map[string]any{}) gob.Register(map[string]int{}) gob.Register(map[string]string{}) diff --git a/server/cluster_gob_test.go b/server/cluster_gob_test.go index 9b864b3..4b0f3bb 100644 --- a/server/cluster_gob_test.go +++ b/server/cluster_gob_test.go @@ -28,6 +28,7 @@ func TestClusterGobRegistersSystemMessagePayloadTypes(t *testing.T) { Content: map[string]any{ "text": "成员加入了群聊", "template": "group_member_joined", + "items": []map[string]any{{"type": "member", "id": "usr123"}}, "actors": []imSystemMessageActor{{ UserID: "usr123", Click: map[string]any{"type": "user_profile", "target_id": "usr123"}, diff --git a/server/hdl_websock.go b/server/hdl_websock.go index 0a2bba3..3f878dc 100644 --- a/server/hdl_websock.go +++ b/server/hdl_websock.go @@ -148,12 +148,18 @@ func (sess *Session) writeLoop() { if !sess.sendMessage(w) { return } + if isAcceptedCtrl(msg) { + immetrics.ObserveCtrlACKDelivery("write", msg.Timestamp) + } } case *ServerComMessage: // single unserialized message w := sess.serializeAndUpdateStats(v) if !sess.sendMessage(w) { return } + if isAcceptedCtrl(v) { + immetrics.ObserveCtrlACKDelivery("write", v.Timestamp) + } default: // serialized message if !sess.sendMessage(v) { return diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index b5d8426..d21dd55 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -105,6 +105,11 @@ var ( Help: "Time a WebSocket outbound channel entry waited before dequeue or purge.", 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, 120}, }, []string{"pod", "result"}) + IMCtrlACKDeliveryDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_ctrl_ack_delivery_duration_seconds", + Help: "Accepted control-response age from request receipt to WebSocket enqueue or completed socket write.", + 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, 20, 30, 60, 120}, + }, []string{"pod", "stage"}) IMClusterLivePeers = prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "im_cluster_live_peers", Help: "Current connected cluster peers.", @@ -230,7 +235,7 @@ func init() { Registry.MustRegister(WSConnectionsActive, IMWSConnectionsActive, IMWSWriteTotal, IMWSWriteDuration, IMWSWriteBytesTotal, IMWSDisconnectTotal, IMWSHeartbeatTotal, IMWSHeartbeatSeconds, IMWSSendQueueCurrent, IMWSSendQueueCapacity, - IMWSSendQueueEnqueueTotal, IMWSSendQueueDepth, IMWSSendQueueWaitDuration, + IMWSSendQueueEnqueueTotal, IMWSSendQueueDepth, IMWSSendQueueWaitDuration, IMCtrlACKDeliveryDuration, IMClusterLivePeers, IMClusterExpectedPeers, HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration, DBInvalidConnectionRetryTotal, RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength, @@ -385,6 +390,13 @@ func ObserveWSSendQueueDequeue(queuedAt time.Time, purged bool) { } } +func ObserveCtrlACKDelivery(stage string, requestAt time.Time) { + if requestAt.IsZero() { + return + } + IMCtrlACKDeliveryDuration.WithLabelValues(metricPodName(), normalizeLabel(stage)).Observe(time.Since(requestAt).Seconds()) +} + func normalizeLabel(value string) string { if value == "" { return "unknown" diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index 6cc86de..2996a80 100644 --- a/server/monitoring_http_test.go +++ b/server/monitoring_http_test.go @@ -96,6 +96,7 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { immetrics.IncWSDisconnect("close_normal") immetrics.ObserveWSSendQueueEnqueue(1, 160, true) immetrics.ObserveWSSendQueueDequeue(time.Now(), false) + immetrics.ObserveCtrlACKDelivery("write", time.Now()) immetrics.ObserveTopicQueueWait("group", time.Now()) immetrics.ObservePushDispatchEnqueue(1, true) immetrics.ObservePushDispatch(time.Now(), 1, time.Now(), nil) @@ -113,6 +114,7 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { "im_ws_write_total", "im_ws_write_duration_seconds", "im_ws_write_bytes_total", "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_ctrl_ack_delivery_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_current", "im_topic_queue_wait_duration_seconds", "im_push_dispatch_queue_enqueue_total", diff --git a/server/session.go b/server/session.go index 79c4ff9..4843a90 100644 --- a/server/session.go +++ b/server/session.go @@ -447,12 +447,19 @@ func (s *Session) queueOut(msg *ServerComMessage) bool { logs.Err.Println("s.queueOut: session's send queue full", s.sid) return false } + if isAcceptedCtrl(msg) && s.proto == WEBSOCK { + immetrics.ObserveCtrlACKDelivery("enqueue", msg.Timestamp) + } if s.isMultiplex() { s.scheduleClusterWriteLoop() } return true } +func isAcceptedCtrl(msg *ServerComMessage) bool { + return msg != nil && msg.Ctrl != nil && msg.Ctrl.Code == http.StatusAccepted && msg.Ctrl.Id != "" +} + // queueOutBytes attempts to send a ServerComMessage already serialized to []byte. // If the send buffer is full, it fails. func (s *Session) queueOutBytes(data []byte) bool {