test: report exact ACK latency percentiles

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

@@ -349,3 +349,28 @@ Run ID：`p2p-200-20260927-04`。ArgoCD Application `im-server-test` 的真实
 Kafka `im-server.message_persist` lag 在发送后峰值为 2946，并在发送结束后约 113 秒归零，超过 60 秒目标。历史 API 和 MySQL 最终均确认 4000 条完整落库，说明这是落库延迟而不是消息丢失。进入 500 对测试前，应优先检查 message persist consumer 的批量写入、worker/partition 并行度和 MySQL 写入耗时。
 
 连接集中关闭期间，三个 Pod 分别记录 10、14、22 次 `failed to close login audit ... Deadlock found`，共 46 次。该问题未影响消息表，但会导致登录审计结束状态不完整。登录审计关闭更新应增加 MySQL 1213/1205 有界重试，并检查更新条件和索引以缩小锁范围。
+
+## 14. 持久化并行与审计死锁修复后的第五轮复测
+
+Run ID：`p2p-200-20260927-05`。本轮已将 `im.message.persist` 和 `im.group.message.persist` 固化为 3 分区、3 副本；同一 Topic 继续作为 Kafka key，保证会话内顺序。服务端同时按 batch 内最大 Seq 合并 Topic 和发送者已读位点更新，并为登录审计关闭增加复合索引和 MySQL 死锁重试。
+
+| 指标 | 结果 |
+|---|---:|
+| WS Ready | 400/400 |
+| Publish 尝试 / ACK | 4000 / 4000 |
+| ACK 后应到达 / 实际到达 | 4000 / 4000 |
+| 丢失 / 重复 / 乱序 / 断连 | 0 / 0 / 0 / 0 |
+| 历史查询 / MySQL 精确匹配 | 4000/4000 / 4000/4000 |
+| MySQL 缺失 / 错配 / 重复 / 非 ACK 落库 | 0 / 0 / 0 / 0 |
+| 总体精确 ACK P50 / P95 / P99 / Max | 375.314 / 547.516 / 670.317 / 2845.347 ms |
+| A 机精确 ACK P95 | 408.155 ms |
+| B 机精确 ACK P95 | 594.533 ms |
+| Pod 0/1/2 Kafka 峰值 lag | 676 / 251 / 143 |
+| 最慢 Kafka 分区归零时间 | 发送结束后约 52.5s |
+| Cluster 队列峰值 / CPU throttled 增量 | 0 / 三 Pod 均为 0 |
+| 登录审计关闭死锁 | 三 Pod 均为 0 |
+| Kafka batch failure / Pod 重启 | 0 / 0 |
+
+结论：三分区已由三个 Pod 实际并行消费，单 Pod 峰值 lag 从第四轮的 2946 降至 676，最慢消费者在 60 秒门槛内归零。消息正确性、最终落库、会话内顺序、CPU 限流和登录审计死锁均通过。
+
+报告仍因 ACK P95 门禁显示失败。直方图只能给出 1000ms 上界，按 4000 条 ACK 原始时间戳和 nearest-rank 口径计算的实际 P95 为 547.516ms。A 机已经低于 500ms，而 B 机为 594.533ms，且 B 机存在 2.845s 最大长尾；两台机器使用相同公共入口和相同服务端版本，因此继续修改服务端前，应先采集 B 机到 `api.118im.com` 的 DNS、TCP、TLS、WebSocket RTT 与时钟偏差，并使用定向 Pod 入口做 A/B 对照，区分公网链路长尾和服务端处理耗时。

loadtest/im-pressure/merge.go

@@ -87,7 +87,7 @@ func runMerge(aPath, bPath, output string) error {
 	if report.PublishAttempted > 0 {
 		report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted)
 	}
-	report.AckP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.AckLatency })
+	report.AckP95MS = mergedAckP95(report.Agents)
 	report.DeliveryP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatency })
 	if report.ExpectedDeliveries > 0 {
 		report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries)
@@ -102,7 +102,7 @@ func runMerge(aPath, bPath, output string) error {
 			fmt.Sprintf("publish success rate=%.6f expected>=0.999900", report.PublishSuccessRate))
 	}
 	if report.AckP95MS > 500 {
-		report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ack p95=%.0fms expected<=500ms", report.AckP95MS))
+		report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ack p95=%.3fms expected<=500ms", report.AckP95MS))
 	}
 	if report.DeliveryP95MS > 1000 {
 		report.FailureReasons = append(report.FailureReasons,
@@ -121,6 +121,26 @@ func runMerge(aPath, bPath, output string) error {
 	return writeJSONFile(output, report)
 }
 
+func mergedAckP95(agents []AgentSummary) float64 {
+	var expected int64
+	latencies := make([]float64, 0)
+	for _, agent := range agents {
+		expected += agent.AckLatency.Count
+		for _, message := range agent.AcceptedLocal {
+			if message.SentAt.IsZero() || message.AckAt.Before(message.SentAt) {
+				continue
+			}
+			latencies = append(latencies, float64(message.AckAt.Sub(message.SentAt))/float64(time.Millisecond))
+		}
+	}
+	if expected == 0 || int64(len(latencies)) != expected {
+		return mergedHistogramP95(agents, func(summary AgentSummary) HistogramSnapshot { return summary.AckLatency })
+	}
+	sort.Float64s(latencies)
+	index := (len(latencies)*95 + 99) / 100
+	return latencies[index-1]
+}
+
 func mergedHistogramP95(agents []AgentSummary, selectHistogram func(AgentSummary) HistogramSnapshot) float64 {
 	var count int64
 	buckets := make([]int64, len(latencyBounds)+1)

loadtest/im-pressure/merge_test.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 )
 
 func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) {
@@ -53,6 +54,32 @@ func TestMergedHistogramP95(t *testing.T) {
 	}
 }
 
+func TestMergedAckP95UsesExactAcceptedTimestamps(t *testing.T) {
+	base := time.Date(2026, 9, 27, 8, 0, 0, 0, time.UTC)
+	agents := []AgentSummary{{
+		AckLatency: HistogramSnapshot{Count: 4, MaxMS: 900, Buckets: map[string]int64{"1s": 4}},
+		AcceptedLocal: []AcceptedMessage{
+			{SentAt: base, AckAt: base.Add(100 * time.Millisecond)},
+			{SentAt: base, AckAt: base.Add(200 * time.Millisecond)},
+			{SentAt: base, AckAt: base.Add(300 * time.Millisecond)},
+			{SentAt: base, AckAt: base.Add(400 * time.Millisecond)},
+		},
+	}}
+	if got := mergedAckP95(agents); got != 400 {
+		t.Fatalf("exact p95=%v want=400", got)
+	}
+}
+
+func TestMergedAckP95FallsBackWhenRawSamplesAreIncomplete(t *testing.T) {
+	agents := []AgentSummary{{
+		AckLatency:    HistogramSnapshot{Count: 2, MaxMS: 900, Buckets: map[string]int64{"1s": 2}},
+		AcceptedLocal: []AcceptedMessage{{SentAt: time.Now(), AckAt: time.Now().Add(100 * time.Millisecond)}},
+	}}
+	if got := mergedAckP95(agents); got != 1000 {
+		t.Fatalf("fallback p95=%v want=1000", got)
+	}
+}
+
 func writeSummaryForTest(t *testing.T, path string, summary AgentSummary) {
 	t.Helper()
 	raw, err := json.Marshal(summary)