fix: align private message ids with snowflake

2026-09-27 02:53:25 +08:00
parent c8d08e26e2
commit ae2dc0d45e
3 changed files with 122 additions and 5 deletions
@@ -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;
@@ -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 恢复前，本轮数据不能视为已持久化。
@@ -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,