diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index e412522..5975509 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -807,3 +807,143 @@ R6 和 R7 均在最后一批连接关闭后记录 1 次 `s.queueOut: session's s - `loadtest/im-pressure/results/s900m-mixed-r6-1001/` - `loadtest/im-pressure/results/s900m-mixed-r7-1001/` + +## 26. 900 人混合 R8 关闭竞态补丁复验 + +Run ID：`s900m-mixed-r8-1002`。本轮使用与 R7 完全相同的 900 条 WebSocket、135 对单聊用户和 224 个群发送者负载，重点验证关闭竞态补丁发布后，业务负载期和 120 秒渐进关闭全过程不再出现 `send queue full`。 + +| 指标 | R8 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 计划发送 / 实际尝试 | 1798 / 1798 | +| 最终接受 / socket 结果未知 | 1796 / 2 | +| 明确拒绝 / ACK 超过 20 秒 | 0 / 0 | +| 精确 ACK P95 / 最大值 | 1041.578ms / 2055.915ms | +| 单聊对端实时投递 | 1350 / 1350 | +| 已接受群消息应投 / 实投 | 401400 / 400954 | +| 已接受消息实时总投递 | 402750 / 402304 | +| 实时缺口 | 446，全部可持久化回查 | +| 重复 / 乱序 | 0 / 0 | +| 压测期非预期断连 | 1，公网读超时后自动重连 | +| 历史回查 | 1796 / 1796 | +| MySQL 精确命中 | 1796 / 1796 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| 负载期 / 关闭期 queue full | 0 / 0 | +| Pod Ready / 重启 | 3 / 3，0 重启 | + +### 26.1 严格业务门禁结论 + +本轮 `report.json` 为 `passed=false`。失败由 A 机第 231 个连接的一次公网读超时引起： + +```text +read tcp 192.168.1.135:57717->216.236.31.191:443: read: operation timed out +``` + +断连发生于 `2026-10-02T06:31:53.391465Z`，客户端约 1.5 秒后报告连接恢复。该用户的两条群发送 `s900m-mixed-r8-1002-g-a-0231-01/02` 分别得到 `websocket closed` 和 `websocket connection unavailable`，未进入服务端，MySQL 中也没有对应额外行。因此实际被服务端接受的业务消息为 1796 条，不能把这两条计为服务端持久化丢失。 + +同一连接在群消息阶段漏收 446 条已接受群消息，形成 446 次实时投递缺口；这些消息全部可通过历史回查获得。其余连接的群投递、全部 1350 次单聊对端投递、重复和顺序检查均正确。按严格验收标准，本轮仍因 `disconnects=1`、`publish_outcome_unknown=2` 和实时缺口判定失败，不能替代 R7 的业务全绿结论。 + +服务端三个 Pod 在超时窗口均保持 Ready、零重启、Kafka lag 为 0，也没有 Topic、Cluster、WS 队列满、RPC error、gob 或持久化错误。证据更符合压测机到单个公网入口的瞬时网络读超时，不支持服务端容量拐点或队列阻塞判断。 + +### 26.2 关闭竞态补丁结论 + +关闭竞态补丁的专项验收通过。两个 agent 均完成实时静默、5 分钟沉淀、历史回查、summary 生成和 120 秒渐进关闭，并以退出码 0 结束；没有提前停止进程。从负载开始到全部压测连接关闭后的完整 Pod 日志中： + +- `s.queueOut: session's send queue full` 为 0； +- Cluster P2M `queue_full` 增量为 0； +- WS send queue `queue_full` 增量为 0，三个 Pod 结算队列深度均为 0； +- 未出现旧 multiplex session 无消费者却继续积压消息的现象； +- 三个 Pod 均保持 Ready、0 重启，Cluster peer 均为 2/2。 + +这证明“清理 Cluster session 前先标记 terminating 并 purge channel”的修复已经消除 R6/R7 关闭末尾出现的孤儿发送队列积压。关闭期日志仅有 1 次客户端渐退附近的 `close 1006 unexpected EOF`，没有引发队列堆积或 Pod 状态异常。 + +### 26.3 服务端与压测端观测 + +按起压前和负载结束快照计算增量： + +| 指标 | 结果 | +|---|---:| +| `Cluster.TopicProxy` | 8227 次，平均 2.285ms，P95 在 10ms 桶内 | +| ctrl ACK enqueue | 1798 次，平均 62.11ms，P95 在 250ms 桶内 | +| ctrl ACK write | 1798 次，平均 62.43ms，P95 在 250ms 桶内 | +| Cluster P2M queue full | 0 | +| WS send queue full | 0 | +| Kafka consumer lag | 全部为 0 | + +A/B 回调队列峰值为 63/47，最大等待为 9.865/10.031ms，结算时等待和处理中均为 0。客户端回调处理没有形成瓶颈。MySQL 校验覆盖 `im_008_group_messages_202610` 与 `im_008_messages`，1796 条服务端已接受消息全部精确命中。 + +本轮结论应分开使用：**关闭竞态修复已通过专项复验；R8 整体业务门禁因一次公网断连失败。**如需获得“修复版本下业务与关闭运维门禁同时全绿”的单轮证据，应保持参数不变再执行 R9；无需因本轮结果继续修改服务端队列代码。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-mixed-r8-1002/report.json` +- `loadtest/im-pressure/results/s900m-mixed-r8-1002/db-verification.json` +- `loadtest/im-pressure/results/s900m-mixed-r8-1002/metrics.jsonl` +- `loadtest/im-pressure/results/s900m-mixed-r8-1002/server-metrics/` +- `loadtest/im-pressure/results/s900m-mixed-r8-1002/im-server-{0,1,2}.log` + +## 27. 900 人混合 R9 全量复验 + +Run ID：`s900m-mixed-r9-1002`。本轮保持 R8 的配置、账号、消息量和验收口径不变。起压前三个公网入口均返回 200，HTTPS 总耗时约 0.48 至 0.56 秒；三个 Pod 使用相同镜像，均为 Ready、0 重启。 + +| 指标 | R9 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 发送尝试 / 最终接受 | 1798 / 1798 | +| 明确拒绝 / 结果未知 | 0 / 0 | +| 超过 20 秒的 ACK | 10 | +| 精确 ACK P95 | 11417.014ms | +| 单聊对端实时投递 | 1350 / 1350 | +| 群聊实时投递 | 403200 / 403200 | +| 实时总投递 | 404550 / 404550 | +| WS 匹配回调总数（含单聊回显） | 405900 / 405900 | +| 实时到达率 / 丢失率 | 100% / 0% | +| 重复 / 乱序 / 断连 | 0 / 0 / 0 | +| 历史回查 | 1798 / 1798 | +| MySQL 精确命中 | 1798 / 1798 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| 负载期 queue full | 0 | +| 渐进关闭期 queue full | 1 | +| Pod Ready / 重启 | 3 / 3，0 重启 | + +### 27.1 业务门禁通过 + +本轮 `report.json` 为 `passed=true`。1798 条发送全部得到最终 ACK，404550 次对端投递全部实时到达；没有发送未知、实时丢失、重复、乱序或非预期断连。A/B 各收到 202950 次匹配回调，其中包含 675 次单聊发送方回显，因此底层回调总数为 405900。 + +历史接口和 MySQL 均精确命中 1798/1798。数据库校验覆盖 `im_008_group_messages_202610` 与 `im_008_messages`，缺失、错配、重复和额外行均为 0。Kafka consumer lag 全部为 0，三个 Pod 全程 Ready、0 重启。 + +A/B 回调队列峰值为 218/210，结算时均完全排空。按起压前和负载结束快照计算，服务端 `Cluster.TopicProxy` 新增 7857 次调用，平均 1.752ms、P95 在 5ms 桶内；ctrl ACK enqueue/write 各 1798 次，平均 72.513/72.812ms，P95 均在 500ms 桶内。负载期没有 Topic、Cluster、WS 队列满、RPC 失败、gob 或持久化错误。 + +因此 R9 已取得修复版本下完整的消息业务全绿证据，并再次验证跨 Pod 顺序修复有效。10 条超过 20 秒的 ACK 均在 90 秒命令窗口内返回，继续按既定口径只做趋势观测。 + +### 27.2 关闭期仍有一个低概率队列满 + +在最后一批连接退出时，Pod `im-server-1` 于 `2026-10-02T07:23:16Z` 记录： + +```text +s.queueOut: session's send queue full 8:grpQ5_hPdAz3cE-im-server-2 +topic[grpQ5_hPdAz3cE]: connection stuck, detaching - 8:grpQ5_hPdAz3cE-im-server-2 +cluster: session proxy closed 8:grpQ5_hPdAz3cE-im-server-2 +``` + +该事件发生在 summary 已生成、1798 条历史核验完成之后，没有影响消息正确性和持久化。结算时三个 Pod 的 WS send queue 与 Cluster P2M queue 深度均为 0，Kafka lag 为 0。 + +R9 证据修正了 R6/R7 时的部分根因判断。此次队列满发生时，`grp...-im-server-2` multiplex session 尚未终止；日志顺序是先队列满、Topic 判定连接卡住并 detach，最后 session 才关闭。因此 R8 发布的 `terminating + purgeChannels` 补丁只能防止“已进入终止流程的旧 session 被继续写入”，不能处理仍存活 multiplex session 在大量连接关闭时的瞬时响应突发。 + +当前 multiplex `send` channel 容量为 `sendQueueLimit + 32 = 160`。Cluster 写任务采用排他调度和同步 Master-to-Proxy RPC；正常负载期吞吐稳定，但关闭阶段多个 unsubscribe/presence 响应可在 drain worker 获得调度前快速填满 160 个槽位。R8 未复现、R9 复现 1 次，符合低概率调度竞态，而不是持续消费能力不足：R9 关闭阶段 `Cluster.TopicProxy` 仍完成 3195 次调用，平均约 0.9ms、P95 在 3ms 桶内。 + +后续修复不应只继续增加 terminating 判断。建议为 MULTIPLEX 出站队列增加一个很短的有界背压重试：首次非阻塞入队失败时确保 drain 已调度，并等待最多约 250ms 获取队列槽位；超时后才维持现有的 `queue full -> detach` 保护。这样可以吸收调度抖动和关闭突发，同时避免永久阻塞 Topic goroutine。需要增加“队列已满但 worker 很快释放槽位”与“worker 永久不消费时按时失败”两类回归测试，发版后再执行 R10 专项复验。 + +上述修复已在 R9 后完成：MULTIPLEX 出站队列首次满时会确保 Cluster drain 已调度，并在 250ms 有界窗口内等待一个槽位；只有持续无消费超过窗口才沿用原来的失败和 detach 保护。新增的 `clusterQueueMu` 将 terminating 检查与入队、以及终止标记与 channel 清理串行化，防止等待期间 session 恰好终止后消息重新进入已清空队列。普通 WebSocket、gRPC 和长轮询队列仍保持原有非阻塞语义，不受此次改动影响。 + +回归测试覆盖短暂队列满后恢复、持续不消费按时失败、等待期间 session 终止不残留消息，以及原有跨 Pod 顺序与终止拒绝写入场景。针对性测试普通模式连续 20 次、race 模式连续 10 次通过，`go test ./server ./loadtest/im-pressure` 全部通过。该修复尚待发版并执行 R10 复验。 + +本轮最终结论：**消息业务门禁通过；完整运维门禁仍因关闭期 1 次 multiplex queue full 未通过。** + +新增产物： + +- `loadtest/im-pressure/results/s900m-mixed-r9-1002/report.json` +- `loadtest/im-pressure/results/s900m-mixed-r9-1002/db-verification.json` +- `loadtest/im-pressure/results/s900m-mixed-r9-1002/metrics.jsonl` +- `loadtest/im-pressure/results/s900m-mixed-r9-1002/server-metrics/` +- `loadtest/im-pressure/results/s900m-mixed-r9-1002/im-server-{0,1,2}.log` diff --git a/server/cluster.go b/server/cluster.go index e9dc595..2247540 100644 --- a/server/cluster.go +++ b/server/cluster.go @@ -1401,8 +1401,10 @@ func (sess *Session) clusterWriteLoop(forTopic string) { // The topic may retain this multiplexing session briefly while its // unsubscribe request is processed. Reject writes to the stale session // instead of filling a send queue which no longer has a drain owner. + sess.clusterQueueMu.Lock() atomic.StoreInt32(&sess.terminating, 1) sess.purgeChannels() + sess.clusterQueueMu.Unlock() sess.closeRPC() globals.sessionStore.Delete(sess) sess.inflightReqs = nil diff --git a/server/cluster_order_test.go b/server/cluster_order_test.go index 618ceae..8659420 100644 --- a/server/cluster_order_test.go +++ b/server/cluster_order_test.go @@ -112,3 +112,54 @@ func TestClusterWriteLoopRejectsWritesAfterTermination(t *testing.T) { t.Fatalf("terminated multiplexing session retained %d outbound messages", len(sess.send)) } } + +func TestMultiplexEnqueueWaitsForDrain(t *testing.T) { + sess := &Session{proto: MULTIPLEX, send: make(chan any, 1)} + sess.send <- "existing" + + go func() { + time.Sleep(10 * time.Millisecond) + <-sess.send + }() + + payload := &ServerComMessage{TenantID: types.TenantID(1)} + if !sess.enqueueMultiplexOutbound(payload, 200*time.Millisecond) { + t.Fatal("multiplex enqueue failed while the drain was making progress") + } + if got := <-sess.send; got != payload { + t.Fatalf("unexpected queued payload: %#v", got) + } +} + +func TestMultiplexEnqueueTimesOutWithoutDrain(t *testing.T) { + sess := &Session{proto: MULTIPLEX, send: make(chan any, 1)} + sess.send <- "existing" + + started := time.Now() + if sess.enqueueMultiplexOutbound("next", 20*time.Millisecond) { + t.Fatal("multiplex enqueue succeeded without queue capacity") + } + if elapsed := time.Since(started); elapsed < 15*time.Millisecond || elapsed > 200*time.Millisecond { + t.Fatalf("multiplex enqueue timeout outside expected bounds: %s", elapsed) + } +} + +func TestMultiplexEnqueueStopsWhenSessionTerminates(t *testing.T) { + sess := &Session{proto: MULTIPLEX, send: make(chan any, 1)} + sess.send <- "existing" + + go func() { + time.Sleep(10 * time.Millisecond) + sess.clusterQueueMu.Lock() + atomic.StoreInt32(&sess.terminating, 1) + <-sess.send + sess.clusterQueueMu.Unlock() + }() + + if !sess.enqueueMultiplexOutbound("next", 200*time.Millisecond) { + t.Fatal("terminating multiplex session should discard the pending write") + } + if len(sess.send) != 0 { + t.Fatalf("terminating multiplex session retained %d outbound messages", len(sess.send)) + } +} diff --git a/server/session.go b/server/session.go index 030ef8f..28d2ccc 100644 --- a/server/session.go +++ b/server/session.go @@ -35,6 +35,11 @@ import ( // Maximum number of queued messages before session is considered stale and dropped. const sendQueueLimit = 128 +// Multiplex sessions may receive a short burst of presence updates while many +// remote connections are closing. Give the cluster drain a bounded chance to +// free one slot before treating the proxy as stuck. +const multiplexEnqueueWait = 250 * time.Millisecond + // Time given to a background session to terminate to avoid tiggering presence notifications. // If session terminates (or unsubscribes from topic) in this time frame notifications are not sent at all. const deferredNotificationsTimeout = time.Second * 5 @@ -171,6 +176,7 @@ type Session struct { // Ensures that one multiplexing session is drained by at most one cluster // write task at a time, preserving per-topic response order. clusterWriteScheduled atomic.Bool + clusterQueueMu sync.Mutex } // Subscription is a mapper of sessions to topics. @@ -190,6 +196,9 @@ type Subscription struct { } func (s *Session) enqueueOutbound(payload any) bool { + if s.isMultiplex() { + return s.enqueueMultiplexOutbound(payload, multiplexEnqueueWait) + } if s.proto != WEBSOCK { select { case s.send <- payload: @@ -212,6 +221,42 @@ func (s *Session) enqueueOutbound(payload any) bool { } } +func (s *Session) enqueueMultiplexOutbound(payload any, wait time.Duration) bool { + deadline := time.NewTimer(wait) + defer deadline.Stop() + retry := time.NewTicker(time.Millisecond) + defer retry.Stop() + + drainScheduled := false + for { + s.clusterQueueMu.Lock() + if atomic.LoadInt32(&s.terminating) > 0 { + s.clusterQueueMu.Unlock() + return true + } + select { + case s.send <- payload: + s.clusterQueueMu.Unlock() + return true + default: + s.clusterQueueMu.Unlock() + } + + if !drainScheduled { + s.scheduleClusterWriteLoop() + drainScheduled = true + } + select { + case <-retry.C: + case <-deadline.C: + s.clusterQueueMu.Lock() + terminating := atomic.LoadInt32(&s.terminating) > 0 + s.clusterQueueMu.Unlock() + return terminating + } + } +} + func (s *Session) observeOutboundDequeue(purged bool) { if s.proto != WEBSOCK { return