From 4b00b5ae37086323019b2335b2d2e56fbd73823d Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Tue, 29 Sep 2026 14:30:49 +0800 Subject: [PATCH] fix: harden single-machine load handling --- docs/test/IM-单机60%容量压力测试方案.md | 184 ++++++++++++ .../IM-单机60%容量压力测试结果-20260929.md | 119 ++++++++ loadtest/im-pressure/.gitignore | 1 + loadtest/im-pressure/README.md | 9 +- loadtest/im-pressure/agent.go | 62 +++- loadtest/im-pressure/agent_history_test.go | 48 +++ .../config.single-machine.example.json | 31 ++ loadtest/im-pressure/dbverify.go | 25 +- loadtest/im-pressure/histogram.go | 38 ++- loadtest/im-pressure/httpapi.go | 11 +- loadtest/im-pressure/merge.go | 13 +- loadtest/im-pressure/merge_test.go | 10 +- loadtest/im-pressure/model.go | 10 + loadtest/im-pressure/model_test.go | 13 + loadtest/im-pressure/prepare.go | 78 ++++- loadtest/im-pressure/recorder.go | 282 +++++++++++++----- loadtest/im-pressure/recorder_test.go | 62 ++++ server/hub.go | 6 +- server/hub_tenant_test.go | 45 +++ server/im_http.go | 12 + server/im_http_test.go | 26 ++ server/store/group_message_runtime_test.go | 29 ++ server/store/store.go | 56 +++- server/topic.go | 4 + 24 files changed, 1030 insertions(+), 144 deletions(-) create mode 100644 docs/test/IM-单机60%容量压力测试方案.md create mode 100644 docs/test/IM-单机60%容量压力测试结果-20260929.md create mode 100644 loadtest/im-pressure/config.single-machine.example.json diff --git a/docs/test/IM-单机60%容量压力测试方案.md b/docs/test/IM-单机60%容量压力测试方案.md new file mode 100644 index 0000000..2e42e21 --- /dev/null +++ b/docs/test/IM-单机60%容量压力测试方案.md @@ -0,0 +1,184 @@ +# IM 单机 60% 容量压力测试方案 + +日期：2026-09-29 + +本方案首轮执行结果见 [IM-单机60%容量压力测试结果-20260929.md](./IM-单机60%容量压力测试结果-20260929.md)。 + +## 1. 目标与口径 + +本方案用一台物理压测机完成单聊、群聊和混合场景，容量按双机正式方案全局规模的 60% 计算。 + +单机不等于只运行一个 agent。为保留现有好友对和投递契约，同一台物理机上同时运行逻辑 agent A 和 agent B，每个 agent 承载 600 个连接，合计 1200 个 WebSocket。Coordinator 仍使用 `expected-agents=2`。 + +| 项目 | 双机正式方案 | 单机 60% 方案 | +|---|---:|---:| +| 物理压测机 | 2 | 1 | +| 逻辑 agent | 2 | 2 | +| 每个 agent 连接 | 1000 | 600 | +| WebSocket 总连接 | 2000 | 1200 | +| 好友对 | 1000 | 600 | +| 群成员 | 2000 | 1200 | + +## 2. 场景容量 + +### 2.1 单聊 + +- 600 对好友，A/B 各 600 人。 +- 每个用户发送 10 条，每秒 1 条。 +- 总发送尝试 12000 条，是双机 20000 条的 60%。 +- 每条 ACK 消息应在配对用户的 WebSocket 到达一次。 + +### 2.2 群聊 + +- 1200 个在线群成员。 +- A/B 各取前 300 个用户作为发送者，合计 600 个发送者。 +- 每人发送 2 条，全局每 100ms 一条，合计 1200 条群消息，耗时约 120 秒。 +- 全部发送被 ACK 时，期望实时投递 `1200 x 1200 = 1,440,000` 次。 + +### 2.3 单聊和群聊混合 + +- 前 180 对好友互发，每人 5 条，共 1800 条单聊。 +- A/B 各 150 个群发送者，每人 2 条，共 600 条群消息。 +- 群消息期望实时投递 `600 x 1200 = 720,000` 次。 +- 单聊和群聊同时开始，不分两轮单独执行。 + +工具按 `users_per_machine` 等比计算发送者数量。`users_per_machine=1000` 时保持原双机参数，设为 600 时自动使用上述单机参数。 + +## 3. 文件隔离 + +单机方案不复用双机的运行文件，避免把 1200 人结果和 2000 人结果混在一起。 + +```text +loadtest/im-pressure/config.single-machine.json +loadtest/im-pressure/manifest.single-machine.json +loadtest/im-pressure/results// +``` + +从模板生成本地配置： + +```bash +cp loadtest/im-pressure/config.single-machine.example.json \ + loadtest/im-pressure/config.single-machine.json +chmod 600 loadtest/im-pressure/config.single-machine.json +``` + +修改租户、HTTP/WS 地址和注册手机前缀。A/B endpoint 可以填同一个公网入口；若需确认跨 Pod 投递，应分别填写两个定向入口。 + +## 4. 一次性数据准备 + +```bash +source loadtest/im-pressure/.env.test +go build -o loadtest/im-pressure/im-pressure ./loadtest/im-pressure +ulimit -n 100000 + +./loadtest/im-pressure/im-pressure prepare \ + -config loadtest/im-pressure/config.single-machine.json \ + -manifest loadtest/im-pressure/manifest.single-machine.json + +./loadtest/im-pressure/im-pressure verify \ + -config loadtest/im-pressure/config.single-machine.json \ + -manifest loadtest/im-pressure/manifest.single-machine.json +``` + +`verify` 必须输出 1200 个账号、600 对好友和 1200 个群成员。单机 manifest 可以复用已注册的 `lta0001..0600` 和 `ltb0001..0600` 账号，但必须使用独立群 Topic。 + +## 5. 正式运行 + +以群聊为例： + +```bash +export RUN_ID='single60-group-r1-YYYYMMDD-01' +export SCENARIO='group' +mkdir -p "loadtest/im-pressure/results/${RUN_ID}" +``` + +终端 1，启动 coordinator： + +```bash +./loadtest/im-pressure/im-pressure coordinator \ + -listen :8090 \ + -run-id "${RUN_ID}" \ + -scenario "${SCENARIO}" \ + -expected-agents 2 \ + -output "loadtest/im-pressure/results/${RUN_ID}/coordinator.json" +``` + +终端 2，启动逻辑 agent A： + +```bash +./loadtest/im-pressure/im-pressure agent \ + -config loadtest/im-pressure/config.single-machine.json \ + -manifest loadtest/im-pressure/manifest.single-machine.json \ + -machine a \ + -coordinator http://127.0.0.1:8090 \ + -metrics-listen :9105 \ + -output "loadtest/im-pressure/results/${RUN_ID}/agent-a-summary.json" +``` + +终端 3，启动逻辑 agent B： + +```bash +./loadtest/im-pressure/im-pressure agent \ + -config loadtest/im-pressure/config.single-machine.json \ + -manifest loadtest/im-pressure/manifest.single-machine.json \ + -machine b \ + -coordinator http://127.0.0.1:8090 \ + -metrics-listen :9106 \ + -output "loadtest/im-pressure/results/${RUN_ID}/agent-b-summary.json" +``` + +两个 agent 都报告 `600/600 READY` 并稳定通过 30 秒屏障后才会开始。任一 agent 未满连接时 coordinator 会撤销开始时间。 + +## 6. 指标采集 + +终端 4 在起压前至少 2 分钟启动： + +```bash +./loadtest/im-pressure/im-pressure collect \ + -duration 20m -interval 5s \ + -coordinator http://127.0.0.1:8090 \ + -coordinator-output "loadtest/im-pressure/results/${RUN_ID}/coordinator.json" \ + -url agent-a=http://127.0.0.1:9105/metrics \ + -url agent-b=http://127.0.0.1:9106/metrics \ + -url node=http://127.0.0.1:9100/metrics \ + -output "loadtest/im-pressure/results/${RUN_ID}/metrics.jsonl" +``` + +同时采集 Redis、Kafka、MySQL、Pod CPU/内存/网络、Kafka consumer lag、MySQL 连接池等指标。如本机没有 `node-exporter :9100`，删除 `node` URL，但不影响 agent 指标采集。 + +## 7. 报告和独立落库核验 + +```bash +./loadtest/im-pressure/im-pressure merge \ + -a "loadtest/im-pressure/results/${RUN_ID}/agent-a-summary.json" \ + -b "loadtest/im-pressure/results/${RUN_ID}/agent-b-summary.json" \ + -output "loadtest/im-pressure/results/${RUN_ID}/report.json" + +export MYSQL_DSN='readonly:password@tcp(mysql-host:3306)/meeting_im?parseTime=true' +export TENANT='006' # 必须与 config.single-machine.json 中的 tenant 一致 +./loadtest/im-pressure/im-pressure verify-db \ + -coordinator "loadtest/im-pressure/results/${RUN_ID}/coordinator.json" \ + -tenant "${TENANT}" \ + -dsn-env MYSQL_DSN \ + -registry-db hotgo \ + -output "loadtest/im-pressure/results/${RUN_ID}/db-verification.json" +``` + +## 8. 验收门禁 + +- READY 必须是 `1200/1200`，统计窗口内断连为 0。 +- publish 明确拒绝和结果未知均为 0，成功率至少 99.99%。 +- ACK P95 不高于 500ms，同机投递 P95 不高于 1000ms。 +- 实时缺失、重复、乱序、未知本轮消息均为 0。 +- ACK 台账和无 ACK 但已观测台账的历史回查命中率必须 100%，独立 MySQL 核验丢失、错配、重复均为 0。 +- 实时发送、历史回查和渐进关闭三个阶段分别统计 Topic/Proxy/Session 队列满，不能只看整轮总数。 + +## 9. 单机特有风险 + +1. 两个 agent 共用一台机器的 CPU、内存、FD、NAT 和出口带宽，测到的是“单机压测端 + 服务端”综合上限，不能代替双机最终验收。 +2. 正式前做 120/600/1200 连接三档阶梯，观察 agent CPU、`go_goroutines`、FD、接收速率和 Ping/Pong 断连，确认压测端本身没有先饱和。当前工具已将投递观测按 `client_mid` 分片、接收账号顺序状态独立加锁，并使用原子延迟直方图；仍需用指标证明本机有余量。 +3. 两个 agent 必须使用不同指标端口，不能同时绑定 `:9105`。 +4. 压测域名如被 VPN/TUN 接管，起压前必须确认所有当前 IP 的路由都指向物理网卡。 +5. 单机 60% 通过后只能进入双机 100% 验收，不能直接得出 2000 连接容量通过的结论。 + +断连时同时查看 agent summary 中的 `disconnect_reasons` 和 `disconnect_samples`。前者按 timeout、abnormal close、connection reset 等原因分类计数，后者最多保留 100 条带账号与时间的原始错误样本，用于区分压测端读循环阻塞、网络中断和服务端主动关闭。 diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md new file mode 100644 index 0000000..02ddfbc --- /dev/null +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -0,0 +1,119 @@ +# IM 单机 60% 容量压力测试结果 + +日期：2026-09-29 + +## 1. 结论 + +本轮使用一台物理压测机同时运行逻辑 agent A/B，每侧 600 个 WebSocket，合计 1200 连接。60% 容量的正确性和性能总验收未通过。 + +- 单聊实时和持久化正确性通过，但 ACK P95 和投递 P95 未达标。 +- 群聊在 1200 条群消息、144 万设计投递量下出现 ACK 超时、连接重置和分钟级投递延迟。 +- 混合场景明显越过容量拐点，737 个连接断开，仅 44.83% 发送得到 ACK。 +- 独立 MySQL 核验确认：三个场景的所有 ACK 消息均精确落库，0 错配、0 重复；所有“无 ACK 但 WS 已观测”消息也均已落库。 +- 压测工具本轮峰值 RSS 不超过 127 MB，每个 agent 峰值 FD 614，未显示本机内存或 FD 先饱和。 +- 混合实时阶段的群 Topic 所在 `im-server-0` 出现 broadcast queue full，服务端 Topic 处理能力是已证实的瓶颈之一。 + +## 2. 环境与口径 + +- 租户：`allrich` +- HTTP/WS 入口：`api.118im.com` +- 物理压测机：1 台 +- 逻辑 agent：A/B 各2个中的一个，每侧 600 连接 +- 账号：1200，好友对：600，群成员：1200 +- API 域名当前 3 个 IP 均经 `en1` 和局域网网关直连，未走 VPN/TUN +- IM Server：3 Pod，无重启；每 Pod request `1 CPU/512Mi`，limit `2 CPU/1Gi` +- 独立群与 manifest：`loadtest/im-pressure/manifest.single-machine.json` + +准备阶段最终验证为 `accounts=1200 pairs=600 group_members=1200`。群批量邀请曾出现入口 502 和幂等任务长时间处理中，工具通过按实际成员补齐缺失用户完成恢复。 + +## 3. 结果汇总 + +| 场景 | Run ID | ACK | 实时投递 | 断连 | ACK P95 | 投递 P95 | 独立 DB 核验 | 结果 | +|---|---|---:|---:|---:|---:|---:|---:|---| +| 单聊 | `s60-p2p-r2-0929` | 12000/12000 | 12000/12000 | 0 | 1341.555ms | 2000ms | 12000 ACK 全部精确命中 | 性能未通过 | +| 群聊 | `s60-group-r1-0929` | 1145/1200 | 1350508/1374000 ACK 台账 | 47 | 11207.859ms | 94621.882ms | 1145 ACK + 39 无 ACK 落库 | 未通过 | +| 混合 | `s60-mixed-r1-0929` | 1076/2400 | 234829/263657 ACK 台账 | 737 | 17491.634ms | 101226.392ms | 1076 ACK + 1017 无 ACK 落库 | 未通过 | + +表中群聊和混合的“期望投递”只按有 ACK 的 ledger 计算。发送结果未知的消息不能在没有 seq 时直接加入期望投递分母，因此该分母小于设计投递量。 + +## 4. 分场景记录 + +### 4.1 单聊 + +R1 的历史查询将 `limit=10` 错误地按目标消息数计算，没有计入 P2P Topic 中双方交错的 seq，造成 4619 条假缺失。修复后 R2 结果： + +- 12000 条全部 ACK，无结果未知。 +- 12000 次对端投递全部到达，无重复、乱序、断连。 +- 12000 条历史回查全部命中，持久化缺失 0。 +- ACK P95 1341.555ms，超过 500ms 门禁；同机投递 P95 落在 2000ms 桶，超过 1000ms 门禁。 + +### 4.2 群聊 + +- 发送 1200 条，1145 条 ACK，43 条 ACK 超时，12 条 socket 结果未知，无明确拒绝。 +- 47 个连接断开：22 个 abnormal close，25 个 connection reset by peer。 +- ACK 台账下实时缺失 23492 次，这些消息最终均可从历史查到。 +- 55 条结果未知中，39 条被接收端实时观测，且全部被独立 SQL 确认落库；另 16 条既无 ACK、无 WS 观测，也无数据库记录。 +- 断连发生时间为 `03:40:10Z` 至 `03:41:15Z`，与群实时发送阶段重合。 + +### 4.3 混合 + +- 设计发送为 1800 条单聊和 600 条群聊，合计 2400 条。 +- 1076 条 ACK，1324 条结果未知，其中 1036 条 ACK 超时、288 条 socket 失败，无明确拒绝。 +- 737 个连接断开：69 个 abnormal close，668 个 connection reset by peer。 +- ACK 台账下实时缺失 28828 次，其中 28633 次对应消息最终已查到持久化记录。 +- 历史验证 ledger 含 2002 条，命中 1331 条，缺失 671 条全部为 P2P。独立 SQL 确认所有 1076 条 ACK 消息和全部 926 条 WS 已观测的无 ACK 消息均已落库，因此这 671 条是 verifier 连接掉线和 3 分钟总预算造成的验证不完整，不是数据库丢失。 +- 1324 条结果未知中，1017 条实际落库：926 条被 WS 观测，另有 91 条未被 WS 观测但已落库。剩余 307 条无 ACK、无 WS 观测、无数据库记录。 + +### 4.4 独立 MySQL 核验 + +核验租户 ID 为 8，消息表为 `im_008_messages` 和 `im_008_group_messages_202609`。 + +| 场景 | ACK expected | 精确命中 | ACK 缺失 | 错配 | 重复 | 无 ACK 但落库 | +|---|---:|---:|---:|---:|---:|---:| +| 单聊 | 12000 | 12000 | 0 | 0 | 0 | 0 | +| 群聊 | 1145 | 1145 | 0 | 0 | 0 | 39 | +| 混合 | 1076 | 1076 | 0 | 0 | 0 | 1017 | + +`verify-db` 对无 ACK 但落库的行按 unexpected 报告，所以群聊和混合命令会以非零状态结束；这不代表数据错误，而是发送结果未知但服务端已持久化的重要诊断证据。 + +## 5. 服务端观测 + +- `im-server-0/1/2` 全部 Running，本轮重启数均为 0。 +- 混合实时阶段 `03:53:00Z..03:53:02Z`，`im-server-0` 记录 55 次 `topic's broadcast queue is full`。 +- 混合渐进关闭末尾 `04:03:45Z`，`im-server-0` 再记录 375 次同类队列满。当前 60 秒 close ramp 仍未完全避免关闭风暴。 +- 同一观测窗口的 `03:45:13Z`，`im-server-2` 在 `serveIMFriendRequestAction` (`server/im_http.go:6256`) 出现一次 nil pointer panic。该端点不属于群实时发送路径，不将其归因为本轮断连，但需单独修复。 +- 当前 kubeconfig 无权读取 metrics-server，未能取得负载窗口的 Pod CPU/内存历史值。 + +## 6. 压测工具修复 + +1. 新增单机 60% 配置和执行方案，A/B 各 600 连接。 +2. 群聊和混合发送者按 `users_per_machine` 比例缩放，1000 用户的原参数保持不变。 +3. 投递观测按 `client_mid` 分为 64 片，每个接收账号独立维护 seq，延迟直方图改为原子计数，移除 WS 读循环中的全局锁争用。 +4. summary/report 新增 `disconnect_reasons` 和最多 100 条 `disconnect_samples`。 +5. 群创建后立即 checkpoint manifest；批量邀请失败后重读实际成员，仅邀请缺失用户。 +6. P2P 历史查询 limit 按 seq 区间扩展，上限 200，修复双方消息交错造成的假缺失。 + +## 7. 已实施改进 + +1. 好友申请接受和自动接受路径增加 store 空结果保护。底层异常返回 `nil, nil` 时记录租户、申请、操作人和状态并返回 HTTP 500，不再触发进程 panic。 +2. Hub 向已满的 Topic client queue 投递失败时，立即向原发送连接返回 `503 service unavailable`，并记录 topic、tenant、队列长度和容量。客户端可据此重试，压测也不再把这类请求归为等待超时的未知结果。 +3. 群消息权限检查改为复用群设置缓存，并为个人禁言增加 Redis 正向和负向缓存；新增禁言和解除禁言会同步更新缓存，到期禁言会自动淘汰。此改动减少群 Topic 串行处理路径上的 MySQL 查询。 +4. `message_publish_stage_duration_seconds` 新增 `group/permission_check` 和 `*/realtime_fanout` 阶段，下轮可区分权限查询、持久化和实时扩散耗时。 +5. A/B 两个逻辑 agent 按账号顺序交错关闭连接。`close_ramp=60s` 时每个 agent 的本地关闭过程约 120 秒，A/B 相差半个时隙，整体约每 100ms 关闭一个连接，避免两侧同时下线形成双倍尖峰。 +6. 已增加空结果防 panic、Topic queue 满返回 503、双 agent 关闭时隙的自动化测试。 + +## 8. 复测顺序 + +1. 发版后先跑单机群聊 60%，验收实时发送阶段和关闭阶段 `topic_client` enqueue failed 均为 0；若仍过载，发送端必须收到 503，不能再出现对应的纯 ACK timeout。 +2. 对比 `permission_check`、`save`、`realtime_fanout` 三阶段的 P50/P95/P99，确认群权限缓存是否消除了数据库检查瓶颈。 +3. 群聊通过后再跑单机混合 60%，要求 ACK、历史和独立 SQL 三方账本一致，连接断开数为 0。 +4. 补齐 Prometheus/metrics-server 权限并同步保存 Pod CPU、内存、网络、Kafka lag、Redis、MySQL 指标；在此之前不进行双机 100% 最终验收。 + +## 9. 产物 + +- 方案：`docs/test/IM-单机60%容量压力测试方案.md` +- 单聊 R2：`loadtest/im-pressure/results/s60-p2p-r2-0929/` +- 群聊 R1：`loadtest/im-pressure/results/s60-group-r1-0929/` +- 混合 R1：`loadtest/im-pressure/results/s60-mixed-r1-0929/` + +结果目录包含 coordinator、A/B summary、合并 report 和 agent 指标 JSONL。 diff --git a/loadtest/im-pressure/.gitignore b/loadtest/im-pressure/.gitignore index 46bc805..c02014d 100644 --- a/loadtest/im-pressure/.gitignore +++ b/loadtest/im-pressure/.gitignore @@ -1,5 +1,6 @@ /im-pressure /config.json +/config.single-machine.json /config.local*.json /.env.test /manifest*.json diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index 67c42e5..ded1ac9 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -2,6 +2,9 @@ 该工具按配置指定的企业租户运行：两台压测机各承载 1000 个 WebSocket 用户，共 2000 个用户、1000 对好友和一个 2000 人群。当前测试环境使用 `allrich`，本轮实时参数见 [MACHINE-B.md](./MACHINE-B.md) 和 [allrich 双机压测执行记录](../../docs/test/IM-allrich双机压测执行记录-20260927.md)。 +只使用一台物理压测机时，请使用 [单机 60% 容量压测方案](../../docs/test/IM-单机60%容量压力测试方案.md) 和 [config.single-machine.example.json](./config.single-machine.example.json)。单机依然运行 A/B 两个逻辑 agent，用于保留好友对、跨逻辑分片投递和合并报告的完整契约。 +首轮实测数据与瓶颈分析见 [单机 60% 容量压测结果](../../docs/test/IM-单机60%容量压力测试结果-20260929.md)。 + 机器 B 上的 AI 或操作人请直接阅读 [MACHINE-B.md](./MACHINE-B.md)。该文件不重复讲方案，只给出机器 B 从核验到启动、观察和回传结果的完整命令。 ## 两台机器的固定分工 @@ -194,8 +197,8 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}" 两个 agent 都建立 1000 条连接后，协调器下发统一的 UTC `start_at`。场景可替换为： - `p2p`：1000 对用户两侧同时发送，默认每人 10 条、间隔 1 秒。 -- `group`：每台机器 500 个发送者，每人 2 条；全局每 100ms 一条，共 2000 条。 -- `mixed`：300 对私聊用户每人 5 条，同时每台 250 个群发送者各发 2 条。 +- `group`：`users_per_machine=1000` 时每个 agent 有 500 个发送者，每人 2 条；全局每 100ms 一条，共 2000 条。其他容量按每个 agent 用户数的 50% 选发送者。 +- `mixed`：`users_per_machine=1000` 时使用 300 对私聊用户和每个 agent 250 个群发送者；其他容量分别按 30% 和 25% 同比计算。 ## 机器 A：回收并合并报告 @@ -215,7 +218,7 @@ scp "${MACHINE_B_SSH}:${MACHINE_B_REPO}/loadtest/im-pressure/results/${RUN_ID}/a -output "loadtest/im-pressure/results/${RUN_ID}/report.json" ``` -`report.json` 给出 ACK、实时投递、重复、乱序、断连和历史持久化核验结果。当前正确性门禁要求上述错误计数全部为 0。 +`report.json` 给出 ACK、实时投递、重复、乱序、断连和历史持久化核验结果。断连会额外输出 `disconnect_reasons` 分类计数和最多 100 条 `disconnect_samples` 原始错误样本。当前正确性门禁要求上述错误计数全部为 0。 每个 agent 在 `:9105/metrics` 暴露连接、发送、接收、重复、乱序、断连以及 Go/process 指标；端口冲突时可通过 `-metrics-listen` 修改，传空字符串可关闭。 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 5170945..0f5d6a3 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -458,8 +458,18 @@ func newHistoryQuery(client *WSClient, topic string, messages []AcceptedMessage, for _, message := range messages { expected = append(expected, message.ClientMid) } - return historyQuery{client: client, topic: topic, since: messages[0].Seq, - before: messages[len(messages)-1].Seq + 1, limit: limit, expected: expected} + since := messages[0].Seq + before := messages[len(messages)-1].Seq + 1 + // P2P topics interleave both senders. The requested range must leave room + // for non-target messages between the expected sequence numbers. + if span := before - since; span > int64(limit) { + limit = int(span) + } + if limit > 200 { + limit = 200 + } + return historyQuery{client: client, topic: topic, since: since, + before: before, limit: limit, expected: expected} } func (r *agentRun) waitHistoryPage(ctx context.Context, expected []string, timeout time.Duration) { @@ -659,13 +669,24 @@ func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic stri case "p2p": return r.scheduleP2P(ctx, start, r.cfg.UsersPerMachine, r.cfg.P2PMessages, mustDuration(r.cfg.P2PInterval)) case "group": - return r.scheduleGroup(ctx, start, groupTopic, 1, 500, 2) + return r.scheduleGroup(ctx, start, groupTopic, 1, scaledUsers(r.cfg.UsersPerMachine, 1, 2), 2) case "mixed": + p2pUsers := scaledUsers(r.cfg.UsersPerMachine, 3, 10) + groupSenders := scaledUsers(r.cfg.UsersPerMachine, 1, 4) + if p2pUsers+groupSenders > r.cfg.UsersPerMachine { + groupSenders = r.cfg.UsersPerMachine - p2pUsers + } + if groupSenders <= 0 { + return fmt.Errorf("users_per_machine=%d is too small for mixed scenario", r.cfg.UsersPerMachine) + } var wg sync.WaitGroup errCh := make(chan error, 2) wg.Add(2) - go func() { defer wg.Done(); errCh <- r.scheduleP2P(ctx, start, 300, 5, 20*time.Second) }() - go func() { defer wg.Done(); errCh <- r.scheduleGroup(ctx, start, groupTopic, 301, 550, 2) }() + go func() { defer wg.Done(); errCh <- r.scheduleP2P(ctx, start, p2pUsers, 5, 20*time.Second) }() + go func() { + defer wg.Done() + errCh <- r.scheduleGroup(ctx, start, groupTopic, p2pUsers+1, p2pUsers+groupSenders, 2) + }() wg.Wait() close(errCh) for err := range errCh { @@ -679,6 +700,17 @@ func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic stri } } +func scaledUsers(total, numerator, denominator int) int { + count := total * numerator / denominator + if count < 1 { + return 1 + } + if count > total { + return total + } + return count +} + func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, messages int, interval time.Duration) error { var wg sync.WaitGroup for index := 1; index <= users; index++ { @@ -858,9 +890,10 @@ func (r *agentRun) drainDuration() (time.Duration, error) { func (r *agentRun) close() { clients := r.clientsSnapshot() - var interval time.Duration - if r.cfg != nil && r.cfg.CloseRamp != "" && len(clients) > 1 { - interval = mustDuration(r.cfg.CloseRamp) / time.Duration(len(clients)-1) + sort.Slice(clients, func(i, j int) bool { return clients[i].account.Index < clients[j].account.Index }) + interval, initialDelay := closeSchedule(r.cfg, r.machine, len(clients)) + if initialDelay > 0 { + time.Sleep(initialDelay) } for index, client := range clients { client.Close() @@ -870,6 +903,19 @@ func (r *agentRun) close() { } } +func closeSchedule(cfg *Config, machine string, clients int) (interval, initialDelay time.Duration) { + if cfg == nil || cfg.CloseRamp == "" || clients <= 1 { + return 0, 0 + } + // The two logical agents share one global ramp. Interleaving their slots avoids + // doubling the disconnect rate when both agents begin cleanup at the same time. + interval = 2 * mustDuration(cfg.CloseRamp) / time.Duration(clients-1) + if machine == "b" { + initialDelay = interval / 2 + } + return interval, initialDelay +} + func waitUntil(ctx context.Context, target time.Time) error { delay := time.Until(target) if delay <= 0 { diff --git a/loadtest/im-pressure/agent_history_test.go b/loadtest/im-pressure/agent_history_test.go index 614013e..8b1e024 100644 --- a/loadtest/im-pressure/agent_history_test.go +++ b/loadtest/im-pressure/agent_history_test.go @@ -2,7 +2,9 @@ package main import ( "errors" + "fmt" "testing" + "time" ) func TestHistoryQueriesForMessagesGroupsRangesByTopic(t *testing.T) { @@ -51,6 +53,40 @@ func TestClassifyPublishErrorTreatsClosedConnectionAsUnknownSocketError(t *testi } } +func TestScenarioUsersScaleWithConfiguredCapacity(t *testing.T) { + tests := []struct { + users int + group int + mixedP2P int + mixedGroup int + }{ + {users: 1000, group: 500, mixedP2P: 300, mixedGroup: 250}, + {users: 600, group: 300, mixedP2P: 180, mixedGroup: 150}, + } + for _, tc := range tests { + if got := scaledUsers(tc.users, 1, 2); got != tc.group { + t.Fatalf("users=%d group=%d want=%d", tc.users, got, tc.group) + } + if got := scaledUsers(tc.users, 3, 10); got != tc.mixedP2P { + t.Fatalf("users=%d mixed p2p=%d want=%d", tc.users, got, tc.mixedP2P) + } + if got := scaledUsers(tc.users, 1, 4); got != tc.mixedGroup { + t.Fatalf("users=%d mixed group=%d want=%d", tc.users, got, tc.mixedGroup) + } + } +} + +func TestNewHistoryQueryIncludesInterleavedSequenceRange(t *testing.T) { + messages := make([]AcceptedMessage, 10) + for i := range messages { + messages[i] = AcceptedMessage{ClientMid: fmt.Sprintf("mid-%d", i), Seq: int64(100 + i*2)} + } + query := newHistoryQuery(nil, "p2p-one", messages, 10) + if query.since != 100 || query.before != 119 || query.limit != 19 { + t.Fatalf("history query range=%d-%d limit=%d want=100-119/19", query.since, query.before, query.limit) + } +} + 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} @@ -65,3 +101,15 @@ func TestMergeMessageLedgersPrefersPeerObservationAndAck(t *testing.T) { t.Fatalf("ACK ledger was not authoritative: %+v", merged) } } + +func TestCloseScheduleInterleavesLogicalAgents(t *testing.T) { + cfg := &Config{CloseRamp: "60s"} + intervalA, delayA := closeSchedule(cfg, "a", 601) + intervalB, delayB := closeSchedule(cfg, "b", 601) + if intervalA != 200*time.Millisecond || intervalB != intervalA { + t.Fatalf("intervals: a=%s b=%s, want 200ms", intervalA, intervalB) + } + if delayA != 0 || delayB != 100*time.Millisecond { + t.Fatalf("delays: a=%s b=%s, want 0/100ms", delayA, delayB) + } +} diff --git a/loadtest/im-pressure/config.single-machine.example.json b/loadtest/im-pressure/config.single-machine.example.json new file mode 100644 index 0000000..63aff5e --- /dev/null +++ b/loadtest/im-pressure/config.single-machine.example.json @@ -0,0 +1,31 @@ +{ + "tenant": "006", + "http_base": "https://im-api.example.com", + "api_key_env": "IM_API_KEY", + "password_env": "IM_LOAD_PASSWORD", + "verification_code_env": "IM_VERIFICATION_CODE", + "registration_phone_prefix": "", + "users_per_machine": 600, + "endpoints": { + "a": ["wss://im-api.example.com/v0/channels"], + "b": ["wss://im-api.example.com/v0/channels"] + }, + "insecure_tls": false, + "prepare_workers": 20, + "connect_ramp": "30s", + "command_timeout": "20s", + "ws_ping_interval": "20s", + "preflight_workers": 32, + "skip_p2p_preflight": false, + "history_workers": 12, + "history_page_size": 10, + "history_query_interval": "500ms", + "history_page_wait": "3s", + "history_settle": "5m", + "history_timeout": "3m", + "close_ramp": "60s", + "p2p_messages_per_user": 10, + "p2p_interval": "1s", + "drain_p2p": "30s", + "drain_group": "60s" +} diff --git a/loadtest/im-pressure/dbverify.go b/loadtest/im-pressure/dbverify.go index 2ad056e..a242938 100644 --- a/loadtest/im-pressure/dbverify.go +++ b/loadtest/im-pressure/dbverify.go @@ -14,17 +14,18 @@ import ( ) type dbVerifyResult struct { - RunID string `json:"run_id"` - TenantCode string `json:"tenant_code"` - TenantID int64 `json:"tenant_id"` - Expected int `json:"expected"` - FoundExact int `json:"found_exact"` - Missing int `json:"missing"` - Mismatch int `json:"mismatch"` - DuplicateRows int `json:"duplicate_rows"` - Unexpected int `json:"unexpected_rows"` - Tables []string `json:"tables"` - MissingMIDs []string `json:"missing_client_mids,omitempty"` + RunID string `json:"run_id"` + TenantCode string `json:"tenant_code"` + TenantID int64 `json:"tenant_id"` + Expected int `json:"expected"` + FoundExact int `json:"found_exact"` + Missing int `json:"missing"` + Mismatch int `json:"mismatch"` + DuplicateRows int `json:"duplicate_rows"` + Unexpected int `json:"unexpected_rows"` + Tables []string `json:"tables"` + MissingMIDs []string `json:"missing_client_mids,omitempty"` + UnexpectedMIDs []string `json:"unexpected_client_mids,omitempty"` } type persistedIdentity struct { @@ -145,9 +146,11 @@ func runDBVerify(ctx context.Context, coordinatorPath, output, tenantCode, dsnEn for mid := range actual { if _, ok := expected[mid]; !ok { result.Unexpected++ + result.UnexpectedMIDs = append(result.UnexpectedMIDs, mid) } } sort.Strings(result.MissingMIDs) + sort.Strings(result.UnexpectedMIDs) resultRaw, _ := json.MarshalIndent(result, "", " ") if output == "" || output == "-" { if _, err = fmt.Fprintln(os.Stdout, string(resultRaw)); err != nil { diff --git a/loadtest/im-pressure/histogram.go b/loadtest/im-pressure/histogram.go index 5e7958b..a6c55e1 100644 --- a/loadtest/im-pressure/histogram.go +++ b/loadtest/im-pressure/histogram.go @@ -2,7 +2,7 @@ package main import ( "sort" - "sync" + "sync/atomic" "time" ) @@ -13,11 +13,10 @@ var latencyBounds = []time.Duration{ } type DurationHistogram struct { - mu sync.Mutex - count int64 - sum time.Duration - max time.Duration - bucket []int64 + count atomic.Int64 + sum atomic.Int64 + max atomic.Int64 + bucket []atomic.Int64 } type HistogramSnapshot struct { @@ -28,37 +27,36 @@ type HistogramSnapshot struct { } func NewDurationHistogram() *DurationHistogram { - return &DurationHistogram{bucket: make([]int64, len(latencyBounds)+1)} + return &DurationHistogram{bucket: make([]atomic.Int64, len(latencyBounds)+1)} } func (h *DurationHistogram) Observe(value time.Duration) { if value < 0 { return } - h.mu.Lock() - defer h.mu.Unlock() - h.count++ - h.sum += value - if value > h.max { - h.max = value + nanos := int64(value) + h.count.Add(1) + h.sum.Add(nanos) + for previous := h.max.Load(); nanos > previous; previous = h.max.Load() { + if h.max.CompareAndSwap(previous, nanos) { + break + } } idx := sort.Search(len(latencyBounds), func(i int) bool { return value <= latencyBounds[i] }) - h.bucket[idx]++ + h.bucket[idx].Add(1) } func (h *DurationHistogram) Snapshot() HistogramSnapshot { - h.mu.Lock() - defer h.mu.Unlock() buckets := make(map[string]int64, len(h.bucket)) - for i, count := range h.bucket { + for i := range h.bucket { label := "+Inf" if i < len(latencyBounds) { label = latencyBounds[i].String() } - buckets[label] = count + buckets[label] = h.bucket[i].Load() } return HistogramSnapshot{ - Count: h.count, SumMS: float64(h.sum) / float64(time.Millisecond), - MaxMS: float64(h.max) / float64(time.Millisecond), Buckets: buckets, + Count: h.count.Load(), SumMS: float64(h.sum.Load()) / float64(time.Millisecond), + MaxMS: float64(h.max.Load()) / float64(time.Millisecond), Buckets: buckets, } } diff --git a/loadtest/im-pressure/httpapi.go b/loadtest/im-pressure/httpapi.go index 4a23830..93c2fc4 100644 --- a/loadtest/im-pressure/httpapi.go +++ b/loadtest/im-pressure/httpapi.go @@ -269,6 +269,11 @@ type groupMembersResponse struct { } func (a *apiClient) countGroupMembers(ctx context.Context, account AccountRecord, topic string) (int, error) { + members, err := a.groupMemberIDs(ctx, account, topic) + return len(members), err +} + +func (a *apiClient) groupMemberIDs(ctx context.Context, account AccountRecord, topic string) (map[string]bool, error) { cursor := "" seen := make(map[string]bool) for { @@ -278,7 +283,7 @@ func (a *apiClient) countGroupMembers(ctx context.Context, account AccountRecord } var page groupMembersResponse if err := a.do(ctx, http.MethodGet, path, account.Token, "", nil, &page); err != nil { - return 0, err + return nil, err } for _, item := range page.Items { seen[item.UserID] = true @@ -287,11 +292,11 @@ func (a *apiClient) countGroupMembers(ctx context.Context, account AccountRecord break } if page.NextCursor == cursor { - return 0, fmt.Errorf("group member cursor did not advance: %s", cursor) + return nil, fmt.Errorf("group member cursor did not advance: %s", cursor) } cursor = page.NextCursor } - return len(seen), nil + return seen, nil } func queryInt(value string, fallback int) int { diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index ab586c8..17f5fba 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -38,6 +38,8 @@ type CombinedReport struct { OutOfOrder int64 `json:"out_of_order"` OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` Disconnects int64 `json:"disconnects"` + DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` + DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` UnexpectedMessages int64 `json:"unexpected_messages"` HistoryChecked int64 `json:"history_checked"` HistoryFound int64 `json:"history_found"` @@ -62,7 +64,7 @@ func runMerge(aPath, bPath, output string) error { return fmt.Errorf("agent summaries do not belong to the same run") } report := CombinedReport{RunID: a.RunID, Scenario: a.Scenario, GeneratedAt: time.Now().UTC(), Agents: []AgentSummary{a, b}, Passed: true, - PublishErrors: make(map[string]int64)} + PublishErrors: make(map[string]int64), DisconnectReasons: make(map[string]int64)} for _, summary := range report.Agents { expectedConnections := summary.ExpectedConnections if expectedConnections == 0 { @@ -92,6 +94,15 @@ func runMerge(aPath, bPath, output string) error { report.OutOfOrder += summary.OutOfOrder report.OrderingViolations = append(report.OrderingViolations, summary.OrderingViolations...) report.Disconnects += summary.Disconnects + for category, count := range summary.DisconnectReasons { + report.DisconnectReasons[category] += count + } + if remaining := 100 - len(report.DisconnectSamples); remaining > 0 { + if remaining > len(summary.DisconnectSamples) { + remaining = len(summary.DisconnectSamples) + } + report.DisconnectSamples = append(report.DisconnectSamples, summary.DisconnectSamples[:remaining]...) + } report.UnexpectedMessages += summary.UnexpectedMessages report.HistoryChecked += summary.HistoryChecked report.HistoryFound += summary.HistoryFound diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index 644f3ca..e961208 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -14,11 +14,14 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { ExpectedConnections: 1000, PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9, UnexpectedMessages: 42, PublishErrors: map[string]int64{"ctrl_timeout": 3, "socket": 2, "503": 1}, - RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10} + RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10, Disconnects: 1, + DisconnectReasons: map[string]int64{"timeout": 1}, + DisconnectSamples: []DisconnectSample{{ReceiverMachine: "a", ReceiverIndex: 3, Category: "timeout"}}} b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000, ExpectedConnections: 1000, PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1, OutOfOrder: 1, + Disconnects: 2, DisconnectReasons: map[string]int64{"timeout": 1, "connection_reset": 1}, OrderingViolations: []OrderingViolation{{ReceiverMachine: "b", ReceiverIndex: 7, PreviousSeq: 12, CurrentSeq: 11}}} aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json") writeSummaryForTest(t, aPath, a) @@ -47,6 +50,11 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { if len(report.OrderingViolations) != 1 || report.OrderingViolations[0].ReceiverIndex != 7 { t.Fatalf("ordering violations were not merged: %+v", report.OrderingViolations) } + if report.Disconnects != 3 || report.DisconnectReasons["timeout"] != 2 || + report.DisconnectReasons["connection_reset"] != 1 || len(report.DisconnectSamples) != 1 { + t.Fatalf("disconnect diagnostics were not merged: count=%d reasons=%v samples=%+v", + report.Disconnects, report.DisconnectReasons, report.DisconnectSamples) + } } func TestMergedHistogramP95(t *testing.T) { diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index fc45bff..ba14388 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -410,6 +410,8 @@ type AgentSummary struct { OutOfOrder int64 `json:"out_of_order"` OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` Disconnects int64 `json:"disconnects"` + DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` + DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` UnexpectedMessages int64 `json:"unexpected_messages"` HistoryChecked int64 `json:"history_checked"` HistoryFound int64 `json:"history_found"` @@ -427,6 +429,14 @@ type AgentSummary struct { AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` } +type DisconnectSample struct { + ReceiverMachine string `json:"receiver_machine"` + ReceiverIndex int `json:"receiver_index"` + Category string `json:"category"` + Error string `json:"error"` + At time.Time `json:"at"` +} + type OrderingViolation struct { ReceiverMachine string `json:"receiver_machine"` ReceiverIndex int `json:"receiver_index"` diff --git a/loadtest/im-pressure/model_test.go b/loadtest/im-pressure/model_test.go index fda6d58..62532ca 100644 --- a/loadtest/im-pressure/model_test.go +++ b/loadtest/im-pressure/model_test.go @@ -24,6 +24,19 @@ func TestLoadConfigAppliesBoundedHistoryDefaults(t *testing.T) { } } +func TestSingleMachineExampleUsesSixtyPercentCapacity(t *testing.T) { + cfg, err := loadConfig("config.single-machine.example.json") + if err != nil { + t.Fatal(err) + } + if cfg.UsersPerMachine != 600 || cfg.CloseRamp != "60s" { + t.Fatalf("unexpected single-machine capacity: users=%d close_ramp=%s", cfg.UsersPerMachine, cfg.CloseRamp) + } + if len(cfg.Endpoints["a"]) != 1 || len(cfg.Endpoints["b"]) != 1 { + t.Fatalf("single-machine endpoints must define both logical agents: %+v", cfg.Endpoints) + } +} + func TestNewManifestDistributesUsersAndEndpoints(t *testing.T) { cfg := &Config{Tenant: "006", UsersPerMachine: 3, Endpoints: map[string][]string{ "a": {"ws://a1", "ws://a2"}, "b": {"ws://b1"}, diff --git a/loadtest/im-pressure/prepare.go b/loadtest/im-pressure/prepare.go index a84f994..4b1c83f 100644 --- a/loadtest/im-pressure/prepare.go +++ b/loadtest/im-pressure/prepare.go @@ -49,7 +49,7 @@ func runPrepare(ctx context.Context, cfg *Config, manifestPath string) error { if err = saveManifest(manifestPath, manifest); err != nil { return err } - if err = prepareGroup(ctx, cfg, manifest, api); err != nil { + if err = prepareGroup(ctx, cfg, manifest, api, manifestPath); err != nil { return err } if err = saveManifest(manifestPath, manifest); err != nil { @@ -184,7 +184,7 @@ func prepareFriendships(ctx context.Context, cfg *Config, manifest *Manifest, ap return err } -func prepareGroup(ctx context.Context, cfg *Config, manifest *Manifest, api *apiClient) error { +func prepareGroup(ctx context.Context, cfg *Config, manifest *Manifest, api *apiClient, manifestPath string) error { if err := parallelAccounts(ctx, manifest.Accounts, cfg.PrepareWorkers, func(account AccountRecord) error { return retryPrepareOperation(ctx, "privacy-"+account.Username, func() error { return api.disableGroupInviteVerification(ctx, account) @@ -200,6 +200,9 @@ func prepareGroup(ctx context.Context, cfg *Config, manifest *Manifest, api *api return err } manifest.GroupTopic = group.Topic + if err = saveManifest(manifestPath, manifest); err != nil { + return fmt.Errorf("checkpoint created group: %w", err) + } log.Printf("created load-test group topic_hash=%s", topicHash(group.Topic)) } memberIDs := make([]string, 0, len(manifest.Accounts)-1) @@ -214,20 +217,9 @@ func prepareGroup(ctx context.Context, cfg *Config, manifest *Manifest, api *api if end > len(memberIDs) { end = len(memberIDs) } - var response *groupInvitationResponse - err := retryPrepareOperation(ctx, fmt.Sprintf("group-invite-%03d", batch), func() error { - var inviteErr error - response, inviteErr = api.inviteGroup(ctx, owner, manifest.GroupTopic, batch, memberIDs[offset:end]) - return inviteErr - }) - if err != nil { + if err := ensureGroupMembers(ctx, api, owner, manifest.GroupTopic, batch, memberIDs[offset:end]); err != nil { return fmt.Errorf("invite group batch %d: %w", batch, err) } - for _, item := range response.Items { - if item.Status != "joined" && item.Status != "already_member" { - return fmt.Errorf("invite user %s status=%s reason=%s", item.UserID, item.Status, item.Reason) - } - } } count, err := api.countGroupMembers(ctx, owner, manifest.GroupTopic) if err != nil { @@ -240,9 +232,60 @@ func prepareGroup(ctx context.Context, cfg *Config, manifest *Manifest, api *api return nil } -func retryPrepareOperation(ctx context.Context, label string, operation func() error) error { +func ensureGroupMembers(ctx context.Context, api *apiClient, owner AccountRecord, topic string, batch int, desired []string) error { var lastErr error - for attempt := 1; attempt <= 6; attempt++ { + for attempt := 1; attempt <= 12; attempt++ { + members, err := api.groupMemberIDs(ctx, owner, topic) + if err != nil { + lastErr = fmt.Errorf("list current members: %w", err) + } else { + missing := make([]string, 0, len(desired)) + for _, userID := range desired { + if !members[userID] { + missing = append(missing, userID) + } + } + if len(missing) == 0 { + return nil + } + response, inviteErr := api.inviteGroup(ctx, owner, topic, batch, missing) + if inviteErr != nil { + lastErr = inviteErr + } else { + lastErr = nil + for _, item := range response.Items { + if item.Status != "joined" && item.Status != "already_member" { + lastErr = fmt.Errorf("invite user %s status=%s reason=%s", item.UserID, item.Status, item.Reason) + break + } + } + if lastErr == nil { + continue + } + } + } + if ctx.Err() != nil { + return ctx.Err() + } + if attempt == 12 { + break + } + delay := time.Duration(attempt*5) * time.Second + log.Printf("operation=group-invite-%03d attempt=%d failed; retry_in=%s: %v", batch, attempt, delay, lastErr) + if err := waitContext(ctx, delay); err != nil { + return err + } + } + return fmt.Errorf("missing members not recovered after retries: %w", lastErr) +} + +func retryPrepareOperation(ctx context.Context, label string, operation func() error) error { + return retryOperation(ctx, label, 6, operation) +} + +func retryOperation(ctx context.Context, label string, attempts int, operation func() error) error { + var lastErr error + for attempt := 1; attempt <= attempts; attempt++ { if err := operation(); err == nil { return nil } else { @@ -251,6 +294,9 @@ func retryPrepareOperation(ctx context.Context, label string, operation func() e if ctx.Err() != nil { return ctx.Err() } + if attempt == attempts { + break + } delay := time.Duration(attempt*5) * time.Second log.Printf("operation=%s attempt=%d failed; retry_in=%s: %v", label, attempt, delay, lastErr) if err := waitContext(ctx, delay); err != nil { diff --git a/loadtest/im-pressure/recorder.go b/loadtest/im-pressure/recorder.go index 57b2263..2465d98 100644 --- a/loadtest/im-pressure/recorder.go +++ b/loadtest/im-pressure/recorder.go @@ -2,8 +2,11 @@ package main import ( "encoding/json" + "errors" "fmt" + "io" "math/bits" + "net" "sort" "strings" "sync" @@ -11,6 +14,8 @@ import ( "time" ) +const observationShardCount = 64 + type observation struct { topic string seq int64 @@ -21,6 +26,17 @@ type observation struct { bits []uint64 } +type observationShard struct { + mu sync.RWMutex + observed map[string]*observation + history map[string]bool +} + +type sequenceState struct { + mu sync.Mutex + last map[string]int64 +} + type deliveryRecorder struct { machine string localUsers int @@ -32,26 +48,35 @@ type deliveryRecorder struct { runID atomic.Value historyMode atomic.Bool - mu sync.Mutex - observed map[string]*observation - lastSeq []map[string]int64 - duplicates int64 - outOfOrder int64 - ordering []OrderingViolation - malformed int64 - crossClockSkewSamples int64 - history map[string]bool + shards []observationShard + sequences []sequenceState + duplicates atomic.Int64 + outOfOrder atomic.Int64 + malformed atomic.Int64 + crossClockSkewSamples atomic.Int64 + + orderingMu sync.Mutex + ordering []OrderingViolation + + disconnectMu sync.Mutex + disconnectReasons map[string]int64 + disconnectSamples []DisconnectSample } func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorder { - lastSeq := make([]map[string]int64, localUsers) - for i := range lastSeq { - lastSeq[i] = make(map[string]int64) + shards := make([]observationShard, observationShardCount) + for i := range shards { + shards[i].observed = make(map[string]*observation) + shards[i].history = make(map[string]bool) + } + sequences := make([]sequenceState, localUsers) + for i := range sequences { + sequences[i].last = make(map[string]int64) } recorder := &deliveryRecorder{ machine: machine, localUsers: localUsers, latency: NewDurationHistogram(), localLatency: NewDurationHistogram(), crossLatency: NewDurationHistogram(), - observed: make(map[string]*observation), lastSeq: lastSeq, history: make(map[string]bool), + shards: shards, sequences: sequences, disconnectReasons: make(map[string]int64), } recorder.runID.Store(runID) return recorder @@ -59,7 +84,13 @@ func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorde func (r *deliveryRecorder) setRunID(runID string) { r.runID.Store(runID) } -func (r *deliveryRecorder) resetDisconnects() { r.disconnects.Store(0) } +func (r *deliveryRecorder) resetDisconnects() { + r.disconnectMu.Lock() + r.disconnects.Store(0) + r.disconnectReasons = make(map[string]int64) + r.disconnectSamples = nil + r.disconnectMu.Unlock() +} func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { runID, _ := r.runID.Load().(string) @@ -75,40 +106,50 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { SentAt int64 `json:"sent_at_unix_nano"` } if json.Unmarshal(data.Content, &content) != nil || content.RunID != runID { - r.mu.Lock() - r.malformed++ - r.mu.Unlock() + r.malformed.Add(1) return } if r.historyMode.Load() { - r.mu.Lock() - r.history[mid] = true - r.mu.Unlock() + shard := r.shard(mid) + shard.mu.Lock() + shard.history[mid] = true + shard.mu.Unlock() return } index := account.Index - 1 if index < 0 || index >= r.localUsers { return } - r.mu.Lock() - defer r.mu.Unlock() - obs := r.observed[mid] + shard := r.shard(mid) + shard.mu.Lock() + obs := shard.observed[mid] if obs == nil { 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 + shard.observed[mid] = obs } word, bit := index/64, uint(index%64) mask := uint64(1) << bit if obs.bits[word]&mask != 0 { - r.duplicates++ + shard.mu.Unlock() + r.duplicates.Add(1) return } obs.bits[word] |= mask + shard.mu.Unlock() r.received.Add(1) - if previous := r.lastSeq[index][data.Topic]; previous > 0 && data.Seq < previous { - r.outOfOrder++ + sequence := &r.sequences[index] + sequence.mu.Lock() + previous := sequence.last[data.Topic] + violation := previous > 0 && data.Seq < previous + if data.Seq > previous { + sequence.last[data.Topic] = data.Seq + } + sequence.mu.Unlock() + if violation { + r.outOfOrder.Add(1) + r.orderingMu.Lock() if len(r.ordering) < 100 { r.ordering = append(r.ordering, OrderingViolation{ ReceiverMachine: r.machine, @@ -119,9 +160,7 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { ClientMid: mid, }) } - } - if data.Seq > r.lastSeq[index][data.Topic] { - r.lastSeq[index][data.Topic] = data.Seq + r.orderingMu.Unlock() } if content.SentAt > 0 { delay := time.Since(time.Unix(0, content.SentAt)) @@ -133,7 +172,7 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { if senderMachine == r.machine { r.localLatency.Observe(delay) } else if delay < 0 { - r.crossClockSkewSamples++ + r.crossClockSkewSamples.Add(1) } else { r.crossLatency.Observe(delay) } @@ -150,26 +189,22 @@ func senderMachineFromClientMid(runID, mid string) string { } func (r *deliveryRecorder) metricCounts() (received, duplicates, outOfOrder, disconnects int64) { - r.mu.Lock() - duplicates = r.duplicates - outOfOrder = r.outOfOrder - r.mu.Unlock() - return r.received.Load(), duplicates, outOfOrder, r.disconnects.Load() + return r.received.Load(), r.duplicates.Load(), r.outOfOrder.Load(), r.disconnects.Load() } func (r *deliveryRecorder) beginHistory() { r.historyMode.Store(true) } func (r *deliveryRecorder) historyFound(mid string) bool { - r.mu.Lock() - defer r.mu.Unlock() - return r.history[mid] + shard := r.shard(mid) + shard.mu.RLock() + found := shard.history[mid] + shard.mu.RUnlock() + return found } func (r *deliveryRecorder) historyFoundAll(messageIDs []string) bool { - r.mu.Lock() - defer r.mu.Unlock() for _, mid := range messageIDs { - if !r.history[mid] { + if !r.historyFound(mid) { return false } } @@ -182,28 +217,31 @@ func (r *deliveryRecorder) unackedObservedMessages(ledger []AcceptedMessage, sce 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 + for i := range r.shards { + shard := &r.shards[i] + shard.mu.RLock() + for mid, obs := range shard.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) } - 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) + shard.mu.RUnlock() } sort.Slice(messages, func(i, j int) bool { return messages[i].ClientMid < messages[j].ClientMid }) return messages @@ -218,34 +256,88 @@ func firstObservedIndex(words []uint64) int { return 0 } -func (r *deliveryRecorder) onDisconnect(_ AccountRecord, _ error) { +func (r *deliveryRecorder) onDisconnect(account AccountRecord, err error) { + category := disconnectCategory(err) + r.disconnectMu.Lock() r.disconnects.Add(1) + r.disconnectReasons[category]++ + if len(r.disconnectSamples) < 100 { + message := "" + if err != nil { + message = err.Error() + if len(message) > 300 { + message = message[:300] + } + } + r.disconnectSamples = append(r.disconnectSamples, DisconnectSample{ + ReceiverMachine: r.machine, ReceiverIndex: account.Index, Category: category, + Error: message, At: time.Now().UTC(), + }) + } + r.disconnectMu.Unlock() +} + +func disconnectCategory(err error) string { + if err == nil { + return "unknown" + } + if errors.Is(err, io.EOF) { + return "eof" + } + if errors.Is(err, net.ErrClosed) { + return "closed_connection" + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return "timeout" + } + message := strings.ToLower(err.Error()) + switch { + case strings.Contains(message, "close 1006") || strings.Contains(message, "abnormal closure"): + return "abnormal_close" + case strings.Contains(message, "close 1000") || strings.Contains(message, "normal closure"): + return "normal_close" + case strings.Contains(message, "connection reset"): + return "connection_reset" + case strings.Contains(message, "broken pipe"): + return "broken_pipe" + case strings.Contains(message, "timeout") || strings.Contains(message, "deadline exceeded"): + return "timeout" + case strings.Contains(message, "use of closed network connection"): + return "closed_connection" + default: + return "other" + } } func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage) AgentSummary { base.DeliveryLatency = r.latency.Snapshot() base.DeliveryLatencyLocal = r.localLatency.Snapshot() base.DeliveryLatencyCross = r.crossLatency.Snapshot() + r.disconnectMu.Lock() base.Disconnects = r.disconnects.Load() + base.DisconnectReasons = cloneCountMap(r.disconnectReasons) + base.DisconnectSamples = append([]DisconnectSample(nil), r.disconnectSamples...) + r.disconnectMu.Unlock() runID, _ := r.runID.Load().(string) accepted := make(map[string]AcceptedMessage, len(ledger)) const missingLimit = 100000 - r.mu.Lock() - defer r.mu.Unlock() - base.Duplicates = r.duplicates - base.OutOfOrder = r.outOfOrder - base.CrossClockSkewSamples = r.crossClockSkewSamples + base.Duplicates = r.duplicates.Load() + base.OutOfOrder = r.outOfOrder.Load() + base.CrossClockSkewSamples = r.crossClockSkewSamples.Load() + r.orderingMu.Lock() base.OrderingViolations = append([]OrderingViolation(nil), r.ordering...) - if r.malformed > 0 { - base.Errors = append(base.Errors, fmt.Sprintf("malformed run messages=%d", r.malformed)) + r.orderingMu.Unlock() + if malformed := r.malformed.Load(); malformed > 0 { + base.Errors = append(base.Errors, fmt.Sprintf("malformed run messages=%d", malformed)) } for _, message := range ledger { if message.RunID != runID { continue } accepted[message.ClientMid] = message - obs := r.observed[message.ClientMid] + obs := r.observationSnapshot(message.ClientMid) switch message.Kind { case "group": base.ExpectedGroup += int64(r.localUsers) @@ -273,10 +365,15 @@ func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage } } } - for mid, obs := range r.observed { - if _, ok := accepted[mid]; !ok { - base.UnexpectedMessages += bitsetCount(obs.bits) + for i := range r.shards { + shard := &r.shards[i] + shard.mu.RLock() + for mid, obs := range shard.observed { + if _, ok := accepted[mid]; !ok { + base.UnexpectedMessages += bitsetCount(obs.bits) + } } + shard.mu.RUnlock() } base.RealtimeLost = (base.ExpectedPeer - base.ReceivedPeer) + (base.ExpectedGroup - base.ReceivedGroup) if len(base.Missing) == missingLimit { @@ -291,14 +388,12 @@ func (r *deliveryRecorder) countRealtimeLostStored(ledger []AcceptedMessage, sto for _, mid := range storedMessageIDs { stored[mid] = true } - r.mu.Lock() - defer r.mu.Unlock() var count int64 for _, message := range ledger { if message.RunID != runID || !stored[message.ClientMid] { continue } - obs := r.observed[message.ClientMid] + obs := r.observationSnapshot(message.ClientMid) switch message.Kind { case "group": for index := 1; index <= r.localUsers; index++ { @@ -315,6 +410,39 @@ func (r *deliveryRecorder) countRealtimeLostStored(ledger []AcceptedMessage, sto return count } +func (r *deliveryRecorder) shard(mid string) *observationShard { + var hash uint32 = 2166136261 + for i := 0; i < len(mid); i++ { + hash ^= uint32(mid[i]) + hash *= 16777619 + } + return &r.shards[hash%uint32(len(r.shards))] +} + +func (r *deliveryRecorder) observationSnapshot(mid string) *observation { + shard := r.shard(mid) + shard.mu.RLock() + obs := shard.observed[mid] + if obs != nil { + copyOfObservation := *obs + copyOfObservation.bits = append([]uint64(nil), obs.bits...) + obs = ©OfObservation + } + shard.mu.RUnlock() + return obs +} + +func cloneCountMap(source map[string]int64) map[string]int64 { + if len(source) == 0 { + return nil + } + clone := make(map[string]int64, len(source)) + for key, value := range source { + clone[key] = value + } + return clone +} + func observationHas(obs *observation, receiverIndex int) bool { if obs == nil || receiverIndex <= 0 { return false diff --git a/loadtest/im-pressure/recorder_test.go b/loadtest/im-pressure/recorder_test.go index 61a58f2..ecd9e32 100644 --- a/loadtest/im-pressure/recorder_test.go +++ b/loadtest/im-pressure/recorder_test.go @@ -2,6 +2,9 @@ package main import ( "encoding/json" + "errors" + "fmt" + "sync" "testing" "time" ) @@ -111,6 +114,65 @@ func TestDeliveryRecorderSeparatesLocalAndCrossMachineLatency(t *testing.T) { } } +func TestDeliveryRecorderConcurrentGroupFanout(t *testing.T) { + const users = 32 + const messages = 64 + recorder := newDeliveryRecorder("run1", "a", users) + ledger := make([]AcceptedMessage, messages) + for messageIndex := range ledger { + ledger[messageIndex] = AcceptedMessage{ + RunID: "run1", Kind: "group", Topic: "grp-one", Seq: 1, + ClientMid: fmt.Sprintf("run1-g-b-%04d-01", messageIndex+1), SenderMachine: "b", + } + } + + var writers sync.WaitGroup + for receiver := 1; receiver <= users; receiver++ { + for _, message := range ledger { + writers.Add(1) + go func(receiver int, message AcceptedMessage) { + defer writers.Done() + recorder.onData(AccountRecord{Index: receiver}, testData(message, time.Now().Add(-time.Millisecond))) + }(receiver, message) + } + } + writers.Wait() + + summary := recorder.summarize(AgentSummary{}, ledger) + want := int64(users * messages) + if summary.ExpectedGroup != want || summary.ReceivedGroup != want || summary.RealtimeLost != 0 { + t.Fatalf("group expected/received/lost = %d/%d/%d, want %d/%d/0", + summary.ExpectedGroup, summary.ReceivedGroup, summary.RealtimeLost, want, want) + } + if summary.Duplicates != 0 || summary.DeliveryLatency.Count != want { + t.Fatalf("duplicates=%d latency samples=%d want=0/%d", summary.Duplicates, summary.DeliveryLatency.Count, want) + } +} + +func TestDeliveryRecorderDisconnectDiagnostics(t *testing.T) { + recorder := newDeliveryRecorder("run1", "a", 1) + recorder.onDisconnect(AccountRecord{Index: 1}, errors.New("websocket: close 1006 (abnormal closure): unexpected EOF")) + recorder.onDisconnect(AccountRecord{Index: 1}, errors.New("read tcp: connection reset by peer")) + + summary := recorder.summarize(AgentSummary{}, nil) + if summary.Disconnects != 2 || summary.DisconnectReasons["abnormal_close"] != 1 || + summary.DisconnectReasons["connection_reset"] != 1 || len(summary.DisconnectSamples) != 2 { + t.Fatalf("unexpected disconnect diagnostics: count=%d reasons=%v samples=%+v", + summary.Disconnects, summary.DisconnectReasons, summary.DisconnectSamples) + } + recorder.resetDisconnects() + summary = recorder.summarize(AgentSummary{}, nil) + if summary.Disconnects != 0 || len(summary.DisconnectReasons) != 0 || len(summary.DisconnectSamples) != 0 { + t.Fatalf("disconnect diagnostics were not reset: %+v", summary) + } +} + +func TestBitsetCountCountsEachBitOnce(t *testing.T) { + if got := bitsetCount([]uint64{0b1011, 0b100000}); got != 4 { + t.Fatalf("bitset count=%d want=4", got) + } +} + func testData(message AcceptedMessage, sentAt time.Time) dataPacket { content, _ := json.Marshal(map[string]any{"run_id": message.RunID, "scenario": message.Scenario, "sender_machine": message.SenderMachine, "sender_index": message.SenderIndex, diff --git a/server/hub.go b/server/hub.go index 02f1330..2e59200 100644 --- a/server/hub.go +++ b/server/hub.go @@ -251,7 +251,11 @@ func (h *Hub) run() { immetrics.ObserveInternalQueueEnqueue("topic_client", len(dst.clientMsg), cap(dst.clientMsg), true) default: immetrics.ObserveInternalQueueEnqueue("topic_client", len(dst.clientMsg), cap(dst.clientMsg), false) - logs.Err.Println("hub: topic's broadcast queue is full", dst.name) + if msg.sess != nil { + msg.sess.queueOut(ErrServiceUnavailableReply(msg, types.TimeNow())) + } + logs.Err.Printf("hub: topic's broadcast queue is full topic=%s tenant=%d len=%d cap=%d", + dst.name, dst.tenantID, len(dst.clientMsg), cap(dst.clientMsg)) } } else { logs.Warn.Println("hub: invalid topic category for broadcast", dst.name) diff --git a/server/hub_tenant_test.go b/server/hub_tenant_test.go index 01c38aa..e538186 100644 --- a/server/hub_tenant_test.go +++ b/server/hub_tenant_test.go @@ -102,3 +102,48 @@ func TestHubRejectsUnroutedClientMessageInsteadOfAcceptingWithoutSeq(t *testing. t.Fatal("timeout waiting for unrouted message response") } } + +func TestHubReturnsServiceUnavailableWhenTopicClientQueueIsFull(t *testing.T) { + h := &Hub{ + topics: &sync.Map{}, + routeCli: make(chan *ClientComMessage, 1), + shutdown: make(chan chan<- bool), + routeSrv: make(chan *ServerComMessage), + join: make(chan *ClientComMessage), + meta: make(chan *ClientComMessage), + unreg: make(chan *topicUnreg), + userStatus: make(chan *userStatusReq), + rehash: make(chan bool), + } + topic := &Topic{ + tenantID: 7, name: "grpBusy", clientMsg: make(chan *ClientComMessage, 1), + exit: make(chan *shutDown, 1), + } + topic.clientMsg <- &ClientComMessage{Id: "occupy"} + h.topicPut(types.TopicKey{TenantID: 7, Topic: topic.name}, topic) + + go h.run() + done := make(chan bool) + defer func() { + h.topicDel(types.TopicKey{TenantID: 7, Topic: topic.name}) + h.shutdown <- done + <-done + }() + + sess := &Session{tenantID: 7, send: make(chan any, 1)} + now := time.Now().UTC() + h.routeCli <- &ClientComMessage{ + Id: "pub-overload", TenantID: 7, RcptTo: topic.name, Original: topic.name, + Timestamp: now, sess: sess, + } + + select { + case raw := <-sess.send: + msg, ok := raw.(*ServerComMessage) + if !ok || msg.Ctrl == nil || msg.Ctrl.Code != http.StatusServiceUnavailable { + t.Fatalf("unexpected response: %#v", raw) + } + case <-time.After(time.Second): + t.Fatal("timeout waiting for overload response") + } +} diff --git a/server/im_http.go b/server/im_http.go index bd1dc97..0ed06a4 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -4079,6 +4079,12 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, acceptErr) return } + if accepted == nil { + logs.Err.Printf("friend request auto-accept returned an empty result: tenant=%d request=%s actor=%s", + principal.TenantID, created.RequestID, target.String()) + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + return + } created = accepted } if imReliableHTTPEventsEnabled(req) { @@ -6391,6 +6397,12 @@ func serveIMFriendRequestAction(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, err) return } + if updated == nil { + logs.Err.Printf("friend request resolution returned an empty result: tenant=%d request=%s actor=%s status=%s", + principal.TenantID, requestID, principal.Uid.String(), status) + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + return + } if status == types.FriendRequestAccepted && imReliableHTTPEventsEnabled(req) { topic := updated.ApplicantID.P2PName(updated.TargetID) publishIMConversationSystemNotice(principal.TenantID, topic, diff --git a/server/im_http_test.go b/server/im_http_test.go index 400f28a..c0ff3e9 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -158,6 +158,7 @@ type testRelationsStore struct { deleteTenant types.TenantID deleteUser types.Uid deletePeer types.Uid + resolveNil bool } type testFavoritesStore struct { @@ -962,6 +963,9 @@ func (s *testRelationsStore) CountPendingFriendRequests(tenantID types.TenantID, } func (s *testRelationsStore) ResolveFriendRequest(_ types.TenantID, requestID string, actor types.Uid, status string, handledAt time.Time) (*types.FriendRequest, error) { + if s.resolveNil { + return nil, nil + } return &types.FriendRequest{ TenantID: 7, RequestID: requestID, @@ -975,6 +979,28 @@ func (s *testRelationsStore) ResolveFriendRequest(_ types.TenantID, requestID st }, nil } +func TestServeIMFriendRequestActionRejectsEmptyStoreResult(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + expectIMBearer(t, ctrl, "token-1", 7, 42) + + oldRelations := store.Relations + store.Relations = &testRelationsStore{resolveNil: true} + t.Cleanup(func() { store.Relations = oldRelations }) + + req := httptest.NewRequest(http.MethodPost, "/v1/im/friend-requests/frq-1/accept", nil) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMFriendRequestAction(rec, req) + + if rec.Code != http.StatusInternalServerError { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusInternalServerError, rec.Body.String()) + } + if response := decodeIMHTTPResponse(t, rec); response.Reason != imReasonInternal { + t.Fatalf("reason: got %q, want %q", response.Reason, imReasonInternal) + } +} + func (s *testRelationsStore) ListFriends(_ types.TenantID, _ types.Uid, _, _, _ string, _ int) ([]types.FriendListItem, string, error) { return s.friends, "", nil } diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 417bc0f..76a24ad 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -3,6 +3,7 @@ package store import ( "context" "encoding/json" + "errors" "strconv" "strings" "testing" @@ -443,6 +444,34 @@ func TestInvalidateGroupSettingsUsesTopicCacheKey(t *testing.T) { } } +func TestGetGroupMuteUsesPositiveAndNegativeCache(t *testing.T) { + oldRuntime := groupMsgRuntime + runtime := newTestGroupMessageRuntime() + groupMsgRuntime = runtime + t.Cleanup(func() { groupMsgRuntime = oldRuntime }) + + tenantID := types.TenantID(7) + topic := "grpAlpha" + uid := types.Uid(42) + cacheID := groupMemberCacheID(topic, uid) + expires := time.Now().UTC().Add(time.Minute) + want := &types.GroupMute{TenantID: tenantID, Topic: topic, UserID: uid, Status: types.GroupMuteStatusActive, ExpiresAt: &expires} + if err := CacheObject(tenantID, cacheNamespaceGroupMute, cacheID, groupMuteCacheValue{Found: true, Mute: want}); err != nil { + t.Fatalf("cache positive mute: %v", err) + } + got, err := (groupBusinessMapper{}).GetMute(tenantID, topic, uid) + if err != nil || got == nil || got.UserID != uid || got.Status != types.GroupMuteStatusActive { + t.Fatalf("positive cache result: mute=%#v err=%v", got, err) + } + + if err = CacheObject(tenantID, cacheNamespaceGroupMute, cacheID, groupMuteCacheValue{}); err != nil { + t.Fatalf("cache negative mute: %v", err) + } + if got, err = (groupBusinessMapper{}).GetMute(tenantID, topic, uid); !errors.Is(err, types.ErrNotFound) || got != nil { + t.Fatalf("negative cache result: mute=%#v err=%v", got, err) + } +} + func TestTopicAndSubscriptionCacheTTLIsShort(t *testing.T) { runtime := &redisGroupMessageRuntime{cfg: redisConfig{CacheTTL: 300, ShortTTL: 600}} for _, namespace := range []string{cacheNamespaceTopic, cacheNamespaceSub} { diff --git a/server/store/store.go b/server/store/store.go index f8d6f8e..e353a81 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -142,6 +142,7 @@ const ( cacheNamespaceSub = "sub" cacheNamespaceGroupSettings = "group_settings" cacheNamespaceGroupMember = "group_member:v2" + cacheNamespaceGroupMute = "group_mute:v1" cacheNamespaceVerification = "verification" cacheNamespaceActionToken = "action_token" cacheNamespaceFriends = "friends" @@ -2422,11 +2423,32 @@ func (groupBusinessMapper) GetMute(tenantID types.TenantID, topic string, uid ty if strings.TrimSpace(topic) == "" || uid.IsZero() { return nil, types.ErrMalformed } + topic = strings.TrimSpace(topic) + cacheID := groupMemberCacheID(topic, uid) + var cached groupMuteCacheValue + if ok, err := GetCachedObject(tenantID, cacheNamespaceGroupMute, cacheID, &cached); err != nil { + logs.Warn.Printf("cache group mute[%s]: read failed: %v", cacheID, err) + _ = DeleteCachedObject(tenantID, cacheNamespaceGroupMute, cacheID) + } else if ok { + if !cached.Found || cached.Mute == nil { + return nil, types.ErrNotFound + } + if cached.Mute.ExpiresAt == nil || cached.Mute.ExpiresAt.After(time.Now().UTC()) { + return cached.Mute, nil + } + _ = DeleteCachedObject(tenantID, cacheNamespaceGroupMute, cacheID) + } tenantAdp, err := tenantBusinessAdapter(tenantID) if err != nil { return nil, err } - return tenantAdp.TenantGroupMuteGet(tenantID, strings.TrimSpace(topic), uid) + mute, err := tenantAdp.TenantGroupMuteGet(tenantID, topic, uid) + if err == nil && mute != nil { + _ = CacheObject(tenantID, cacheNamespaceGroupMute, cacheID, groupMuteCacheValue{Found: true, Mute: mute}) + } else if errors.Is(err, types.ErrNotFound) { + _ = CacheObject(tenantID, cacheNamespaceGroupMute, cacheID, groupMuteCacheValue{}) + } + return mute, err } func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.GroupMute) (*types.GroupMute, error) { @@ -2440,6 +2462,12 @@ func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.Group updated, err := tenantAdp.TenantGroupMuteUpsert(tenantID, mute) if err == nil { invalidateGroupMemberPages(tenantID, mute.Topic) + cacheID := groupMemberCacheID(mute.Topic, mute.UserID) + if updated != nil { + _ = CacheObject(tenantID, cacheNamespaceGroupMute, cacheID, groupMuteCacheValue{Found: true, Mute: updated}) + } else { + _ = DeleteCachedObject(tenantID, cacheNamespaceGroupMute, cacheID) + } } return updated, err } @@ -2456,6 +2484,7 @@ func (groupBusinessMapper) DeleteMute(tenantID types.TenantID, topic string, uid return err } invalidateGroupMemberPages(tenantID, topic) + _ = CacheObject(tenantID, cacheNamespaceGroupMute, groupMemberCacheID(topic, uid), groupMuteCacheValue{}) return nil } @@ -2581,11 +2610,32 @@ func (groupBusinessMapper) CanPost(tenantID types.TenantID, topic string, uid ty if strings.TrimSpace(topic) == "" || uid.IsZero() || now.IsZero() { return false, types.ErrMalformed } - tenantAdp, err := tenantBusinessAdapter(tenantID) + if admin { + return true, nil + } + topic = strings.TrimSpace(topic) + mapper := groupBusinessMapper{} + settings, err := mapper.GetSettings(tenantID, topic) + if err != nil && !errors.Is(err, types.ErrNotFound) { + return false, err + } + if settings != nil && settings.GroupStatus == types.GroupStatusActive && settings.MuteAll { + return false, nil + } + mute, err := mapper.GetMute(tenantID, topic, uid) + if errors.Is(err, types.ErrNotFound) { + return true, nil + } if err != nil { return false, err } - return tenantAdp.TenantGroupCanPost(tenantID, strings.TrimSpace(topic), uid, admin, now) + return mute == nil || mute.Status != types.GroupMuteStatusActive || + (mute.ExpiresAt != nil && !mute.ExpiresAt.After(now)), nil +} + +type groupMuteCacheValue struct { + Found bool `json:"found"` + Mute *types.GroupMute `json:"mute,omitempty"` } func groupMemberCacheID(topic string, uid types.Uid) string { diff --git a/server/topic.go b/server/topic.go index 9b0a7fe..f1ee8bb 100644 --- a/server/topic.go +++ b/server/topic.go @@ -1348,7 +1348,9 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, if t.cat == types.TopicCatGrp && msg.sess != nil && msg.sess.authLvl != auth.LevelRoot && store.GroupBusiness != nil && store.Store != nil && store.Store.IsOpen() { access := pud.modeWant & pud.modeGiven + started := time.Now() canPost, err := store.GroupBusiness.CanPost(t.tenantID, t.name, asUid, access.IsAdmin(), msg.Timestamp) + immetrics.ObserveMessagePublishStage("group", "permission_check", started, err) if err != nil { logs.Warn.Printf("topic[%s]: failed to check group mute policy: %v", t.name, err) msg.sess.queueOut(ErrUnknown(msg.Id, t.original(asUid), msg.Timestamp)) @@ -1457,8 +1459,10 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, // Tell the plugins that a message was accepted for delivery pluginMessage(t.tenantID, data.Data, plgActCreate) + fanoutStarted := time.Now() t.broadcastToSessions(data) publishRealtimeTopicMessage(t.name, data) + immetrics.ObserveMessagePublishStage(kind, "realtime_fanout", fanoutStarted, nil) // sendPush will update unread message count and send push notification. if pushRcpt := t.pushForData(asUid, data.Data, markedReadBySender); pushRcpt != nil {