test: harden group load verification

deploy/argocd/im-server-test/README.md

@@ -9,12 +9,25 @@ patches:
   - path: statefulset-performance-patch.yaml
 ```
 
+Also include `backend-debug-metrics-rbac.yaml` as a resource in the test
+overlay. It grants the existing `argocd/backend-debug` service account
+namespaced read access to Pod metrics so the load-test collector can run
+`kubectl top pods -n im-server-system`:
+
+```yaml
+resources:
+  - backend-debug-metrics-rbac.yaml
+```
+
 After ArgoCD sync, verify the effective configuration:
 
 ```bash
 kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].resources}'
 kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].livenessProbe.httpGet.path}'
 kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].env}'
+kubectl auth can-i list pods.metrics.k8s.io \
+  --as=system:serviceaccount:argocd:backend-debug \
+  -n im-server-system
 ```
 
 Expected values are requests `1 CPU / 512Mi`, limits `2 CPU / 1Gi`, and

deploy/argocd/im-server-test/backend-debug-metrics-rbac.yaml

@@ -0,0 +1,27 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: backend-debug-metrics-reader
+  namespace: im-server-system
+rules:
+  - apiGroups:
+      - metrics.k8s.io
+    resources:
+      - pods
+    verbs:
+      - get
+      - list
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: backend-debug-metrics-reader
+  namespace: im-server-system
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: backend-debug-metrics-reader
+subjects:
+  - kind: ServiceAccount
+    name: backend-debug
+    namespace: argocd

docs/test/IM-group2k-r4-群聊复测分析-20260928.md

@@ -0,0 +1,134 @@
+# IM 2000 人群聊 r4 复测分析
+
+日期：2026-09-28
+
+Run ID：`group2k-r4-20260928-01`
+
+场景：机器 A/B 各 1000 个 WebSocket、各发送 1000 条，计划共发送 2000 条群消息。
+
+## 1. 结论
+
+本轮不能按合并报告中的 `14.39%` 实时丢失率直接判定服务端群消息丢失。两台压测机结果存在明显单边故障：
+
+- 机器 B 的 1000 个连接全部稳定，1000 次发送全部收到 ACK，1497000 次 ACK 台账内投递全部到达，断线和实时缺失均为 0。
+- 机器 A 出现 654 次断线、314 次 socket 错误和 189 次 ACK 超时。A 的 TCP 源地址错误中包含 `broken pipe`、`websocket closed`，是本轮 430832 次实时缺失的主要来源。
+- 服务端实际接受并持久化 1601 条消息，其中客户端收到 ACK 1497 条，另外 104 条属于“客户端未收到 ACK，但服务端已接受”。B 恰好多收到 104000 次消息，即 104 条乘以 B 的 1000 个连接，账目完全吻合。
+- ACK 台账内 1497 条消息历史回查全部找到，`persistence_lost=0`；Kafka 群消息消费 lag 在采样点均为 0。
+- 上一轮的 O(群成员数) Redis fanout 放大已经消除：1601 条服务端消息只产生 1601 次 `group_unread/users` publish，每个远端 Pod 各 receive 1601 次，不再产生约 320 万次逐用户 Redis publish。
+- 实时发送与 60 秒收尾阶段没有新增 `topic_server` 或 `cluster_topic_proxy` 满队列。满队列发生在约 5 分钟后的并发历史回查阶段，属于另一项需要优化的问题，不能作为本轮实时丢失的原因。
+
+因此，本轮结论为：**批量群未读 fanout 修复通过；B 侧实时投递通过；数据库持久化通过；A 侧网络稳定性失败；历史回查并发会打满 Topic/Proxy 队列。整体尚不能完成 2000 人群聊验收。**
+
+## 2. 客户端结果
+
+| 指标 | A | B | 合计 |
+|---|---:|---:|---:|
+| READY 连接 | 1000 | 1000 | 2000 |
+| 发送尝试 | 1000 | 1000 | 2000 |
+| 收到 2xx ACK | 497 | 1000 | 1497 |
+| 明确拒绝 | 0 | 0 | 0 |
+| ACK 超时 | 189 | 0 | 189 |
+| socket 结果未知 | 314 | 0 | 314 |
+| 断线 | 654 | 0 | 654 |
+| ACK 台账内应收 | 1497000 | 1497000 | 2994000 |
+| ACK 台账内实收 | 1066168 | 1497000 | 2563168 |
+| ACK 台账内实时缺失 | 430832 | 0 | 430832 |
+| ACK 台账外额外投递 | 68668 | 104000 | 172668 |
+| 重复 / 乱序 | 0 / 0 | 0 / 0 | 0 / 0 |
+| 历史检查 / 找到 | 1497 / 1497 | 0 / 0 | 1497 / 1497 |
+
+合并报告按 ACK 台账计算出的到达率为 `85.61%`，但这只是 A/B 网络条件混合后的观测值，不代表服务端总体投递率。B 单机的 ACK 台账内到达率是 `100%`，A 单机是 `71.22%`。
+
+A 端记录了 160074 个跨机器负延迟样本，说明两机时钟没有达到单向延迟统计要求。本轮不使用跨机器 P95/P99 判断性能，只使用 ACK、到达计数、断线、历史回查和服务端阶段指标判断正确性。
+
+## 3. 未收到 ACK 的 503 条不是服务端拒绝
+
+本轮没有任何明确的 HTTP/Tinode `503` 响应。报告字段 `publish_rejected=503` 实际由以下结果未知项组成：
+
+- `ctrl_timeout=189`
+- `socket=314`
+
+服务端持久化计数增加 1601，而 ACK 只有 1497，说明至少 104 条结果未知消息已被服务端成功接受。B 的 `unexpected_messages=104000` 与这 104 条消息对 1000 个 B 连接的广播数量完全一致。
+
+后续报告应把字段继续显示为 `publish_unknown`，不能把它展示为 `rejected`，否则容易误判服务端主动拒绝。
+
+## 4. 服务端与中间件
+
+### 4.1 Kafka 与数据库落库
+
+- `im_kafka_message_persist_total{kind="group",result="batch_ok"}` 增加 1601。
+- `im.group.message.persist` consumer lag 在采样点均为 0。
+- 1497 条 ACK 台账消息通过历史接口全部找到，持久化缺失为 0。
+- 服务端群消息 `save` 总耗时增加约 26.27s，平均约 16.4ms/条。
+- Kafka produce 总耗时增加约 19.15s，平均约 12.0ms/条。
+
+这证明已接受消息能够及时进入 Kafka 并完成落库。104 条结果未知消息也计入 1601 条 batch success，但当前历史校验只以 ACK 台账为准，下一版应对“已观测但未 ACK”的消息也做 message ID 回查。
+
+### 4.2 Redis fanout 修复验证
+
+1601 条服务端接受消息对应：
+
+- master Pod `group_unread/users publish ok` 增加 1601。
+- 两个远端 Pod 的 `group_unread/users receive ok` 各增加 1601。
+- 普通 `message/user` 在原生 Cluster 开启时继续使用 `skip_native_cluster`，没有恢复逐成员 Redis 发布。
+
+这符合“一条群消息一次 Redis publish、每个 Pod 一次本地扫描”的设计，上一轮每条消息约 2000 次 Redis publish 的放大已经消失。
+
+### 4.3 队列满发生在历史回查阶段
+
+按时间序列拆分：
+
+- `16:25:05Z` 开始发送。
+- `16:28:36Z` 最后一台压测机结束发送，之后进入收尾和历史等待。
+- 截至 `16:34:01Z`，`topic_server/full` 仍停留在测试前基线 1349，`cluster_topic_proxy/full` 仍为 0。
+- `16:34:25Z` 起，历史回查并发开始使 `topic_server/full` 增加。
+- 最终历史阶段新增 `topic_server/full=1355`、`cluster_topic_proxy/full=55`。
+
+因此这些队列丢弃不是实时发送阶段 430832 次缺失的原因。历史回查一次并发拉取大量消息，经 Topic/Proxy 回送时会形成独立洪峰，应限制每机历史校验并发或分页节流，并单独验收历史吞吐。
+
+## 5. A 机器故障判断
+
+A 与 B 使用相同代码、相同公共入口和相同发送模型，但只有 A 发生 654 次断线。A 的错误包含：
+
+- `websocket closed`
+- `write: broken pipe`
+- `ctrl timeout`
+- 到多个公网入口 IP 的 TCP 读写超时
+
+这种强烈单边性不符合服务端全局过载的表现。当前最可能的故障域依次是 A 本机网络/NAT/安全软件、A 到公网 LB 的链路、A 的文件描述符或本地调度压力。由于 Kubernetes Metrics API 被 RBAC 拒绝，本轮没有取得 Pod CPU/内存数据，不能用资源指标进一步排除 Pod 饱和。
+
+## 6. 下一轮执行顺序
+
+1. 先做 A/B 角色互换：B 仍发 1000 条，A 暂时只保持 1000 个连接不发送。若 A 仍大量断线，可确认与服务端发送负载无关。
+2. 在 A 机同步采集 `ping`、TCP retransmit、连接跟踪、文件描述符、CPU、内存和进程调度数据；压测前检查 `ulimit -n`。
+3. 将历史回查改为分页和限并发，或先关闭自动历史回查，仅验证实时阶段；实时阶段结束后再启动独立历史校验。
+4. 修正报告字段，把超时/socket 明确归类为 `publish_unknown`，并把 104 条 unacked-observed 消息纳入历史校验。
+5. 补充 `backend-debug` 对 `metrics.k8s.io/pods` 的只读权限，否则下一轮仍无法采集 Pod CPU/内存。
+6. A 链路稳定后，使用相同 2000 连接、2000 条消息模型重新验收。门禁要求两侧断线为 0、明确拒绝为 0、实时到达率 100%、持久化丢失为 0，并分别报告实时阶段和历史阶段的队列指标。
+
+## 7. 结果文件
+
+本轮原始结果位于：
+
+```text
+loadtest/im-pressure/results/group2k-r4-20260928-01/
+  agent-a-summary.json
+  agent-b-summary.json
+  coordinator.json
+  report.json
+  pre-metrics.log
+  pod-metrics.log
+  post-metrics.log
+```
+
+## 8. 已实施改造
+
+本分析完成后，压测工具已实施以下改造：
+
+1. agent 分开输出 `publish_rejected` 和 `publish_outcome_unknown`；前者只包含服务端明确拒绝，后者包含 ACK 超时和 socket 失败。
+2. A/B 在 drain 后向 coordinator 提交“实时已观测但未 ACK”的消息台账，coordinator 去重合并后将其加入历史落库核验。
+3. 报告新增 `history_unacked_observed_checked` 和 `history_unacked_observed_found`，可直接确认结果未知消息是否最终落库。
+4. 历史回查默认调整为 1 个 worker、每页 50 条、全局每 100ms 最多发起一次查询，避免核验程序自身制造 Topic/Proxy 洪峰。
+5. 新增 `backend-debug-metrics-rbac.yaml`，为测试环境的 `argocd/backend-debug` ServiceAccount 提供 `metrics.k8s.io/pods` 的 namespace 只读权限。
+
+上述改造只解决统计口径、历史查询压力和资源指标权限。机器 A 的公网连接异常仍需通过下一轮主机与网络观测定位。

loadtest/im-pressure/MACHINE-B.md

@@ -148,7 +148,7 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}"
 3. 如果机器 A 尚未 READY，当前命令会等待，这是正常现象。
 4. 等待期间连接断开会自动重连并重新订阅；在 A/B 都实时保持 1000 条连接并完成最终屏障前，agent 不会退出或开始发压。
 5. coordinator 下发 `run_id`、场景和统一 UTC 开始时间。
-6. agent 完成发送和 WS 到达统计，等待 `history_settle`（默认 5 分钟）后，再以有界并发执行历史持久化回查；等待期间不要停止进程。
+6. agent 完成发送和 WS 到达统计后，先向协调器提交未 ACK 但已观测的消息台账；等待 `history_settle`（默认 5 分钟）后，以限速分页方式执行历史持久化回查。等待期间不要停止进程。
 7. agent 写出 summary、上报 coordinator，然后正常退出。
 
 机器 B 不需要传场景参数。`p2p`、`group` 或 `mixed` 由机器 A coordinator 决定。

loadtest/im-pressure/README.md

@@ -50,10 +50,10 @@
 - 发送成功以 `{ctrl code=2xx, params.seq>0}` 为准，形成两台机器共享的 ACK 台账。
 - 实时到达只统计同时匹配本轮 `run_id` 和 `x-im-client-mid` 的 `{data}`。
 - 群消息设置 `noecho=false`，每条服务端接受的群消息期望 2000 个连接各收到一次。
-- 实时 drain 结束后先等待 `history_settle`（默认 5 分钟），让 Kafka 在不受核验查询干扰的情况下自然排空，再通过 `{get what=data}` 按 ACK 的 Topic/Seq 回查历史。
+- 实时 drain 结束后，两机先汇总 ACK 台账和“已实时观测但未收到 ACK”的消息台账；等待 `history_settle`（默认 5 分钟）让 Kafka 自然排空，再通过 `{get what=data}` 按 Topic/Seq 回查两类消息。
 - 每轮建立 WS 后、正式计时前，以 32 个 worker 对本机 1000 个账号做好友关系、P2P Topic 和发布权限预检；任一异常会终止本轮。
-- 历史回查默认每机使用 12 个 worker、总超时 3 分钟，并在超时内持续重试缺失区间。历史读取并发能力应使用独立场景测试，不与落库吞吐测试混在一起。
+- 历史回查默认每机使用 1 个 worker、每页 50 条、查询间隔 100ms、总超时 3 分钟，并在超时内持续重试缺失区间。该限速用于避免正确性核验本身制造 Topic/Proxy 洪峰；历史读取吞吐应使用独立场景测试。
-- publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`，不受文本错误样本上限影响。
+- publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`。`publish_rejected` 只统计明确拒绝，超时和 socket 错误单独计入 `publish_outcome_unknown`。
 - `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/agent.go

