perf: expose and reduce message publish latency

docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md

@@ -482,3 +482,34 @@ Run ID：`p2p500-fix-20260927-02`。A/B 各建立 500 条 WebSocket，共 500
 - `tenant.updated`、`tenant.status.changed`、`tenant.limits.updated`、`tenant.features.updated` 和 `cache.tenant.invalidate` 继续使用现有 Kafka 广播事件；`InvalidateTenantConfigCache` 现在同时使数据库路由缓存失效。失效与刷新并发时通过 generation 校验阻止旧查询重新写入有效缓存。
 - 压测配置新增 `history_settle`，默认 5 分钟。实时 drain 完成后先等待该窗口，再发起历史查询，从而单独测量 Kafka 自然排空时间。
 - 历史查询默认并发从每机 48 降为 12；历史读取高并发能力后续使用独立读写混合场景验证，不再污染消息落库吞吐结果。
+
+## 17. 路由缓存发布后 500 对复测与延迟改造
+
+Run ID：`p2p500-r3-20260927-03`。A/B 各 500 条 WebSocket、每人发送 10 条消息，总计 10000 次 Publish。
+
+| 指标 | 结果 |
+|---|---:|
+| WS Ready | 1000/1000 |
+| Publish ACK | 10000/10000 |
+| 实时到达 / 丢失 | 10000/10000 / 0 |
+| 历史回查 / 落库丢失 | 10000/10000 / 0 |
+| 重复 / 乱序 / 断连 | 0 / 0 / 0 |
+| 总体精确 ACK P95 | 1308.150 ms |
+| A / B 精确 ACK P95 | 1121.2 / 1519.2 ms |
+| 原报告投递 P95 | 2000 ms（受双机时钟偏差污染） |
+| Kafka 最终 lag | 0 |
+| `hotgo` pool wait | 0 |
+
+租户路由缓存已经消除 `hotgo` 热路径等待，消息正确性和最终落库全部通过。剩余 ACK 延迟集中在消息 4–6 的突发积压阶段。发压窗口内，`meeting_im` 连接池出现明显等待；同时 Kafka persistence consumer 正在使用同一连接池。Cluster Route RPC 平均约 1–2ms，P2M 队列始终为 0，因此跨 Pod RPC 不是秒级延迟来源。
+
+原投递 P95 不能作为服务端结论：压测程序使用发送机绝对时间戳在接收机计算单向耗时。A 只接受了 5777 个非负样本，而 B 接受了全部 10000 个样本，说明 B 时钟领先 A；A 丢弃负值，B 则把时钟偏差计入延迟。
+
+本次完成以下改造：
+
+1. 压测程序新增同机与跨机两个投递直方图；`delivery_local_p95_ms` 作为门禁，跨机负延迟计入 `cross_machine_negative_latency_samples`，不再静默丢弃后继续生成误导性 P95。
+2. 压测消息显式携带 `sender_machine`，旧消息仍可从 `client_mid` 兼容识别来源。
+3. 好友关系首次数据库确认后双向回填 Redis，减少双向聊天后续回源。
+4. 新增 `im_message_publish_stage_duration_seconds{kind,stage,result}`，覆盖 `relationship_check`、`redis_idempotency`、`redis_sequence`、`kafka_produce` 和整体 `save` 阶段。
+5. 新增 `log_client_packets` 配置，默认关闭逐包 INFO 日志，避免每条消息额外 JSON 解析和容器 stdout I/O；需要诊断时才短期开启。
+
+下一轮仍使用 500 对复测。重点比较 `relationship_check`、`kafka_produce` 和 `save` 的 P95，并以 `delivery_local_p95_ms` 判断投递延迟；跨机延迟只有在 chrony/NTP 偏差确认小于 5ms 后才用于验收。数据库连接池暂不扩大，避免再次超过 MySQL 全局 170 连接预算；若 `relationship_check` 已稳定但 pool wait 与 Kafka lag 仍持续增长，再设计 persistence 独立连接预算或批量更新方案。

loadtest/im-pressure/README.md

