diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index e2d5347..6c61c68 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -694,3 +694,52 @@ ACK 从 WS 队列入队到 socket write 完成只增加约 0.3至0.6ms，而客 - `loadtest/im-pressure/results/s900m-mixed-r4-1001/metrics.jsonl` - `loadtest/im-pressure/results/s900m-mixed-r4-1001/server-metrics/` - `loadtest/im-pressure/results/s900m-mixed-r4-1001/im-server-{0,1,2}.log` + +## 24. 900 人混合 R5 顺序复测 + +Run ID：`s900m-mixed-r5-1001`。本轮在 R4 相同的 900 条 WebSocket、135 对单聊用户和 224 个群发送者负载下，验证长延迟直方图、Cluster gob 类型修复及消息顺序。 + +| 指标 | 结果 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 发送尝试 / 90 秒命令窗口内 ACK | 1798 / 1797 | +| 晚到 2xx / 最终接受 | 1 / 1798 | +| 明确拒绝 / 最终未知 | 0 / 0 | +| 超过 20 秒的 ACK | 375 | +| ctrl ACK P95 桶 / 精确 ACK P95 | 45s / 41746.854ms | +| 单聊对端实时投递 | 1350 / 1350 | +| 群聊实时投递 | 403200 / 403200 | +| 实时总投递 / 丢失 | 404550 / 404550，0 丢失 | +| 重复 / 乱序 / 断连 | 0 / 1 / 0 | +| 历史回查 | 1798 / 1798 | +| 无及时 ACK 消息历史命中 | 1 / 1 | +| MySQL 精确命中 | 1798 / 1798 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| Pod Ready / 重启 | 3 / 3，0 重启 | + +唯一超时消息 `s900m-mixed-r5-1001-g-a-0136-02` 在 95.028 秒收到 202，因此属于晚到成功，不是拒绝或丢失。压测报告已区分 `publish_accepted`（命令等待窗口内）与 `publish_final_accepted`（包含晚到 2xx）；用新口径重算后发送成功率为 100%，报告只因真实乱序而失败。长延迟直方图的 P95 正确落在 45 秒桶，不再退化为最大值。 + +本轮唯一乱序发生在 A 机 97 号连接的同一单聊 Topic：客户端先收到 seq=322，后收到 seq=321。两条消息由会话两端在几乎同一时刻发送，服务端均实时投递并精确落库，所以问题不在 MySQL、Kafka 或历史回查，而在 Topic master 到 proxy Pod 的实时转发顺序。 + +代码检查确认，同一个 multiplex session 每次入队都会调度 `clusterWriteLoop`；此前没有排他状态，多个 worker 可以同时消费同一个 `send` channel。随后使用 fire-and-forget RPC 时，先取出的 seq=321 与后取出的 seq=322 会在远端并发执行，导致 seq=322 先进入 proxy Topic。该竞态与本轮现象完全一致。 + +修复内容： + +1. multiplex session 增加原子调度所有权，同一 session 同时只允许一个 Cluster 写任务排空；释放所有权时再次检查队列，覆盖入队与退出之间的竞态。 +2. Master 到 Proxy 的同一 Topic 响应改为逐条等待 RPC 完成，保证远端处理完 seq=N 后才发送 seq=N+1；不同 Topic 仍由 worker pool 并行处理。 +3. 新增本地 RPC 回归测试，故意延迟 seq=1 的处理并重复调度同一 session，验证接收顺序始终为 1、2；该测试在 `go test -race` 下通过。 +4. 压测工具新增晚到 2xx 最终接受计数，已解决“最终已 202 但报告仍算发送失败”的重复判错。 + +服务端三个 Pod 共记录 1798 条 ctrl ACK，分布为 574/611/613；enqueue/write 平均约 60至66ms，P95 均不超过 250ms。未出现 Topic、Cluster proxy、WS send queue full、Kafka 错误、WS text write 错误或 Pod 重启。A/B 回调队列峰值为 253/290，最大等待 10.703/5.929ms，结算时均已排空。完整 Pod 日志中未再出现 `types.KVMap` 或其他 gob 注册错误，证明 R4 后的 Cluster gob 修复已生效。 + +本轮 MySQL 校验覆盖 `im_008_group_messages_202610` 与 `im_008_messages`，1798 条业务消息全部精确命中。结论是持久化、实时到达、长 ACK 统计和 gob 修复均通过，当前唯一阻断项为已完成代码修复但尚待发版复验的跨 Pod 顺序竞态。 + +下一步发版后执行完全相同的 900 人混合 R6。R6 通过条件为：最终接受 1798/1798、实时投递 404550/404550、重复/乱序/断连均为 0、历史和 MySQL 均为 1798/1798；ACK 延迟继续观测但不作为阻断门禁。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-mixed-r5-1001/report.json` +- `loadtest/im-pressure/results/s900m-mixed-r5-1001/db-verification.json` +- `loadtest/im-pressure/results/s900m-mixed-r5-1001/metrics.jsonl` +- `loadtest/im-pressure/results/s900m-mixed-r5-1001/server-metrics/` +- `loadtest/im-pressure/results/s900m-mixed-r5-1001/im-server-{0,1,2}.log` diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 0a94f05..2288053 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -40,6 +40,7 @@ type agentRun struct { publishCtrlObserved atomic.Int64 publishCtrlOver20 atomic.Int64 publishLateCtrl atomic.Int64 + publishLateAccepted atomic.Int64 mu sync.Mutex accepted []AcceptedMessage @@ -982,6 +983,9 @@ func (r *agentRun) observeCtrl(account AccountRecord, observation ctrlObservatio } if observation.Late { r.publishLateCtrl.Add(1) + if observation.Code >= 200 && observation.Code < 300 { + r.publishLateAccepted.Add(1) + } } if observation.Late || observation.Latency > 20*time.Second { r.mu.Lock() @@ -1001,6 +1005,7 @@ func (r *agentRun) applyCtrlObservations(summary *AgentSummary) { summary.PublishCtrlObserved = r.publishCtrlObserved.Load() summary.PublishCtrlOver20 = r.publishCtrlOver20.Load() summary.PublishLateCtrl = r.publishLateCtrl.Load() + summary.PublishLateAccepted = r.publishLateAccepted.Load() summary.PublishCtrlUnresolved = r.unresolvedCtrl() summary.PublishOutcomeUnknown = r.outcomeUnknown() summary.PublishCtrlLatency = r.ctrlLatency.Snapshot() diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index f054290..e5a60ea 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "sort" + "strings" "time" ) @@ -18,6 +19,8 @@ type CombinedReport struct { ReadyConnections int `json:"ready_connections"` PublishAttempted int64 `json:"publish_attempted"` PublishAccepted int64 `json:"publish_accepted"` + PublishLateAccepted int64 `json:"publish_late_accepted"` + PublishFinalAccepted int64 `json:"publish_final_accepted"` PublishRejected int64 `json:"publish_rejected"` PublishExplicitReject int64 `json:"publish_explicitly_rejected"` PublishOutcomeUnknown int64 `json:"publish_outcome_unknown"` @@ -80,6 +83,17 @@ func runMerge(aPath, bPath, output string) error { report.ReadyConnections += summary.ReadyConnections report.PublishAttempted += summary.PublishAttempted report.PublishAccepted += summary.PublishAccepted + lateAccepted := summary.PublishLateAccepted + // Backfill summaries generated before publish_ctrl_late_accepted was added. + if summary.CtrlTrackingEnabled && lateAccepted == 0 { + for _, sample := range summary.CtrlObservationSamples { + if sample.Late && sample.Code >= 200 && sample.Code < 300 { + lateAccepted++ + } + } + } + report.PublishLateAccepted += lateAccepted + report.PublishFinalAccepted += summary.PublishAccepted + lateAccepted for category, count := range summary.PublishErrors { report.PublishErrors[category] += count } @@ -129,12 +143,20 @@ func runMerge(aPath, bPath, output string) error { report.PersistenceLost += summary.PersistenceLost report.CrossClockSkewSamples += summary.CrossClockSkewSamples report.Missing = append(report.Missing, summary.Missing...) - if len(summary.Errors) > 0 { - report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("agent %s errors=%d", summary.Machine, len(summary.Errors))) + actionableErrors := len(summary.Errors) + lateTimeouts := int(lateAccepted) + for _, message := range summary.Errors { + if lateTimeouts > 0 && strings.Contains(message, "ctrl timeout") { + actionableErrors-- + lateTimeouts-- + } + } + if actionableErrors > 0 { + report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("agent %s unresolved errors=%d", summary.Machine, actionableErrors)) } } if report.PublishAttempted > 0 { - report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted) + report.PublishSuccessRate = float64(report.PublishFinalAccepted) / float64(report.PublishAttempted) } report.AckP95MS = mergedAckP95(report.Agents) report.PublishCtrlP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.PublishCtrlLatency }) diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index b59e4e3..2a312a4 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -148,6 +148,43 @@ func TestRunMergeReportsSameClockDeliveryLatency(t *testing.T) { } } +func TestRunMergeTreatsLateSuccessAsFinalAcceptance(t *testing.T) { + dir := t.TempDir() + late := AgentSummary{ + RunID: "run1", Scenario: "group", Machine: "a", ExpectedConnections: 1, ReadyConnections: 1, + PublishAttempted: 1, PublishAccepted: 0, CtrlTrackingEnabled: true, + PublishCtrlObserved: 1, PublishLateCtrl: 1, PublishCtrlUnresolved: 0, + PublishErrors: map[string]int64{"ctrl_timeout": 1}, + CtrlObservationSamples: []CtrlObservationSample{{Code: 202, Late: true}}, + Errors: []string{"publish message-1: ctrl timeout id=pub-a-1"}, + } + timely := AgentSummary{ + RunID: "run1", Scenario: "group", Machine: "b", ExpectedConnections: 1, ReadyConnections: 1, + PublishAttempted: 1, PublishAccepted: 1, CtrlTrackingEnabled: true, + PublishCtrlObserved: 1, PublishErrors: map[string]int64{}, + } + aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json") + writeSummaryForTest(t, aPath, late) + writeSummaryForTest(t, bPath, timely) + if err := runMerge(aPath, bPath, reportPath); err != nil { + t.Fatal(err) + } + var report CombinedReport + raw, err := os.ReadFile(reportPath) + if err != nil { + t.Fatal(err) + } + if err = json.Unmarshal(raw, &report); err != nil { + t.Fatal(err) + } + if report.PublishAccepted != 1 || report.PublishLateAccepted != 1 || report.PublishFinalAccepted != 2 || report.PublishSuccessRate != 1 { + t.Fatalf("late acceptance was not included in final publish outcome: %+v", report) + } + if !report.Passed || len(report.FailureReasons) != 0 { + t.Fatalf("resolved late ACK must not fail the report: %+v", report.FailureReasons) + } +} + func writeSummaryForTest(t *testing.T, path string, summary AgentSummary) { t.Helper() raw, err := json.Marshal(summary) diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index 93af4d8..85b1441 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -424,6 +424,7 @@ type AgentSummary struct { PublishCtrlObserved int64 `json:"publish_ctrl_observed"` PublishCtrlOver20 int64 `json:"publish_ctrl_over_20s"` PublishLateCtrl int64 `json:"publish_ctrl_late"` + PublishLateAccepted int64 `json:"publish_ctrl_late_accepted"` PublishCtrlUnresolved int64 `json:"publish_ctrl_unresolved"` PublishCtrlLatency HistogramSnapshot `json:"publish_ctrl_latency"` CtrlObservationSamples []CtrlObservationSample `json:"ctrl_observation_samples,omitempty"` diff --git a/server/cluster.go b/server/cluster.go index bf14e52..61ea700 100644 --- a/server/cluster.go +++ b/server/cluster.go @@ -489,14 +489,16 @@ func clusterReqShard(msg *ClusterReq, count int) int { return int(h.Sum32() % uint32(count)) } -// masterToProxyAsync forwards response from topic master to topic proxy -// in a fire-and-forget manner. -func (n *ClusterNode) masterToProxyAsync(msg *ClusterResp) error { - var unused bool - if c := n.callAsync("Cluster.TopicProxy", msg, &unused, nil); c.Error != nil { - return c.Error +// masterToProxy forwards responses synchronously. A multiplexing session drains +// one topic in order, so waiting for each RPC prevents the RPC server from +// concurrently enqueueing later sequence IDs before earlier ones. +func (n *ClusterNode) masterToProxy(msg *ClusterResp) error { + var rejected bool + err := n.call("Cluster.TopicProxy", msg, &rejected) + if err == nil && rejected { + return errClusterOverloaded } - return nil + return err } // route routes server message within the cluster. @@ -1437,7 +1439,7 @@ func (sess *Session) clusterWriteLoop(forTopic string) { srvMsg.RcptTo = forTopic response.RcptTo = forTopic - if err := sess.clnode.masterToProxyAsync(response); err != nil { + if err := sess.clnode.masterToProxy(response); err != nil { logs.Warn.Printf("cluster: response to proxy failed \"%s\": %s", sess.sid, err.Error()) return } @@ -1454,6 +1456,12 @@ func (sess *Session) clusterWriteLoop(forTopic string) { case <-sess.detach: return default: + sess.clusterWriteScheduled.Store(false) + // An enqueue can race with releasing ownership. Either the enqueuer + // schedules the next drain or this check does; the CAS prevents both. + if len(sess.send) > 0 { + sess.scheduleClusterWriteLoop() + } terminate = false return } diff --git a/server/cluster_order_test.go b/server/cluster_order_test.go new file mode 100644 index 0000000..f84a1fe --- /dev/null +++ b/server/cluster_order_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "net" + "net/rpc" + "sync" + "testing" + "time" + + "github.com/tinode/chat/server/concurrency" + "github.com/tinode/chat/server/store/types" +) + +type orderedTopicProxy struct { + mu sync.Mutex + received []int64 + done chan struct{} +} + +func (proxy *orderedTopicProxy) TopicProxy(msg *ClusterResp, rejected *bool) error { + if msg.SrvMsg.Data.SeqId == 1 { + time.Sleep(50 * time.Millisecond) + } + proxy.mu.Lock() + proxy.received = append(proxy.received, msg.SrvMsg.Data.SeqId) + if len(proxy.received) == 2 { + close(proxy.done) + } + proxy.mu.Unlock() + return nil +} + +func TestClusterWriteLoopPreservesTopicOrder(t *testing.T) { + serverConn, clientConn := net.Pipe() + rpcServer := rpc.NewServer() + proxy := &orderedTopicProxy{done: make(chan struct{})} + if err := rpcServer.RegisterName("Cluster", proxy); err != nil { + t.Fatal(err) + } + go rpcServer.ServeConn(serverConn) + + pool := concurrency.NewGoRoutinePool(4) + previousCluster := globals.cluster + globals.cluster = &Cluster{proxyEventQueue: pool} + t.Cleanup(func() { + globals.cluster = previousCluster + _ = clientConn.Close() + _ = serverConn.Close() + pool.Stop() + }) + + sess := &Session{ + sid: "multiplex-order-test", + proto: MULTIPLEX, + proxiedTopic: "p2p-order-test", + send: make(chan any, 2), + clnode: &ClusterNode{endpoint: rpc.NewClient(clientConn), connected: true}, + } + tenantID := types.TenantID(1) + sess.send <- &ServerComMessage{TenantID: tenantID, Data: &MsgServerData{SeqId: 1}} + sess.send <- &ServerComMessage{TenantID: tenantID, Data: &MsgServerData{SeqId: 2}} + + // Scheduling twice used to start two concurrent drains for the same topic. + sess.scheduleClusterWriteLoop() + sess.scheduleClusterWriteLoop() + + select { + case <-proxy.done: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for cluster responses") + } + deadline := time.Now().Add(2 * time.Second) + for sess.clusterWriteScheduled.Load() && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if sess.clusterWriteScheduled.Load() { + t.Fatal("cluster write task did not release session ownership") + } + + proxy.mu.Lock() + defer proxy.mu.Unlock() + if len(proxy.received) != 2 || proxy.received[0] != 1 || proxy.received[1] != 2 { + t.Fatalf("cluster responses arrived out of order: %v", proxy.received) + } +} diff --git a/server/session.go b/server/session.go index 4843a90..030ef8f 100644 --- a/server/session.go +++ b/server/session.go @@ -168,6 +168,9 @@ type Session struct { // Type of proxy to master request being handled. proxyReq ProxyReqType + // 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 } // Subscription is a mapper of sessions to topics. @@ -332,7 +335,7 @@ func (s *Session) isCluster() bool { } func (s *Session) scheduleClusterWriteLoop() { - if globals.cluster != nil && globals.cluster.proxyEventQueue != nil { + if globals.cluster != nil && globals.cluster.proxyEventQueue != nil && s.clusterWriteScheduled.CompareAndSwap(false, true) { globals.cluster.proxyEventQueue.Schedule( func() { s.clusterWriteLoop(s.proxiedTopic) }) }