@@ -28,7 +28,6 @@ type agentRun struct {
 	recorder       *deliveryRecorder
 	ack            *DurationHistogram
 	attempted      atomic.Int64
-	rejected       atomic.Int64
 	ready          atomic.Int64
 	publish403     atomic.Int64
 	publish502     atomic.Int64
@@ -207,11 +206,21 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo
 	if err = waitContext(ctx, drain); err != nil {
 		return err
 	}
+	observedLocal := run.recorder.unackedObservedMessages(ledger, start.Scenario)
+	if err = coord.submitObserved(ctx, machine, observedLocal); err != nil {
+		return err
+	}
+	observedLedger, err := coord.observed(ctx)
+	if err != nil {
+		return err
+	}
+	verificationLedger := mergeMessageLedgers(ledger, observedLedger)
 	base := AgentSummary{
 		RunID: start.RunID, Scenario: start.Scenario, Machine: machine, StartedAt: startedAt,
 		LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(), ExpectedConnections: cfg.UsersPerMachine,
 		ReadyConnections: run.clientCount(),
-		PublishAttempted: run.attempted.Load(), PublishAccepted: int64(len(local)), PublishRejected: run.rejected.Load(),
+		PublishAttempted: run.attempted.Load(), PublishAccepted: int64(len(local)),
+		PublishRejected: run.explicitRejected(), PublishOutcomeUnknown: run.outcomeUnknown(),
 		PublishErrors: run.publishErrorCounts(),
 		AckLatency:    run.ack.Snapshot(), AcceptedLocal: local,
 	}
@@ -224,9 +233,12 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo
 	if err = waitContext(ctx, historySettle); err != nil {
 		return err
 	}
-	checked, found, persistenceMissing := run.reconcileHistory(ctx, ledger)
+	history := run.reconcileHistory(ctx, verificationLedger, ledger)
-	summary.HistoryChecked = int64(checked)
+	summary.HistoryChecked = int64(history.checked)
-	summary.HistoryFound = int64(found)
+	summary.HistoryFound = int64(history.found)
+	summary.HistoryUnackedChecked = int64(history.unackedChecked)
+	summary.HistoryUnackedFound = int64(history.unackedFound)
+	persistenceMissing := history.missing
 	summary.PersistenceLost = int64(len(persistenceMissing))
 	summary.RealtimeLostStored = run.recorder.countRealtimeLostStored(ledger)
 	if len(persistenceMissing) > 10000 {
@@ -256,8 +268,20 @@ type historyQuery struct {
 	limit  int
 }
 
-func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessage) (int, int, []string) {
+type historyResult struct {
+	checked        int
+	found          int
+	unackedChecked int
+	unackedFound   int
+	missing        []string
+}
+
+func (r *agentRun) reconcileHistory(ctx context.Context, ledger, ackLedger []AcceptedMessage) historyResult {
 	expected := make(map[string]AcceptedMessage)
+	acked := make(map[string]bool, len(ackLedger))
+	for _, message := range ackLedger {
+		acked[message.ClientMid] = true
+	}
 	byTopic := make(map[string][]AcceptedMessage)
 	for _, message := range ledger {
 		if message.RunID != r.runID {
@@ -272,7 +296,7 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag
 		}
 	}
 	if len(expected) == 0 {
-		return 0, 0, nil
+		return historyResult{}
 	}
 	r.recorder.beginHistory()
 	queries := make([]historyQuery, 0, len(byTopic))
@@ -286,13 +310,14 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag
 			r.addError(fmt.Sprintf("history topic=%s has no verifier client", topicHash(topic)))
 			continue
 		}
-		for start := 0; start < len(messages); start += 200 {
+		pageSize := r.historyPageSize()
-			end := start + 200
+		for start := 0; start < len(messages); start += pageSize {
+			end := start + pageSize
 			if end > len(messages) {
 				end = len(messages)
 			}
 			queries = append(queries, historyQuery{client: client, topic: topic,
-				since: messages[start].Seq, before: messages[end-1].Seq + 1, limit: 256})
+				since: messages[start].Seq, before: messages[end-1].Seq + 1, limit: pageSize})
 		}
 	}
 	historyTimeout, _ := configDuration(r.cfg.HistoryTimeout)
