fix: preserve cross-pod message order

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`

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()

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 {
+		actionableErrors := len(summary.Errors)
-			report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("agent %s errors=%d", summary.Machine, 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 })

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)

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"`

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
+// masterToProxy forwards responses synchronously. A multiplexing session drains
-// in a fire-and-forget manner.
+// one topic in order, so waiting for each RPC prevents the RPC server from
-func (n *ClusterNode) masterToProxyAsync(msg *ClusterResp) error {
+// concurrently enqueueing later sequence IDs before earlier ones.
-	var unused bool
+func (n *ClusterNode) masterToProxy(msg *ClusterResp) error {
-	if c := n.callAsync("Cluster.TopicProxy", msg, &unused, nil); c.Error != nil {
+	var rejected bool
-		return c.Error
+	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
 		}

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)
+	}
+}

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) })
 	}