@@ -40,7 +40,7 @@
 - 如果由机器 A 集中采集 agent 指标，机器 A 必须能访问 `http://机器B内网IP:9105/metrics`。
 - 两台机器都必须能访问配置中的 HTTP API 和各自的 WS 地址。
 - `8090` 是无鉴权的压测协调端口，只允许压测内网访问，不要暴露到公网。
-- 两台机器必须使用 NTP/chrony 同步时间；端到端延迟依赖两边时钟一致。
+- 两台机器必须使用 NTP/chrony 同步时间。报告会分别输出同机时钟延迟和跨机器单向延迟；跨机器出现负延迟样本时，后者仅用于诊断，不参与门禁。
 
 ## 统计口径
 
@@ -51,6 +51,8 @@
 - 每轮建立 WS 后、正式计时前，以 32 个 worker 对本机 1000 个账号做好友关系、P2P Topic 和发布权限预检；任一异常会终止本轮。
 - 历史回查默认每机使用 12 个 worker、总超时 3 分钟，并在超时内持续重试缺失区间。历史读取并发能力应使用独立场景测试，不与落库吞吐测试混在一起。
 - publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`，不受文本错误样本上限影响。
+- `delivery_local_p95_ms` 使用发送方本机收到 echo 的同机时钟样本，是投递延迟门禁口径；兼容字段 `delivery_p95_ms` 与它保持一致。
+- `delivery_cross_machine_p95_ms` 是跨机器单向观测值，只有两机时钟同步达标时才有意义；`cross_machine_negative_latency_samples > 0` 明确表示存在时钟偏差。
 - `manifest.json` 含登录 token，文件权限为 `0600`，不得提交到 Git 或放入结果包。
 
 ## 两台机器初始化

loadtest/im-pressure/merge.go

@@ -10,35 +10,38 @@ import (
 )
 
 type CombinedReport struct {
-	RunID               string              `json:"run_id"`
+	RunID                 string              `json:"run_id"`
-	Scenario            string              `json:"scenario"`
+	Scenario              string              `json:"scenario"`
-	GeneratedAt         time.Time           `json:"generated_at"`
+	GeneratedAt           time.Time           `json:"generated_at"`
-	Agents              []AgentSummary      `json:"agents"`
+	Agents                []AgentSummary      `json:"agents"`
-	ExpectedConnections int                 `json:"expected_connections"`
+	ExpectedConnections   int                 `json:"expected_connections"`
-	ReadyConnections    int                 `json:"ready_connections"`
+	ReadyConnections      int                 `json:"ready_connections"`
-	PublishAttempted    int64               `json:"publish_attempted"`
+	PublishAttempted      int64               `json:"publish_attempted"`
-	PublishAccepted     int64               `json:"publish_accepted"`
+	PublishAccepted       int64               `json:"publish_accepted"`
-	PublishRejected     int64               `json:"publish_rejected"`
+	PublishRejected       int64               `json:"publish_rejected"`
-	PublishErrors       map[string]int64    `json:"publish_errors,omitempty"`
+	PublishErrors         map[string]int64    `json:"publish_errors,omitempty"`
-	PublishSuccessRate  float64             `json:"publish_success_rate"`
+	PublishSuccessRate    float64             `json:"publish_success_rate"`
-	AckP95MS            float64             `json:"ack_p95_ms"`
+	AckP95MS              float64             `json:"ack_p95_ms"`
-	DeliveryP95MS       float64             `json:"delivery_p95_ms"`
+	DeliveryP95MS         float64             `json:"delivery_p95_ms"`
-	ExpectedDeliveries  int64               `json:"expected_deliveries"`
+	DeliveryLocalP95MS    float64             `json:"delivery_local_p95_ms"`
-	ReceivedDeliveries  int64               `json:"received_deliveries"`
+	DeliveryCrossP95MS    float64             `json:"delivery_cross_machine_p95_ms"`
-	RealtimeLost        int64               `json:"realtime_lost"`
+	CrossClockSkewSamples int64               `json:"cross_machine_negative_latency_samples"`
-	RealtimeLostStored  int64               `json:"realtime_lost_but_persisted"`
+	ExpectedDeliveries    int64               `json:"expected_deliveries"`
-	ArrivalRate         float64             `json:"arrival_rate"`
+	ReceivedDeliveries    int64               `json:"received_deliveries"`
-	LossRate            float64             `json:"loss_rate"`
+	RealtimeLost          int64               `json:"realtime_lost"`
-	Duplicates          int64               `json:"duplicates"`
+	RealtimeLostStored    int64               `json:"realtime_lost_but_persisted"`
-	OutOfOrder          int64               `json:"out_of_order"`
+	ArrivalRate           float64             `json:"arrival_rate"`
-	OrderingViolations  []OrderingViolation `json:"ordering_violations,omitempty"`
+	LossRate              float64             `json:"loss_rate"`
-	Disconnects         int64               `json:"disconnects"`
+	Duplicates            int64               `json:"duplicates"`
-	HistoryChecked      int64               `json:"history_checked"`
+	OutOfOrder            int64               `json:"out_of_order"`
-	HistoryFound        int64               `json:"history_found"`
+	OrderingViolations    []OrderingViolation `json:"ordering_violations,omitempty"`
-	PersistenceLost     int64               `json:"persistence_lost"`
+	Disconnects           int64               `json:"disconnects"`
-	Missing             []MissingDelivery   `json:"missing,omitempty"`
+	HistoryChecked        int64               `json:"history_checked"`
-	Passed              bool                `json:"passed"`
+	HistoryFound          int64               `json:"history_found"`
-	FailureReasons      []string            `json:"failure_reasons,omitempty"`
+	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 {
@@ -79,6 +82,7 @@ func runMerge(aPath, bPath, output string) error {
 		report.HistoryChecked += summary.HistoryChecked
 		report.HistoryFound += summary.HistoryFound
 		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)))
@@ -88,7 +92,14 @@ func runMerge(aPath, bPath, output string) error {
 		report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted)
 	}
 	report.AckP95MS = mergedAckP95(report.Agents)
-	report.DeliveryP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatency })
+	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
+	// samples. Cross-machine wall clocks cannot produce a valid one-way latency.
+	report.DeliveryP95MS = report.DeliveryLocalP95MS
+	if report.DeliveryP95MS == 0 {
+		report.DeliveryP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatency })
+	}
 	if report.ExpectedDeliveries > 0 {
 		report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries)
 		report.LossRate = float64(report.RealtimeLost) / float64(report.ExpectedDeliveries)

loadtest/im-pressure/merge_test.go

@@ -80,6 +80,37 @@ func TestMergedAckP95FallsBackWhenRawSamplesAreIncomplete(t *testing.T) {
 	}
 }
 
+func TestRunMergeUsesSameClockDeliveryLatencyForGate(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}}
+	a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ExpectedConnections: 1,
+		ReadyConnections: 1, PublishAttempted: 1, PublishAccepted: 1, ExpectedPeer: 1, ReceivedPeer: 1,
+		DeliveryLatencyLocal: local, DeliveryLatencyCross: cross, CrossClockSkewSamples: 3}
+	b := a
+	b.Machine = "b"
+	aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json")
+	writeSummaryForTest(t, aPath, a)
+	writeSummaryForTest(t, bPath, b)
+	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.DeliveryP95MS != 200 || report.DeliveryLocalP95MS != 200 || report.DeliveryCrossP95MS != 5000 {
+		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)
+	}
+}
+
 func writeSummaryForTest(t *testing.T, path string, summary AgentSummary) {
 	t.Helper()
 	raw, err := json.Marshal(summary)

loadtest/im-pressure/model.go

@@ -350,40 +350,43 @@ type AcceptedMessage struct {
 }
 
 type AgentSummary struct {
-	RunID               string              `json:"run_id"`
+	RunID                 string              `json:"run_id"`
-	Scenario            string              `json:"scenario"`
+	Scenario              string              `json:"scenario"`
-	Machine             string              `json:"machine"`
+	Machine               string              `json:"machine"`
-	StartedAt           time.Time           `json:"started_at"`
+	StartedAt             time.Time           `json:"started_at"`
-	LoadEndedAt         time.Time           `json:"load_ended_at"`
+	LoadEndedAt           time.Time           `json:"load_ended_at"`
-	FinishedAt          time.Time           `json:"finished_at"`
+	FinishedAt            time.Time           `json:"finished_at"`
-	ExpectedConnections int                 `json:"expected_connections"`
+	ExpectedConnections   int                 `json:"expected_connections"`
-	ReadyConnections    int                 `json:"ready_connections"`
+	ReadyConnections      int                 `json:"ready_connections"`
-	PublishAttempted    int64               `json:"publish_attempted"`
+	PublishAttempted      int64               `json:"publish_attempted"`
-	PublishAccepted     int64               `json:"publish_accepted"`
+	PublishAccepted       int64               `json:"publish_accepted"`
-	PublishRejected     int64               `json:"publish_rejected"`
+	PublishRejected       int64               `json:"publish_rejected"`
-	PublishErrors       map[string]int64    `json:"publish_errors,omitempty"`
+	PublishErrors         map[string]int64    `json:"publish_errors,omitempty"`
-	ExpectedPeer        int64               `json:"expected_peer"`
+	ExpectedPeer          int64               `json:"expected_peer"`
-	ReceivedPeer        int64               `json:"received_peer"`
+	ReceivedPeer          int64               `json:"received_peer"`
-	ExpectedEcho        int64               `json:"expected_echo"`
+	ExpectedEcho          int64               `json:"expected_echo"`
-	ReceivedEcho        int64               `json:"received_echo"`
+	ReceivedEcho          int64               `json:"received_echo"`
-	ExpectedGroup       int64               `json:"expected_group"`
+	ExpectedGroup         int64               `json:"expected_group"`
-	ReceivedGroup       int64               `json:"received_group"`
+	ReceivedGroup         int64               `json:"received_group"`
-	RealtimeLost        int64               `json:"realtime_lost"`
+	RealtimeLost          int64               `json:"realtime_lost"`
-	RealtimeLostStored  int64               `json:"realtime_lost_but_persisted"`
+	RealtimeLostStored    int64               `json:"realtime_lost_but_persisted"`
-	Duplicates          int64               `json:"duplicates"`
+	Duplicates            int64               `json:"duplicates"`
-	OutOfOrder          int64               `json:"out_of_order"`
+	OutOfOrder            int64               `json:"out_of_order"`
-	OrderingViolations  []OrderingViolation `json:"ordering_violations,omitempty"`
+	OrderingViolations    []OrderingViolation `json:"ordering_violations,omitempty"`
-	Disconnects         int64               `json:"disconnects"`
+	Disconnects           int64               `json:"disconnects"`
-	UnexpectedMessages  int64               `json:"unexpected_messages"`
+	UnexpectedMessages    int64               `json:"unexpected_messages"`
-	HistoryChecked      int64               `json:"history_checked"`
+	HistoryChecked        int64               `json:"history_checked"`
-	HistoryFound        int64               `json:"history_found"`
+	HistoryFound          int64               `json:"history_found"`
-	PersistenceLost     int64               `json:"persistence_lost"`
+	PersistenceLost       int64               `json:"persistence_lost"`
-	AckLatency          HistogramSnapshot   `json:"ack_latency"`
+	AckLatency            HistogramSnapshot   `json:"ack_latency"`
-	DeliveryLatency     HistogramSnapshot   `json:"delivery_latency"`
+	DeliveryLatency       HistogramSnapshot   `json:"delivery_latency"`
-	Errors              []string            `json:"errors,omitempty"`
+	DeliveryLatencyLocal  HistogramSnapshot   `json:"delivery_latency_local"`
-	Missing             []MissingDelivery   `json:"missing,omitempty"`
+	DeliveryLatencyCross  HistogramSnapshot   `json:"delivery_latency_cross_machine"`
-	PersistenceMissing  []string            `json:"persistence_missing,omitempty"`
+	CrossClockSkewSamples int64               `json:"cross_machine_negative_latency_samples"`
-	AcceptedLocal       []AcceptedMessage   `json:"accepted_local,omitempty"`
+	Errors                []string            `json:"errors,omitempty"`
+	Missing               []MissingDelivery   `json:"missing,omitempty"`
+	PersistenceMissing    []string            `json:"persistence_missing,omitempty"`
+	AcceptedLocal         []AcceptedMessage   `json:"accepted_local,omitempty"`
 }
 
 type OrderingViolation struct {

loadtest/im-pressure/recorder.go

@@ -17,22 +17,25 @@ type observation struct {
 }
 
 type deliveryRecorder struct {
-	machine     string
+	machine      string
-	localUsers  int
+	localUsers   int
-	latency     *DurationHistogram
+	latency      *DurationHistogram
-	disconnects atomic.Int64
+	localLatency *DurationHistogram
-	received    atomic.Int64
+	crossLatency *DurationHistogram
-	runID       atomic.Value
+	disconnects  atomic.Int64
-	historyMode atomic.Bool
+	received     atomic.Int64
+	runID        atomic.Value
+	historyMode  atomic.Bool
 
-	mu         sync.Mutex
+	mu                    sync.Mutex
-	observed   map[string]*observation
+	observed              map[string]*observation
-	lastSeq    []map[string]int64
+	lastSeq               []map[string]int64
-	duplicates int64
+	duplicates            int64
-	outOfOrder int64
+	outOfOrder            int64
-	ordering   []OrderingViolation
+	ordering              []OrderingViolation
-	malformed  int64
+	malformed             int64
-	history    map[string]bool
+	crossClockSkewSamples int64
+	history               map[string]bool
 }
 
 func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorder {
@@ -42,6 +45,7 @@ func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorde
 	}
 	recorder := &deliveryRecorder{
 		machine: machine, localUsers: localUsers, latency: NewDurationHistogram(),
+		localLatency: NewDurationHistogram(), crossLatency: NewDurationHistogram(),
 		observed: make(map[string]*observation), lastSeq: lastSeq, history: make(map[string]bool),
 	}
 	recorder.runID.Store(runID)
@@ -57,8 +61,9 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) {
 		return
 	}
 	var content struct {
-		RunID  string `json:"run_id"`
+		RunID         string `json:"run_id"`
-		SentAt int64  `json:"sent_at_unix_nano"`
+		SenderMachine string `json:"sender_machine"`
+		SentAt        int64  `json:"sent_at_unix_nano"`
 	}
 	if json.Unmarshal(data.Content, &content) != nil || content.RunID != runID {
 		r.mu.Lock()
@@ -108,10 +113,31 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) {
 		r.lastSeq[index][data.Topic] = data.Seq
 	}
 	if content.SentAt > 0 {
-		r.latency.Observe(time.Since(time.Unix(0, content.SentAt)))
+		delay := time.Since(time.Unix(0, content.SentAt))
+		r.latency.Observe(delay)
+		senderMachine := content.SenderMachine
+		if senderMachine == "" {
+			senderMachine = senderMachineFromClientMid(runID, mid)
+		}
+		if senderMachine == r.machine {
+			r.localLatency.Observe(delay)
+		} else if delay < 0 {
+			r.crossClockSkewSamples++
+		} else {
+			r.crossLatency.Observe(delay)
+		}
 	}
 }
 
+func senderMachineFromClientMid(runID, mid string) string {
+	prefix := strings.TrimPrefix(mid, runID+"-")
+	parts := strings.SplitN(prefix, "-", 3)
+	if len(parts) < 2 {
+		return ""
+	}
+	return parts[1]
+}
+
 func (r *deliveryRecorder) metricCounts() (received, duplicates, outOfOrder, disconnects int64) {
 	r.mu.Lock()
 	duplicates = r.duplicates
@@ -134,6 +160,8 @@ func (r *deliveryRecorder) onDisconnect(_ AccountRecord, _ error) {
 
 func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage) AgentSummary {
 	base.DeliveryLatency = r.latency.Snapshot()
+	base.DeliveryLatencyLocal = r.localLatency.Snapshot()
+	base.DeliveryLatencyCross = r.crossLatency.Snapshot()
 	base.Disconnects = r.disconnects.Load()
 	runID, _ := r.runID.Load().(string)
 	accepted := make(map[string]AcceptedMessage, len(ledger))
@@ -143,6 +171,7 @@ func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage
 	defer r.mu.Unlock()
 	base.Duplicates = r.duplicates
 	base.OutOfOrder = r.outOfOrder
+	base.CrossClockSkewSamples = r.crossClockSkewSamples
 	base.OrderingViolations = append([]OrderingViolation(nil), r.ordering...)
 	if r.malformed > 0 {
 		base.Errors = append(base.Errors, fmt.Sprintf("malformed run messages=%d", r.malformed))

loadtest/im-pressure/recorder_test.go

@@ -66,6 +66,23 @@ func TestDeliveryRecorderCapturesOrderingViolationDetails(t *testing.T) {
 	}
 }
 
+func TestDeliveryRecorderSeparatesLocalAndCrossMachineLatency(t *testing.T) {
+	recorder := newDeliveryRecorder("run1", "a", 1)
+	local := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-01", Topic: "p2p-one", Seq: 1}
+	cross := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-b-0001-01", Topic: "p2p-one", Seq: 2}
+	recorder.onData(AccountRecord{Index: 1}, testData(local, time.Now().Add(-10*time.Millisecond)))
+	recorder.onData(AccountRecord{Index: 1}, testData(cross, time.Now().Add(time.Second)))
+
+	summary := recorder.summarize(AgentSummary{}, nil)
+	if summary.DeliveryLatencyLocal.Count != 1 || summary.DeliveryLatencyCross.Count != 0 {
+		t.Fatalf("unexpected split latency counts: local=%d cross=%d",
+			summary.DeliveryLatencyLocal.Count, summary.DeliveryLatencyCross.Count)
+	}
+	if summary.CrossClockSkewSamples != 1 {
+		t.Fatalf("cross clock skew samples=%d", summary.CrossClockSkewSamples)
+	}
+}
+
 func testData(message AcceptedMessage, sentAt time.Time) dataPacket {
 	content, _ := json.Marshal(map[string]any{"run_id": message.RunID, "sent_at_unix_nano": sentAt.UnixNano()})
 	return dataPacket{Topic: message.Topic, Seq: message.Seq,

loadtest/im-pressure/wsclient.go

@@ -245,7 +245,8 @@ func (c *WSClient) Publish(ctx context.Context, topic, clientMid, runID, scenari
 		"id": id, "topic": topic, "noecho": false,
 		"head": map[string]any{"mime": "application/json", "x-im-schema": 1, "x-im-type": "text", "x-im-client-mid": clientMid},
 		"content": map[string]any{"text": "loadtest", "run_id": runID, "scenario": scenario,
-			"sender_index": c.account.Index, "message_index": messageIndex, "sent_at_unix_nano": sentAt.UnixNano()},
+			"sender_machine": c.account.Machine, "sender_index": c.account.Index,
+			"message_index": messageIndex, "sent_at_unix_nano": sentAt.UnixNano()},
 	}})
 	if err != nil {
 		return AcceptedMessage{}, err

server/kafka_message_persist.go

@@ -53,7 +53,9 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 	msg.ClientMsgID = clientMsgID
 	fromUid := types.ParseUid(msg.From)
 	if clientMsgID != "" {
+		started := time.Now()
 		claim, claimed, runtimeReady, claimErr := store.ClaimMessageUnique(tenantID, msg.Topic, fromUid, clientMsgID)
+		immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "redis_idempotency", started, claimErr)
 		if claimErr != nil {
 			return nil, claimErr
 		}
@@ -69,12 +71,16 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 		}
 	}
 	candidateMonth := msg.CreatedAt.UTC().Format("200601")
-	if seq, shardMonth, runtimeReady, seqErr := store.ReserveMessageSeq(tenantID, msg.Topic, msg.SeqId-1, candidateMonth); seqErr != nil {
+	seqStarted := time.Now()
+	seq, shardMonth, runtimeReady, seqErr := store.ReserveMessageSeq(tenantID, msg.Topic, msg.SeqId-1, candidateMonth)
+	immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "redis_sequence", seqStarted, seqErr)
+	if seqErr != nil {
 		if clientMsgID != "" {
 			_ = store.MarkMessageUniqueFailed(tenantID, msg.Topic, fromUid, clientMsgID)
 		}
 		return nil, seqErr
-	} else if runtimeReady {
+	}
+	if runtimeReady {
 		msg.SeqId = seq
 		if isGroupMessage {
 			msg.ShardMonth = shardMonth
@@ -83,7 +89,9 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 	if msg.ShardMonth == "" && isGroupMessage {
 		msg.ShardMonth = candidateMonth
 	}
+	kafkaStarted := time.Now()
 	if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil {
+		immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "kafka_produce", kafkaStarted, err)
 		if clientMsgID != "" {
 			if markErr := store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); markErr != nil {
 				logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, markErr)
@@ -92,6 +100,7 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 		logs.Warn.Printf("kafka message persist: publish state unknown topic=%s seq=%d client_mid=%s err=%v", msg.Topic, msg.SeqId, clientMsgID, err)
 		return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, err
 	}
+	immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "kafka_produce", kafkaStarted, nil)
 	if clientMsgID != "" {
 		if err = store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); err != nil {
 			logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, err)
@@ -105,11 +114,19 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 	return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, nil
 }
 
-func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error {
+func messageKind(topic string) string {
-	kind := "p2p"
+	switch types.GetTopicCat(topic) {
-	if types.GetTopicCat(msg.Topic) == types.TopicCatGrp {
+	case types.TopicCatGrp:
-		kind = "group"
+		return "group"
+	case types.TopicCatP2P:
+		return "p2p"
+	default:
+		return "other"
 	}
+}
+
+func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error {
+	kind := messageKind(msg.Topic)
 	payload, err := json.Marshal(kafkaMessagePersistPayload{
 		Message:        *msg,
 		AttachmentURLs: attachmentURLs,

server/main.go

@@ -182,6 +182,8 @@ var globals struct {
 	tlsRedirectHTTP string
 	// Maximum message size allowed from peer.
 	maxMessageSize int64
+	// Log complete client packets. Disabled by default because it is expensive on the publish hot path.
+	logClientPackets bool
 	// Maximum number of group topic subscribers.
 	maxSubscriberCount int
 	// Maximum number of indexable tags.
@@ -335,6 +337,8 @@ type configType struct {
 	// Maximum message size allowed from client. Intended to prevent malicious client from sending
 	// very large files inband (does not affect out of band uploads).
 	MaxMessageSize int `json:"max_message_size"`
+	// Log every decoded client packet. Intended only for short diagnostic sessions.
+	LogClientPackets bool `json:"log_client_packets"`
 	// Maximum number of group topic subscribers.
 	MaxSubscriberCount int `json:"max_subscriber_count"`
 	// Masked tags namespaces: tags immutable on User (mask), mutable on Topic only within the mask.
@@ -670,6 +674,7 @@ func main() {
 	if globals.maxMessageSize <= 0 {
 		globals.maxMessageSize = defaultMaxMessageSize
 	}
+	globals.logClientPackets = config.LogClientPackets
 	// Maximum number of group topic subscribers
 	globals.maxSubscriberCount = config.MaxSubscriberCount
 	if globals.maxSubscriberCount <= 1 {

server/meetingim.conf

@@ -10,6 +10,7 @@
 
   "api_key_salt": "T713/rYYgW7g4m3vG6zGRh7+FM1t0T8j13koXScOAj4=",
   "max_message_size": 131072,
+  "log_client_packets": false,
   "max_subscriber_count": 128,
   "max_tag_count": 16,
   "permanent_accounts": false,

server/metrics/metrics.go

@@ -107,6 +107,11 @@ var (
 		Help:    "Kafka message persistence duration by message kind and result.",
 		Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
 	}, []string{"kind", "result"})
+	MessagePublishStageDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_message_publish_stage_duration_seconds",
+		Help:    "Message publish hot-path duration by message kind, stage, and result.",
+		Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+	}, []string{"kind", "stage", "result"})
 	KafkaMessagePersistTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "im_kafka_message_persist_total",
 		Help: "Kafka message persistence outcomes by message kind and result.",
@@ -130,7 +135,8 @@ func init() {
 		HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration,
 		RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength,
 		ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration,
-		ClusterRouteTotal, ClusterProxyResponseTotal, MessagePersistDuration, KafkaMessagePersistTotal)
+		ClusterRouteTotal, ClusterProxyResponseTotal, MessagePersistDuration, MessagePublishStageDuration,
+		KafkaMessagePersistTotal)
 	Registry.MustRegister(KafkaConsumerLag, KafkaBacklogOldestAge, KafkaDeadLettersTotal)
 }
 
@@ -149,6 +155,15 @@ func ObserveDB(driverName, operation string, started time.Time, err error) {
 	DBQueryDuration.WithLabelValues(driverName, operation, status).Observe(time.Since(started).Seconds())
 }
 
+func ObserveMessagePublishStage(kind, stage string, started time.Time, err error) {
+	status := "ok"
+	if err != nil {
+		status = "error"
+	}
+	MessagePublishStageDuration.WithLabelValues(normalizeLabel(kind), normalizeLabel(stage), status).
+		Observe(time.Since(started).Seconds())
+}
+
 func IncRealtimeFanout(stage, kind, targetType, status string) {
 	RealtimeFanoutEventsTotal.WithLabelValues(normalizeLabel(stage), normalizeLabel(kind),
 		normalizeLabel(targetType), normalizeLabel(status)).Inc()

server/session.go

@@ -519,13 +519,15 @@ func (s *Session) dispatchRaw(raw []byte) {
 		s.queueOut(ErrMalformed("", "", now))
 		return
 	}
-	toLog := redactClientMessageForLog(raw)
+	if globals.logClientPackets {
-	truncated := ""
+		toLog := redactClientMessageForLog(raw)
-	if len(toLog) > 512 {
+		truncated := ""
-		toLog = toLog[:512]
+		if len(toLog) > 512 {
-		truncated = "<...>"
+			toLog = toLog[:512]
+			truncated = "<...>"
+		}
+		logs.Info.Printf("in: '%s%s' sid='%s' uid='%s'", toLog, truncated, s.sid, s.uid)
 	}
-	logs.Info.Printf("in: '%s%s' sid='%s' uid='%s'", toLog, truncated, s.sid, s.uid)
 
 	s.dispatch(&msg)
 }

server/store/store.go

@@ -3373,7 +3373,7 @@ func relationshipContext(tenantAdp adapter.TenantBusinessAdapter, tenantID types
 	if friendship, err := tenantAdp.TenantFriendshipGet(tenantID, userID, peerID); err != nil {
 		return nil, err
 	} else if friendship != nil && friendship.Status == types.RelationshipFriend {
-		cacheFriendOneWay(tenantID, userID, peerID)
+		cacheFriendPair(tenantID, userID, peerID)
 		ctx.Relationship = types.RelationshipFriend
 		ctx.CanSend = true
 		ctx.Remark = friendship.Remark

server/topic.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/tinode/chat/server/auth"
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/moderation"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
@@ -1362,7 +1363,9 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid,
 		store.Relations != nil && store.Store != nil && store.Store.IsOpen() {
 		peer := t.p2pOtherUser(asUid)
 		if !peer.IsZero() {
+			started := time.Now()
 			canSend, _, err := store.Relations.CanSendP2P(t.tenantID, asUid, peer)
+			immetrics.ObserveMessagePublishStage("p2p", "relationship_check", started, err)
 			if err != nil {
 				logs.Warn.Printf("topic[%s]: failed to check p2p relationship: %v", t.name, err)
 				msg.sess.queueOut(ErrUnknown(msg.Id, t.original(asUid), msg.Timestamp))
@@ -1375,6 +1378,7 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid,
 		}
 	}
 
+	persistStarted := time.Now()
 	saveResult, err := saveMessageMaybeKafka(context.Background(), t.tenantID,
 		&types.Message{
 			ObjHeader: types.ObjHeader{CreatedAt: msg.Timestamp},
@@ -1385,6 +1389,13 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid,
 			Head:      head,
 			Content:   content,
 		}, attachments, (pud.modeGiven & pud.modeWant).IsReader())
+	kind := "other"
+	if t.cat == types.TopicCatP2P {
+		kind = "p2p"
+	} else if t.cat == types.TopicCatGrp {
+		kind = "group"
+	}
+	immetrics.ObserveMessagePublishStage(kind, "save", persistStarted, err)
 	if err != nil {
 		logs.Warn.Printf("topic[%s]: failed to save message: %v", t.name, err)
 		msg.sess.queueOut(ErrUnknown(msg.Id, t.original(asUid), msg.Timestamp))