@@ -315,7 +340,17 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag
 		missingIDs = append(missingIDs, message.ClientMid)
 	}
 	sort.Strings(missingIDs)
-	return len(expected), len(expected) - len(missing), missingIDs
+	result := historyResult{checked: len(expected), found: len(expected) - len(missing), missing: missingIDs}
+	for mid := range expected {
+		if acked[mid] {
+			continue
+		}
+		result.unackedChecked++
+		if r.recorder.historyFound(mid) {
+			result.unackedFound++
+		}
+	}
+	return result
 }
 
 func (r *agentRun) historyQueriesForMessages(messages []AcceptedMessage) []historyQuery {
@@ -333,13 +368,14 @@ func (r *agentRun) historyQueriesForMessages(messages []AcceptedMessage) []histo
 		if client == nil {
 			continue
 		}
-		for start := 0; start < len(topicMessages); start += 200 {
+		pageSize := r.historyPageSize()
-			end := start + 200
+		for start := 0; start < len(topicMessages); start += pageSize {
+			end := start + pageSize
 			if end > len(topicMessages) {
 				end = len(topicMessages)
 			}
 			queries = append(queries, historyQuery{client: client, topic: topic,
-				since: topicMessages[start].Seq, before: topicMessages[end-1].Seq + 1, limit: 256})
+				since: topicMessages[start].Seq, before: topicMessages[end-1].Seq + 1, limit: pageSize})
 		}
 	}
 	return queries
