From ae2dc0d45ef699e2e5eecbaf166012bf7a564432 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 27 Sep 2026 02:53:10 +0800 Subject: [PATCH] fix: align private message ids with snowflake --- .../IM-消息表Snowflake-ID-BIGINT修复SQL.sql | 69 +++++++++++++++++++ ...压测Pod日志根因分析与改进方案-20260927.md | 56 +++++++++++++-- scripts/IM-租户业务库初始化.sql | 2 +- 3 files changed, 122 insertions(+), 5 deletions(-) create mode 100644 docs/sql/IM-消息表Snowflake-ID-BIGINT修复SQL.sql diff --git a/docs/sql/IM-消息表Snowflake-ID-BIGINT修复SQL.sql b/docs/sql/IM-消息表Snowflake-ID-BIGINT修复SQL.sql new file mode 100644 index 0000000..8ba0230 --- /dev/null +++ b/docs/sql/IM-消息表Snowflake-ID-BIGINT修复SQL.sql @@ -0,0 +1,69 @@ +-- IM private-message Snowflake ID compatibility migration. +-- Scope: the shared business database (for example meeting_im). +-- Review and back up before execution. This script does not replay dead letters. + +-- 1. Preflight: every listed id column must become BIGINT. +SELECT TABLE_NAME, COLUMN_TYPE, IS_NULLABLE, EXTRA +FROM information_schema.COLUMNS +WHERE TABLE_SCHEMA = DATABASE() + AND (TABLE_NAME = 'im_messages' OR TABLE_NAME REGEXP '^im_[0-9]+_messages$') + AND COLUMN_NAME = 'id' +ORDER BY TABLE_NAME; + +-- 2. Upgrade the template/base table used by new tenant shards. +ALTER TABLE `im_messages` + MODIFY COLUMN `id` BIGINT NOT NULL AUTO_INCREMENT; + +-- 3. Upgrade every existing tenant private-message shard. +DELIMITER $$ + +DROP PROCEDURE IF EXISTS migrate_private_message_id_to_bigint$$ +CREATE PROCEDURE migrate_private_message_id_to_bigint() +BEGIN + DECLARE done INT DEFAULT 0; + DECLARE shard_table VARCHAR(64); + DECLARE shard_cursor CURSOR FOR + SELECT TABLE_NAME + FROM information_schema.TABLES + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME REGEXP '^im_[0-9]+_messages$' + ORDER BY TABLE_NAME; + DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1; + + OPEN shard_cursor; + shard_loop: LOOP + FETCH shard_cursor INTO shard_table; + IF done = 1 THEN + LEAVE shard_loop; + END IF; + SET @alter_message_id_sql = CONCAT( + 'ALTER TABLE `', REPLACE(shard_table, '`', '``'), + '` MODIFY COLUMN `id` BIGINT NOT NULL AUTO_INCREMENT' + ); + PREPARE alter_message_id_stmt FROM @alter_message_id_sql; + EXECUTE alter_message_id_stmt; + DEALLOCATE PREPARE alter_message_id_stmt; + END LOOP; + CLOSE shard_cursor; +END$$ + +CALL migrate_private_message_id_to_bigint()$$ +DROP PROCEDURE migrate_private_message_id_to_bigint$$ + +DELIMITER ; + +-- 4. Post-check: this query must return zero rows. +SELECT TABLE_NAME, COLUMN_TYPE, IS_NULLABLE, EXTRA +FROM information_schema.COLUMNS +WHERE TABLE_SCHEMA = DATABASE() + AND (TABLE_NAME = 'im_messages' OR TABLE_NAME REGEXP '^im_[0-9]+_messages$') + AND COLUMN_NAME = 'id' + AND DATA_TYPE <> 'bigint'; + +-- 5. Observe pending recovery work. Replay only after the post-check passes. +SELECT status, last_reason, COUNT(*) AS message_count, + MIN(created_at) AS first_created_at, MAX(created_at) AS last_created_at +FROM im_kafka_dead_letters +WHERE event_type = 'message.persist.p2p' +GROUP BY status, last_reason +ORDER BY status, last_reason; diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index 39cc5e1..b933c9c 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -4,7 +4,12 @@ ## 1. 结论 -`p2p-20260927-01` 不通过的主因已定位为 **Tinode 原生 Cluster 转发能力在突发流量下耗尽**，不是 Pod 崩溃、Pod 重启或集群节点断开。 +`p2p-20260927-01` 暴露了两个相互独立的 P0 故障： + +1. **Tinode 原生 Cluster 转发能力在突发流量下耗尽**，造成 8133 次 publish 拒绝或超时。 +2. **Kafka 异步落库写入 Snowflake ID 时与 MySQL 32 位 `INT` 字段不兼容**，造成客户端已 ACK 的消息没有落入 MySQL。 + +两者都不是 Pod 崩溃、Pod 重启或集群节点断开导致。实时投递接近成功不代表持久化成功，本轮必须同时判定 Cluster 投递和 MySQL 落库不通过。 直接故障链路： @@ -14,7 +19,7 @@ 4. Topic proxy 将该错误转换为 WebSocket `502 cluster unreachable`。 5. 反向投递进入 `hub.routeSrv` 时，4096 容量队列也出现满载；当前 RPC 调用方忽略 `rejected` 结果，存在静默丢弃路径。 -因此，本轮 `40.665%` 的发送失败率是 Cluster 内部排队与串行 RPC 的结果。Redis/Kafka 没有参与这条 Topic proxy-to-master 请求路径，不是本次故障点。 +因此，本轮 `40.665%` 的发送失败率是 Cluster 内部排队与串行 RPC 的结果。Redis/Kafka 没有参与这条 Topic proxy-to-master 请求路径，不是发送失败的故障点；但 Kafka 消费到 MySQL 的异步持久化链路另有独立故障。 ## 2. 压测结果对照 @@ -35,6 +40,22 @@ 8 条丢失全部集中在 `B -> A` 用户序号 0959，与 A 机记录到的 1 次 WS 断连同时出现。 +### 2.1 MySQL 直接核验 + +2026-09-27 通过 Pod 当前的只读数据库连接，对租户 `allrich`（`tenant_id=8`）进行了两层核验： + +| 核验项 | 结果 | +|---|---:| +| 客户端收到 ACK | 11867 | +| `im_008_messages` 按本次 `client_msg_id` 前缀命中 | 0 | +| `im_008_messages` 按 `content.run_id=p2p-20260927-01` 命中 | 0 | +| `im_messages` 按 `content.run_id=p2p-20260927-01` 命中 | 0 | +| ACK 消息精确匹配 `client_mid + topic + seq` | 0/11867 | +| 本轮已生成 Kafka 死信（核验时点） | 27 | +| 死信错误 | `Error 1264: Out of range value for column 'id'` | + +结论：本轮 ACK 消息的 **当前 MySQL 落库率为 0%**。27 是核验时已完成重试并转入死信的数量，不是最终总数；消费者正在逐条重试失败消息，其余消息仍可能位于 Kafka backlog 或处理中。 + ## 3. Kubernetes 现状 | 项目 | 实际值 | @@ -94,6 +115,21 @@ Pod 没有 OOM/Kill，内存 cgroup 没有 `oom` 或 `oom_kill`。未观察到 每个 Pod 的 CPU limit 只有 500m，同时承担 1000 量级 WS、Topic goroutine、net/rpc、MySQL 持久化和实时投递。当前 cgroup 累计存在 CPU throttling，但没有压测时间窗的 Prometheus 历史样本，所以只能将 CPU 限制定义为高概率放大因素，不将其单独认定为根因。 +### 5.5 Kafka 异步落库主键类型不兼容 + +当前代码在 Kafka 入队前通过 `store.Store.GetUid()` 生成 Snowflake 消息 ID。消费者批量落库时，`TenantMessageSaveBatch` 显式把该 ID 写入 `im_008_messages.id`。 + +线上实际字段类型为： + +| 表 | `id` 类型 | +|---|---| +| `im_messages` | `INT` | +| `im_008_messages` | `INT` | + +Snowflake ID 超过有符号 32 位 `INT` 上限，批量插入立即报 1264。代码降级为逐条处理时仍调用同一个批量写入方法，因此不会绕过该错误；达到重试上限后消息进入 `im_kafka_dead_letters`。与此同时，WebSocket 202/ACK 在 Kafka publish 成功后就返回，因而客户端会观察到“ACK 成功、实时广播成功、历史和数据库没有消息”。 + +这是数据库 DDL 与当前异步持久化代码契约不一致。仓库初始化 SQL 也仍把 `im_messages.id` 定义为 `INT`，因此新环境同样会复现。 + ## 6. 次要问题 ### 6.1 403 permission denied @@ -102,7 +138,7 @@ Pod 没有 OOM/Kill，内存 cgroup 没有 `oom` 或 `oom_kill`。未观察到 ### 6.2 历史回查程序串行阻塞 -压测程序对 1000 个 Topic 串行调用 history GET，每次控制应答最长等 20 秒，导致收尾耗时达 13-19 分钟。`history_found=0` 只能说本轮回查无效，不能据此认定 MySQL 丢数据。 +压测程序对 1000 个 Topic 串行调用 history GET，每次控制应答最长等 20 秒，导致收尾耗时达 13-19 分钟。原始 `history_found=0` 本身不能证明 MySQL 丢数据；本报告新增的 MySQL 直查与 Kafka 死信证据已经独立确认落库失败。 ### 6.3 日志保留不足 @@ -119,6 +155,16 @@ Pod 没有 OOM/Kill，内存 cgroup 没有 `oom` 或 `oom_kill`。未观察到 5. 对已持久化的 ServerComMessage 建立有界重试（例如 3 次指数退避）。超过重试上限要记录可检索的 tenant/topic/seq/target pod，并由客户端重连后历史同步补齐。 6. 将 overload 与 node unreachable 分开：队列满返回 503/`cluster overloaded`，只有节点断连才返回 502/`cluster unreachable`。 +### P0：恢复 Kafka 消息持久化 + +1. 暂停新的压力流量，避免继续扩大 Kafka backlog 和死信数量。 +2. 审核并执行 `docs/sql/IM-消息表Snowflake-ID-BIGINT修复SQL.sql`，把基础表及所有 `im_NNN_messages.id` 改为 `BIGINT NOT NULL AUTO_INCREMENT`。保留 `AUTO_INCREMENT` 是为了兼容 Kafka 关闭时不显式传 ID 的同步写入路径。 +3. 变更后先发 1 条带唯一 `client_mid` 的探针消息，确认 Kafka 消费、MySQL 行、Topic seq 和历史查询一致。 +4. 再通过现有死信管理接口重放本轮 `message.persist.p2p` 死信；Kafka 中尚未提交的 backlog 会由消费者继续处理。不要直接删除死信或重置 offset。 +5. 以本轮 11867 个 ACK ledger 做最终精确核对，要求 `client_mid + topic + seq` 为 11867/11867，重复为 0。 +6. 增加启动前 schema contract 检查：Kafka 持久化启用时，若私聊消息基础表或任何租户分表的 `id` 不是 BIGINT，Pod readiness 必须失败并输出明确的表名和字段类型。 +7. 增加 `kafka_message_persist_total{result=batch_ok|single_ok|retry|dead}`、消费 lag、死信累计量和最老 backlog age 告警。客户端 ACK 语义应明确为“Kafka 已接收（pending）”，服务端另行提供最终持久化状态；若产品要求 ACK 等于已落库，则必须改为同步确认或增加持久化回执。 + ### P0：调整部署资源与入口 1. 测试环境首轮将每 Pod request 调整为 `cpu: 1, memory: 512Mi`，limit 调整为 `cpu: 2, memory: 1Gi`；运行阶梯压测后再根据 P95 调整。 @@ -166,6 +212,7 @@ Pod 没有 OOM/Kill，内存 cgroup 没有 `oom` 或 `oom_kill`。未观察到 - Cluster enqueue timeout、hub full、peer rejected 均为 0。 - ACK P95 < 500ms，对端投递 P95 < 1s。 - MySQL 落库校验 100%。 + - Kafka consumer lag 在发送停止后 60 秒内归零，死信新增为 0。 - 单 Pod CPU P95 < 70%，内存 P95 < limit 的 70%，无持续 CPU throttling。 ## 9. 本次调查限制 @@ -173,5 +220,6 @@ Pod 没有 OOM/Kill，内存 cgroup 没有 `oom` 或 `oom_kill`。未观察到 - 调试 ServiceAccount 可读 Pod/StatefulSet/log，但不可读 Service/Endpoint 和 metrics-server 历史数据。 - `im-server-1/2` 的压测窗口日志已部分轮转，日志次数只能作为下界。 - 未获得当时的 CPU、内存、网络、MySQL、Redis、Kafka 时序监控，不对这些组件的历史利用率做无证据推断。 +- Kafka backlog 的精确条数无法通过当前调试 ServiceAccount 直接读取；死信 27 条只代表核验时已经完成失败重试的部分。 -即使存在上述限制，`cluster: load exceeded` 与 `cluster Route: server busy` 已与硬编码队列/超时路径完整对应，足以确认 Cluster 转发过载是本轮的主故障。 +即使存在上述限制，现有证据已足以分别确认 Cluster 转发过载和 Kafka 持久化主键溢出。修复并完成死信/backlog 恢复前，本轮数据不能视为已持久化。 diff --git a/scripts/IM-租户业务库初始化.sql b/scripts/IM-租户业务库初始化.sql index 0ffb286..ca10f77 100644 --- a/scripts/IM-租户业务库初始化.sql +++ b/scripts/IM-租户业务库初始化.sql @@ -1104,7 +1104,7 @@ CREATE TABLE IF NOT EXISTS `im_message_revisions` ( ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; CREATE TABLE IF NOT EXISTS `im_messages` ( - `id` int NOT NULL AUTO_INCREMENT, + `id` bigint NOT NULL AUTO_INCREMENT, `createdat` datetime(3) NOT NULL, `updatedat` datetime(3) NOT NULL, `deletedat` datetime(3) DEFAULT NULL,