From b1545b687682dc0070b940c0a2d7bc43a52ee15a Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Tue, 29 Sep 2026 17:09:16 +0800 Subject: [PATCH] fix: keep load connections through verification --- .../IM-单机60%容量压力测试结果-20260929.md | 31 ++++- loadtest/im-pressure/agent.go | 113 +++++++++++++----- 2 files changed, 111 insertions(+), 33 deletions(-) diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index 02ddfbc..58063b7 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -109,11 +109,40 @@ R1 的历史查询将 `limit=10` 错误地按目标消息数计算，没有计 3. 群聊通过后再跑单机混合 60%，要求 ACK、历史和独立 SQL 三方账本一致，连接断开数为 0。 4. 补齐 Prometheus/metrics-server 权限并同步保存 Pod CPU、内存、网络、Kafka lag、Redis、MySQL 指标；在此之前不进行双机 100% 最终验收。 -## 9. 产物 +## 9. 发版后群聊 R2 + +Run ID：`s60-group-r2-0929`。 + +本轮开始前 3 个 Pod 均为 Running、0 重启，账号和群成员核验为 `accounts=1200 pairs=600 group_members=1200`。测试完整运行到 A/B 历史验证完成，并执行了错峰关闭和独立 MySQL 核验。 + +| 指标 | 结果 | +|---|---:| +| 发送尝试 / ACK | 1200 / 871 | +| 结果未知 | 329（222 timeout、107 socket） | +| 断连 | 360（315 reset、44 abnormal close、1 timeout） | +| ACK P95 | 16630.441ms | +| 投递 P95 | 101823.156ms | +| ACK 台账期望 / 实时收到 | 1045200 / 889777 | +| 实时缺失 | 155423，最终全部查到持久化 | +| 历史回查 | 1080 / 1080 | +| DB：ACK 精确命中 | 871 / 871 | +| DB：无 ACK 但落库 | 209 | +| DB：缺失 / 错配 / 重复 | 0 / 0 / 0 | + +329 条结果未知中，209 条已被 WS 观测且独立 SQL 确认落库，剩余 120 条数据库中不存在。断连样本主要是公网对端 `connection reset by peer`，且样本连接均指向 `64.118.158.131:443`；本机到域名全部当前 IP 均使用 `en1` 静态直连，没有经过 VPN/TUN。 + +服务端指标显示群 Topic 所在 `im-server-0` 的 1080 条 `save` 全部低于 100ms，Kafka consumer lag 为 0；但 `topic_server` 队列累计出现 393 次 full。压测客户端 CPU、RSS 不构成瓶颈。 + +本轮不能作为提交 `75bb448` 的有效回归：Pod 中运行的二进制不包含该提交新增的 `permission_check`、`realtime_fanout` 阶段字符串和好友申请空结果保护日志，Prometheus 也没有相应时序，说明实际部署镜像不是该提交构建结果。必须正确部署 `75bb448` 或后续提交后重跑。 + +此外，本轮发现 agent 只在正式开始前维护连接，统计窗口内断线后不会恢复，导致历史核验依赖剩余连接。后续已改为持续维护到 A/B 历史验证均完成，并将重连并发限制为每 agent 16，避免恢复风暴；每次计划发送前重新取得当前连接。 + +## 10. 产物 - 方案：`docs/test/IM-单机60%容量压力测试方案.md` - 单聊 R2：`loadtest/im-pressure/results/s60-p2p-r2-0929/` - 群聊 R1：`loadtest/im-pressure/results/s60-group-r1-0929/` - 混合 R1：`loadtest/im-pressure/results/s60-mixed-r1-0929/` +- 群聊 R2：`loadtest/im-pressure/results/s60-group-r2-0929/` 结果目录包含 coordinator、A/B summary、合并 report 和 agent 指标 JSONL。 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 0f5d6a3..e556cea 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -154,31 +154,31 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo defer close(maintenanceDone) run.maintainConnections(maintenanceCtx, accounts, manifest.GroupTopic, apiKey, password) }() + var stopMaintenanceOnce sync.Once + stopConnectionMaintenance := func() { + stopMaintenanceOnce.Do(func() { + stopMaintenance() + <-maintenanceDone + }) + } + defer stopConnectionMaintenance() start, err := coord.ready(ctx, func() AgentReady { return AgentReady{ Machine: machine, ReadyConnections: run.aliveConnections(), ExpectedConnections: len(accounts), } }) if err != nil { - stopMaintenance() - <-maintenanceDone return err } if !runIDPattern.MatchString(start.RunID) || !validScenario(start.Scenario) { - stopMaintenance() - <-maintenanceDone return fmt.Errorf("invalid coordinator start: %+v", start) } run.runID = start.RunID run.scenario = start.Scenario run.recorder.setRunID(start.RunID) if err = waitUntil(ctx, start.StartAt); err != nil { - stopMaintenance() - <-maintenanceDone return err } - stopMaintenance() - <-maintenanceDone if alive := run.aliveConnections(); alive != len(accounts) { return fmt.Errorf("connections alive at scheduled start=%d expected=%d", alive, len(accounts)) } @@ -241,6 +241,9 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo if err != nil { return err } + // Both agents have completed history verification. Connection maintenance can + // now stop without making the other agent verify through stale sockets. + stopConnectionMaintenance() summary.HistoryChecked = int64(history.checked) summary.HistoryFound = int64(history.found) summary.HistoryUnackedChecked = int64(history.unackedChecked) @@ -601,29 +604,57 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe if ctx.Err() != nil { return } + jobs := make(chan AccountRecord) var wg sync.WaitGroup - for index, account := range byIndex { + workers := 16 + if workers > len(accounts) { + workers = len(accounts) + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + for account := range jobs { + index := account.Index + previous := r.client(index) + if previous != nil && previous.alive() { + continue + } + if previous != nil { + previous.Close() + } + client, err := r.connectAccountWithRetry(ctx, account, groupTopic, apiKey, password) + if err != nil { + if ctx.Err() == nil { + log.Printf("account=%s reconnect failed: %v", account.Username, err) + } + continue + } + r.setClient(index, client) + log.Printf("account=%s connection restored", account.Username) + } + }() + } + indexes := make([]int, 0, len(byIndex)) + for index := range byIndex { + indexes = append(indexes, index) + } + sort.Ints(indexes) + for _, index := range indexes { + account := byIndex[index] current := r.client(index) if current != nil && current.alive() { continue } - wg.Add(1) - go func(index int, account AccountRecord, previous *WSClient) { - defer wg.Done() - if previous != nil { - previous.Close() - } - client, err := r.connectAccountWithRetry(ctx, account, groupTopic, apiKey, password) - if err != nil { - if ctx.Err() == nil { - log.Printf("account=%s pre-start reconnect failed: %v", account.Username, err) - } - return - } - r.setClient(index, client) - log.Printf("account=%s pre-start connection restored", account.Username) - }(index, account, current) + select { + case jobs <- account: + case <-ctx.Done(): + close(jobs) + wg.Wait() + return + } } + close(jobs) wg.Wait() select { case <-ctx.Done(): @@ -714,12 +745,11 @@ func scaledUsers(total, numerator, denominator int) int { func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, messages int, interval time.Duration) error { var wg sync.WaitGroup for index := 1; index <= users; index++ { - client := r.client(index) - if client == nil { + if r.client(index) == nil { return fmt.Errorf("missing client %d", index) } wg.Add(1) - go func(index int, client *WSClient) { + go func(index int) { defer wg.Done() jitter := time.Duration((index*37)%200) * time.Millisecond for messageIndex := 1; messageIndex <= messages; messageIndex++ { @@ -728,6 +758,11 @@ func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, mess return } mid := fmt.Sprintf("%s-p-%s-%04d-%02d", r.runID, r.machine, index, messageIndex) + client := r.client(index) + if client == nil || !client.alive() { + r.recordUnavailablePublish(mid) + continue + } accepted, err := r.publish(ctx, client, client.account.P2PTopic, mid, messageIndex) if err == nil { accepted.PeerIndex = index @@ -739,7 +774,7 @@ func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, mess r.addAccepted(accepted) } } - }(index, client) + }(index) } wg.Wait() return ctx.Err() @@ -757,10 +792,9 @@ func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic str } var wg sync.WaitGroup for index := first; index <= last; index++ { - client := r.client(index) position := index - first wg.Add(1) - go func(index, position int, client *WSClient) { + go func(index, position int) { defer wg.Done() for round := 1; round <= rounds; round++ { target := start.Add(time.Duration(round-1)*roundDuration + time.Duration(position)*200*time.Millisecond + offset) @@ -768,17 +802,32 @@ func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic str return } mid := fmt.Sprintf("%s-g-%s-%04d-%02d", r.runID, r.machine, index, round) + client := r.client(index) + if client == nil || !client.alive() { + r.recordUnavailablePublish(mid) + continue + } accepted, err := r.publish(ctx, client, topic, mid, round) if err == nil { r.addAccepted(accepted) } } - }(index, position, client) + }(index, position) } wg.Wait() return ctx.Err() } +func (r *agentRun) recordUnavailablePublish(mid string) { + r.attempted.Add(1) + r.publishSocket.Add(1) + r.mu.Lock() + if len(r.errors) < 1000 { + r.errors = append(r.errors, fmt.Sprintf("publish %s: websocket connection unavailable", mid)) + } + r.mu.Unlock() +} + func (r *agentRun) publish(ctx context.Context, client *WSClient, topic, mid string, messageIndex int) (AcceptedMessage, error) { r.attempted.Add(1) accepted, err := client.Publish(ctx, topic, mid, r.runID, r.scenario, messageIndex)