@@ -354,12 +390,23 @@ func (r *agentRun) issueHistoryQueries(ctx context.Context, queries []historyQue
 		return
 	}
 	jobs := make(chan historyQuery)
+	interval := 100 * time.Millisecond
+	if r.cfg != nil && r.cfg.HistoryInterval != "" {
+		interval = mustDuration(r.cfg.HistoryInterval)
+	}
+	limiter := time.NewTicker(interval)
+	defer limiter.Stop()
 	var wg sync.WaitGroup
 	for i := 0; i < workers; i++ {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
 			for query := range jobs {
+				select {
+				case <-limiter.C:
+				case <-ctx.Done():
+					return
+				}
 				if err := query.client.GetData(ctx, query.topic, query.since, query.before, query.limit); err != nil && ctx.Err() == nil {
 					r.addError(fmt.Sprintf("history get topic=%s range=%d-%d: %v", topicHash(query.topic), query.since, query.before, err))
 				}
@@ -379,6 +426,13 @@ func (r *agentRun) issueHistoryQueries(ctx context.Context, queries []historyQue
 	wg.Wait()
 }
 
+func (r *agentRun) historyPageSize() int {
+	if r.cfg != nil && r.cfg.HistoryPageSize > 0 {
+		return r.cfg.HistoryPageSize
+	}
+	return 50
+}
+
 func missingHistory(recorder *deliveryRecorder, expected map[string]AcceptedMessage) []AcceptedMessage {
 	missing := make([]AcceptedMessage, 0)
 	for mid, message := range expected {
@@ -389,6 +443,33 @@ func missingHistory(recorder *deliveryRecorder, expected map[string]AcceptedMess
 	return missing
 }
 
+func mergeMessageLedgers(accepted, observed []AcceptedMessage) []AcceptedMessage {
+	byID := make(map[string]AcceptedMessage, len(accepted)+len(observed))
+	for _, message := range observed {
+		if message.ClientMid == "" {
+			continue
+		}
+		current, exists := byID[message.ClientMid]
+		currentIsPeer := current.PeerMachine != "" && current.PeerMachine != current.SenderMachine
+		candidateIsPeer := message.PeerMachine != "" && message.PeerMachine != message.SenderMachine
+		if !exists || (!currentIsPeer && candidateIsPeer) {
+			byID[message.ClientMid] = message
+		}
+	}
+	// ACK-derived records carry authoritative sender and timing fields.
+	for _, message := range accepted {
+		if message.ClientMid != "" {
+			byID[message.ClientMid] = message
+		}
+	}
+	merged := make([]AcceptedMessage, 0, len(byID))
+	for _, message := range byID {
+		merged = append(merged, message)
+	}
+	sort.Slice(merged, func(i, j int) bool { return merged[i].ClientMid < merged[j].ClientMid })
+	return merged
+}
+
 func (r *agentRun) addError(message string) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -615,7 +696,6 @@ func (r *agentRun) publish(ctx context.Context, client *WSClient, topic, mid str
 	r.attempted.Add(1)
 	accepted, err := client.Publish(ctx, topic, mid, r.runID, r.scenario, messageIndex)
 	if err != nil {
-		r.rejected.Add(1)
 		r.classifyPublishError(err)
 		r.mu.Lock()
 		if len(r.errors) < 1000 {
@@ -654,6 +734,14 @@ func (r *agentRun) publishErrorCounts() map[string]int64 {
 	}
 }
 
+func (r *agentRun) explicitRejected() int64 {
+	return r.publish403.Load() + r.publish502.Load() + r.publish503.Load() + r.publishOther.Load()
+}
+
+func (r *agentRun) outcomeUnknown() int64 {
+	return r.publishTimeout.Load() + r.publishSocket.Load()
+}
+
 func (r *agentRun) addAccepted(message AcceptedMessage) {
 	r.mu.Lock()
 	r.accepted = append(r.accepted, message)

loadtest/im-pressure/agent_history_test.go

@@ -5,7 +5,7 @@ import "testing"
 func TestHistoryQueriesForMessagesGroupsRangesByTopic(t *testing.T) {
 	p2pClient := &WSClient{}
 	groupClient := &WSClient{}
-	run := &agentRun{clients: map[int]*WSClient{1: groupClient, 7: p2pClient}}
+	run := &agentRun{cfg: &Config{HistoryPageSize: 200}, clients: map[int]*WSClient{1: groupClient, 7: p2pClient}}
 	messages := make([]AcceptedMessage, 0, 207)
 	for seq := int64(205); seq >= 1; seq-- {
 		messages = append(messages, AcceptedMessage{Kind: "p2p", Topic: "p2p-a", PeerIndex: 7, Seq: seq})
@@ -25,7 +25,7 @@ func TestHistoryQueriesForMessagesGroupsRangesByTopic(t *testing.T) {
 		switch query.topic {
 		case "p2p-a":
 			p2pQueries++
-			if query.client != p2pClient || query.limit != 256 {
+			if query.client != p2pClient || query.limit != 200 {
 				t.Fatalf("unexpected p2p query: %#v", query)
 			}
 		case "grp-a":
@@ -39,3 +39,18 @@ func TestHistoryQueriesForMessagesGroupsRangesByTopic(t *testing.T) {
 		t.Fatalf("p2p queries=%d group queries=%d", p2pQueries, groupQueries)
 	}
 }
+
+func TestMergeMessageLedgersPrefersPeerObservationAndAck(t *testing.T) {
+	senderEcho := AcceptedMessage{ClientMid: "mid-1", SenderMachine: "a", PeerMachine: "a", Seq: 10}
+	peerDelivery := AcceptedMessage{ClientMid: "mid-1", SenderMachine: "a", PeerMachine: "b", PeerIndex: 7, Seq: 10}
+	merged := mergeMessageLedgers(nil, []AcceptedMessage{senderEcho, peerDelivery})
+	if len(merged) != 1 || merged[0].PeerMachine != "b" || merged[0].PeerIndex != 7 {
+		t.Fatalf("peer observation was not preferred: %+v", merged)
+	}
+	ack := AcceptedMessage{ClientMid: "mid-1", SenderMachine: "a", PeerMachine: "b", PeerIndex: 7, Seq: 10,
+		Scenario: "p2p"}
+	merged = mergeMessageLedgers([]AcceptedMessage{ack}, []AcceptedMessage{peerDelivery})
+	if len(merged) != 1 || merged[0].Scenario != "p2p" {
+		t.Fatalf("ACK ledger was not authoritative: %+v", merged)
+	}
+}

loadtest/im-pressure/agent_metrics.go

@@ -36,8 +36,11 @@ func startAgentMetrics(address string, run *agentRun) (func(), error) {
 			defer run.mu.Unlock()
 			return float64(len(run.accepted))
 		}),
-		prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_rejected_total", Help: "Publish commands rejected or timed out.", ConstLabels: labels}, func() float64 {
+		prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_rejected_total", Help: "Publish commands explicitly rejected by the server.", ConstLabels: labels}, func() float64 {
-			return float64(run.rejected.Load())
+			return float64(run.explicitRejected())
+		}),
+		prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_outcome_unknown_total", Help: "Publish commands with ACK timeout or socket failure and therefore unknown outcome.", ConstLabels: labels}, func() float64 {
+			return float64(run.outcomeUnknown())
 		}),
 		prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_403_total", Help: "Publish commands rejected with 403.", ConstLabels: labels}, func() float64 { return float64(run.publish403.Load()) }),
 		prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_502_total", Help: "Publish commands rejected with 502.", ConstLabels: labels}, func() float64 { return float64(run.publish502.Load()) }),

loadtest/im-pressure/config.example.json

@@ -17,7 +17,9 @@
   "ws_ping_interval": "20s",
   "preflight_workers": 32,
   "skip_p2p_preflight": false,
-  "history_workers": 12,
+  "history_workers": 1,
+  "history_page_size": 50,
+  "history_query_interval": "100ms",
   "history_settle": "5m",
   "history_timeout": "3m",
   "p2p_messages_per_user": 10,

loadtest/im-pressure/coordinator.go

@@ -38,6 +38,7 @@ type coordinatorState struct {
 	ready          map[string]AgentReady
 	armed          map[string]bool
 	ledgers        map[string][]AcceptedMessage
+	observed       map[string][]AcceptedMessage
 	summaries      map[string]AgentSummary
 	complete       chan struct{}
 	completeOnce   sync.Once
@@ -57,12 +58,14 @@ func runCoordinator(ctx context.Context, listen, runID, scenario, output string,
 		start: StartInfo{RunID: runID, Scenario: scenario}, startDelay: startDelay, commitDelay: 5 * time.Second,
 		expectedAgents: expectedAgents,
 		ready:          make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage),
-		armed: make(map[string]bool), summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
+		observed: make(map[string][]AcceptedMessage),
+		armed:    make(map[string]bool), summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/v1/ready", state.handleReady)
 	mux.HandleFunc("/v1/start", state.handleStart)
 	mux.HandleFunc("/v1/ledger", state.handleLedger)
+	mux.HandleFunc("/v1/observed", state.handleObserved)
 	mux.HandleFunc("/v1/summary", state.handleSummary)
 	mux.HandleFunc("/v1/status", state.handleStatus)
 	server := &http.Server{Addr: listen, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
@@ -91,6 +94,44 @@ func runCoordinator(ctx context.Context, listen, runID, scenario, output string,
 	return nil
 }
 
+func (s *coordinatorState) handleObserved(w http.ResponseWriter, r *http.Request) {
+	machine := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("machine")))
+	switch r.Method {
+	case http.MethodPost:
+		if machine != "a" && machine != "b" {
+			http.Error(w, "invalid machine", http.StatusBadRequest)
+			return
+		}
+		var messages []AcceptedMessage
+		if !decodeLimitedJSON(w, r, &messages) {
+			return
+		}
+		s.mu.Lock()
+		if s.observed == nil {
+			s.observed = make(map[string][]AcceptedMessage)
+		}
+		s.observed[machine] = messages
+		count := len(s.observed)
+		s.mu.Unlock()
+		writeJSON(w, http.StatusOK, map[string]int{"submitted_agents": count})
+	case http.MethodGet:
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		if len(s.observed) < s.expectedAgents {
+			writeJSON(w, http.StatusAccepted, map[string]int{"submitted_agents": len(s.observed), "expected_agents": s.expectedAgents})
+			return
+		}
+		var merged []AcceptedMessage
+		for _, messages := range s.observed {
+			merged = append(merged, messages...)
+		}
+		merged = mergeMessageLedgers(nil, merged)
+		writeJSON(w, http.StatusOK, merged)
+	default:
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+	}
+}
+
 func (s *coordinatorState) handleReady(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
@@ -223,7 +264,7 @@ func (s *coordinatorState) handleStatus(w http.ResponseWriter, _ *http.Request)
 	defer s.mu.Unlock()
 	writeJSON(w, http.StatusOK, map[string]any{
 		"start": s.start, "ready": s.ready, "armed": s.armed,
-		"ledger_agents": len(s.ledgers), "summary_agents": len(s.summaries),
+		"ledger_agents": len(s.ledgers), "observed_agents": len(s.observed), "summary_agents": len(s.summaries),
 	})
 }
 
@@ -233,8 +274,9 @@ func (s *coordinatorState) writeOutput(path string) error {
 		Start     StartInfo                    `json:"start"`
 		Ready     map[string]AgentReady        `json:"ready"`
 		Ledgers   map[string][]AcceptedMessage `json:"ledgers"`
+		Observed  map[string][]AcceptedMessage `json:"observed"`
 		Summaries map[string]AgentSummary      `json:"summaries"`
-	}{s.start, s.ready, s.ledgers, s.summaries}
+	}{s.start, s.ready, s.ledgers, s.observed, s.summaries}
 	s.mu.Unlock()
 	raw, err := json.MarshalIndent(data, "", "  ")
 	if err != nil {
@@ -324,6 +366,27 @@ func (c *coordinatorClient) ledger(ctx context.Context) ([]AcceptedMessage, erro
 	}
 }
 
+func (c *coordinatorClient) submitObserved(ctx context.Context, machine string, messages []AcceptedMessage) error {
+	_, err := c.request(ctx, http.MethodPost, "/v1/observed?machine="+machine, messages, nil)
+	return err
+}
+
+func (c *coordinatorClient) observed(ctx context.Context) ([]AcceptedMessage, error) {
+	for {
+		var messages []AcceptedMessage
+		status, err := c.request(ctx, http.MethodGet, "/v1/observed", nil, &messages)
+		if err != nil {
+			return nil, err
+		}
+		if status == http.StatusOK {
+			return messages, nil
+		}
+		if err = waitContext(ctx, time.Second); err != nil {
+			return nil, err
+		}
+	}
+}
+
 func (c *coordinatorClient) submitSummary(ctx context.Context, summary AgentSummary) error {
 	_, err := c.request(ctx, http.MethodPost, "/v1/summary", summary, nil)
 	return err

loadtest/im-pressure/coordinator_test.go

@@ -90,6 +90,16 @@ func TestCoordinatorClientWaitsForBothAgentsAndMergesLedger(t *testing.T) {
 	if err != nil || len(ledger) != 2 || ledger[0].ClientMid != "a" || ledger[1].ClientMid != "b" {
 		t.Fatalf("ledger=%+v err=%v", ledger, err)
 	}
+	if err = client.submitObserved(context.Background(), "a", []AcceptedMessage{{ClientMid: "unknown", Seq: 10}}); err != nil {
+		t.Fatal(err)
+	}
+	if err = client.submitObserved(context.Background(), "b", []AcceptedMessage{{ClientMid: "unknown", Seq: 10}}); err != nil {
+		t.Fatal(err)
+	}
+	observed, err := client.observed(context.Background())
+	if err != nil || len(observed) != 1 || observed[0].ClientMid != "unknown" {
+		t.Fatalf("observed=%+v err=%v", observed, err)
+	}
 }
 
 func testCoordinatorMux(state *coordinatorState) *http.ServeMux {
@@ -97,6 +107,7 @@ func testCoordinatorMux(state *coordinatorState) *http.ServeMux {
 	mux.HandleFunc("/v1/ready", state.handleReady)
 	mux.HandleFunc("/v1/start", state.handleStart)
 	mux.HandleFunc("/v1/ledger", state.handleLedger)
+	mux.HandleFunc("/v1/observed", state.handleObserved)
 	mux.HandleFunc("/v1/summary", state.handleSummary)
 	mux.HandleFunc("/v1/status", state.handleStatus)
 	return mux

loadtest/im-pressure/merge.go

@@ -41,6 +41,8 @@ type CombinedReport struct {
 	UnexpectedMessages    int64               `json:"unexpected_messages"`
 	HistoryChecked        int64               `json:"history_checked"`
 	HistoryFound          int64               `json:"history_found"`
+	HistoryUnackedChecked int64               `json:"history_unacked_observed_checked"`
+	HistoryUnackedFound   int64               `json:"history_unacked_observed_found"`
 	PersistenceLost       int64               `json:"persistence_lost"`
 	Missing               []MissingDelivery   `json:"missing,omitempty"`
 	Passed                bool                `json:"passed"`
@@ -70,13 +72,18 @@ func runMerge(aPath, bPath, output string) error {
 		report.ReadyConnections += summary.ReadyConnections
 		report.PublishAttempted += summary.PublishAttempted
 		report.PublishAccepted += summary.PublishAccepted
-		report.PublishRejected += summary.PublishRejected
 		for category, count := range summary.PublishErrors {
 			report.PublishErrors[category] += count
 		}
-		report.PublishOutcomeUnknown += summary.PublishErrors["ctrl_timeout"] + summary.PublishErrors["socket"]
+		unknown := summary.PublishOutcomeUnknown
-		report.PublishExplicitReject += summary.PublishErrors["403"] + summary.PublishErrors["502"] +
+		if unknown == 0 {
+			unknown = summary.PublishErrors["ctrl_timeout"] + summary.PublishErrors["socket"]
+		}
+		explicit := summary.PublishErrors["403"] + summary.PublishErrors["502"] +
 			summary.PublishErrors["503"] + summary.PublishErrors["other"]
+		report.PublishOutcomeUnknown += unknown
+		report.PublishExplicitReject += explicit
+		report.PublishRejected += explicit
 		report.ExpectedDeliveries += summary.ExpectedPeer + summary.ExpectedGroup
 		report.ReceivedDeliveries += summary.ReceivedPeer + summary.ReceivedGroup
 		report.RealtimeLost += summary.RealtimeLost
@@ -88,6 +95,8 @@ func runMerge(aPath, bPath, output string) error {
 		report.UnexpectedMessages += summary.UnexpectedMessages
 		report.HistoryChecked += summary.HistoryChecked
 		report.HistoryFound += summary.HistoryFound
+		report.HistoryUnackedChecked += summary.HistoryUnackedChecked
+		report.HistoryUnackedFound += summary.HistoryUnackedFound
 		report.PersistenceLost += summary.PersistenceLost
 		report.CrossClockSkewSamples += summary.CrossClockSkewSamples
 		report.Missing = append(report.Missing, summary.Missing...)

loadtest/im-pressure/merge_test.go

@@ -40,7 +40,8 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) {
 	if report.Passed || report.PersistenceLost != 1 {
 		t.Fatalf("correctness gate did not fail: %+v", report)
 	}
-	if report.UnexpectedMessages != 42 || report.PublishOutcomeUnknown != 5 || report.PublishExplicitReject != 1 {
+	if report.UnexpectedMessages != 42 || report.PublishOutcomeUnknown != 5 ||
+		report.PublishExplicitReject != 1 || report.PublishRejected != 1 {
 		t.Fatalf("ambiguous publish outcomes were not merged: %+v", report)
 	}
 	if len(report.OrderingViolations) != 1 || report.OrderingViolations[0].ReceiverIndex != 7 {

loadtest/im-pressure/model.go

@@ -33,6 +33,8 @@ type Config struct {
 	PhonePrefix      string              `json:"registration_phone_prefix"`
 	VerifyCodeEnv    string              `json:"verification_code_env"`
 	HistoryWorkers   int                 `json:"history_workers"`
+	HistoryPageSize  int                 `json:"history_page_size"`
+	HistoryInterval  string              `json:"history_query_interval"`
 	HistorySettle    string              `json:"history_settle"`
 	HistoryTimeout   string              `json:"history_timeout"`
 	PreflightWorkers int                 `json:"preflight_workers"`
@@ -82,11 +84,23 @@ func loadConfig(path string) (*Config, error) {
 		cfg.DrainGroup = "60s"
 	}
 	if cfg.HistoryWorkers <= 0 {
-		cfg.HistoryWorkers = 12
+		cfg.HistoryWorkers = 1
 	}
 	if cfg.HistoryWorkers > 128 {
 		return nil, errors.New("history_workers must not exceed 128")
 	}
+	if cfg.HistoryPageSize <= 0 {
+		cfg.HistoryPageSize = 50
+	}
+	if cfg.HistoryPageSize > 200 {
+		return nil, errors.New("history_page_size must not exceed 200")
+	}
+	if cfg.HistoryInterval == "" {
+		cfg.HistoryInterval = "100ms"
+	}
+	if _, err = configDuration(cfg.HistoryInterval); err != nil {
+		return nil, fmt.Errorf("history_query_interval: %w", err)
+	}
 	if cfg.HistoryTimeout == "" {
 		cfg.HistoryTimeout = "3m"
 	}
@@ -368,6 +382,7 @@ type AgentSummary struct {
 	PublishAttempted      int64               `json:"publish_attempted"`
 	PublishAccepted       int64               `json:"publish_accepted"`
 	PublishRejected       int64               `json:"publish_rejected"`
+	PublishOutcomeUnknown int64               `json:"publish_outcome_unknown"`
 	PublishErrors         map[string]int64    `json:"publish_errors,omitempty"`
 	ExpectedPeer          int64               `json:"expected_peer"`
 	ReceivedPeer          int64               `json:"received_peer"`
@@ -384,6 +399,8 @@ type AgentSummary struct {
 	UnexpectedMessages    int64               `json:"unexpected_messages"`
 	HistoryChecked        int64               `json:"history_checked"`
 	HistoryFound          int64               `json:"history_found"`
+	HistoryUnackedChecked int64               `json:"history_unacked_observed_checked"`
+	HistoryUnackedFound   int64               `json:"history_unacked_observed_found"`
 	PersistenceLost       int64               `json:"persistence_lost"`
 	AckLatency            HistogramSnapshot   `json:"ack_latency"`
 	DeliveryLatency       HistogramSnapshot   `json:"delivery_latency"`

loadtest/im-pressure/model_test.go

@@ -16,7 +16,8 @@ func TestLoadConfigAppliesBoundedHistoryDefaults(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if cfg.HistoryWorkers != 12 || cfg.HistorySettle != "5m" || cfg.HistoryTimeout != "3m" ||
+	if cfg.HistoryWorkers != 1 || cfg.HistoryPageSize != 50 || cfg.HistoryInterval != "100ms" ||
+		cfg.HistorySettle != "5m" || cfg.HistoryTimeout != "3m" ||
 		cfg.PreflightWorkers != 32 || cfg.WSPingInterval != "20s" {
 		t.Fatalf("unexpected defaults: %+v", cfg)
 	}

loadtest/im-pressure/recorder.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/bits"
+	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -11,9 +12,13 @@ import (
 )
 
 type observation struct {
-	topic string
+	topic         string
-	seq   int64
+	seq           int64
-	bits  []uint64
+	senderMachine string
+	senderIndex   int
+	scenario      string
+	sentAt        int64
+	bits          []uint64
 }
 
 type deliveryRecorder struct {
@@ -65,6 +70,8 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) {
 	var content struct {
 		RunID         string `json:"run_id"`
 		SenderMachine string `json:"sender_machine"`
+		SenderIndex   int    `json:"sender_index"`
+		Scenario      string `json:"scenario"`
 		SentAt        int64  `json:"sent_at_unix_nano"`
 	}
 	if json.Unmarshal(data.Content, &content) != nil || content.RunID != runID {
@@ -87,7 +94,9 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) {
 	defer r.mu.Unlock()
 	obs := r.observed[mid]
 	if obs == nil {
-		obs = &observation{topic: data.Topic, seq: data.Seq, bits: make([]uint64, (r.localUsers+63)/64)}
+		obs = &observation{topic: data.Topic, seq: data.Seq, senderMachine: content.SenderMachine,
+			senderIndex: content.SenderIndex, scenario: content.Scenario, sentAt: content.SentAt,
+			bits: make([]uint64, (r.localUsers+63)/64)}
 		r.observed[mid] = obs
 	}
 	word, bit := index/64, uint(index%64)
@@ -156,6 +165,48 @@ func (r *deliveryRecorder) historyFound(mid string) bool {
 	return r.history[mid]
 }
 
+func (r *deliveryRecorder) unackedObservedMessages(ledger []AcceptedMessage, scenario string) []AcceptedMessage {
+	accepted := make(map[string]bool, len(ledger))
+	for _, message := range ledger {
+		accepted[message.ClientMid] = true
+	}
+	runID, _ := r.runID.Load().(string)
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	messages := make([]AcceptedMessage, 0)
+	for mid, obs := range r.observed {
+		if accepted[mid] || obs.seq <= 0 || obs.topic == "" {
+			continue
+		}
+		kind := "p2p"
+		if strings.HasPrefix(obs.topic, "grp") {
+			kind = "group"
+		}
+		observedScenario := obs.scenario
+		if observedScenario == "" {
+			observedScenario = scenario
+		}
+		message := AcceptedMessage{RunID: runID, Scenario: observedScenario, Kind: kind,
+			ClientMid: mid, Topic: obs.topic, Seq: obs.seq, SenderMachine: obs.senderMachine,
+			SenderIndex: obs.senderIndex, PeerMachine: r.machine, PeerIndex: firstObservedIndex(obs.bits)}
+		if obs.sentAt > 0 {
+			message.SentAt = time.Unix(0, obs.sentAt).UTC()
+		}
+		messages = append(messages, message)
+	}
+	sort.Slice(messages, func(i, j int) bool { return messages[i].ClientMid < messages[j].ClientMid })
+	return messages
+}
+
+func firstObservedIndex(words []uint64) int {
+	for wordIndex, word := range words {
+		if word != 0 {
+			return wordIndex*64 + bits.TrailingZeros64(word) + 1
+		}
+	}
+	return 0
+}
+
 func (r *deliveryRecorder) onDisconnect(_ AccountRecord, _ error) {
 	r.disconnects.Add(1)
 }

loadtest/im-pressure/recorder_test.go

@@ -48,6 +48,22 @@ func TestDeliveryRecorderSeparatesHistoryFromRealtime(t *testing.T) {
 	}
 }
 
+func TestDeliveryRecorderExportsObservedMessagesWithoutAck(t *testing.T) {
+	recorder := newDeliveryRecorder("run1", "b", 2)
+	acked := AcceptedMessage{RunID: "run1", Scenario: "group", Kind: "group", ClientMid: "run1-g-a-0001-01",
+		Topic: "grp-one", Seq: 10, SenderMachine: "a", SenderIndex: 1}
+	unacked := AcceptedMessage{RunID: "run1", Scenario: "group", Kind: "group", ClientMid: "run1-g-a-0002-01",
+		Topic: "grp-one", Seq: 11, SenderMachine: "a", SenderIndex: 2}
+	recorder.onData(AccountRecord{Index: 1}, testData(acked, time.Now()))
+	recorder.onData(AccountRecord{Index: 2}, testData(unacked, time.Now()))
+
+	messages := recorder.unackedObservedMessages([]AcceptedMessage{acked}, "group")
+	if len(messages) != 1 || messages[0].ClientMid != unacked.ClientMid || messages[0].Seq != 11 ||
+		messages[0].PeerMachine != "b" || messages[0].PeerIndex != 2 || messages[0].SenderIndex != 2 {
+		t.Fatalf("unexpected observed ledger: %+v", messages)
+	}
+}
+
 func TestDeliveryRecorderCapturesOrderingViolationDetails(t *testing.T) {
 	recorder := newDeliveryRecorder("run1", "b", 1)
 	newer := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-02", Topic: "p2p-one", Seq: 12}
@@ -84,7 +100,9 @@ func TestDeliveryRecorderSeparatesLocalAndCrossMachineLatency(t *testing.T) {
 }
 
 func testData(message AcceptedMessage, sentAt time.Time) dataPacket {
-	content, _ := json.Marshal(map[string]any{"run_id": message.RunID, "sent_at_unix_nano": sentAt.UnixNano()})
+	content, _ := json.Marshal(map[string]any{"run_id": message.RunID, "scenario": message.Scenario,
+		"sender_machine": message.SenderMachine, "sender_index": message.SenderIndex,
+		"sent_at_unix_nano": sentAt.UnixNano()})
 	return dataPacket{Topic: message.Topic, Seq: message.Seq,
 		Head: map[string]any{"x-im-client-mid": message.ClientMid}, Content: content}
 }