From 43808c36d1a302a9bafe65a13f2a64ba1746a2f4 Mon Sep 17 00:00:00 2001 From: Austin Date: Tue, 13 Oct 2026 17:07:41 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=B0=86im=5Ffilemsglinks.msgid?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E4=BB=8EINT=E6=94=B9=E4=B8=BABIGINT=E9=80=82?= =?UTF-8?q?=E9=85=8D=E9=9B=AA=E8=8A=B1ID?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复msgid为INT类型无法容纳雪花ID的问题，避免插入时数值溢出导致Kafka消费分区停滞 同时新增数据库 schema 校验逻辑，提前发现不符合要求的字段类型 --- .../IM-文件消息链接表MsgId-BIGINT修复SQL.sql | 44 +++++++ scripts/IM-租户业务库初始化.sql | 2 +- server/db/mysql/tenant_adapter.go | 113 ++++++++++++++---- 3 files changed, 133 insertions(+), 26 deletions(-) create mode 100644 docs/sql/IM-文件消息链接表MsgId-BIGINT修复SQL.sql diff --git a/docs/sql/IM-文件消息链接表MsgId-BIGINT修复SQL.sql b/docs/sql/IM-文件消息链接表MsgId-BIGINT修复SQL.sql new file mode 100644 index 0000000..ae8f699 --- /dev/null +++ b/docs/sql/IM-文件消息链接表MsgId-BIGINT修复SQL.sql @@ -0,0 +1,44 @@ +-- IM file-message link msgid Snowflake compatibility migration. +-- Scope: the shared business database (for example meeting_im). +-- Fixes: im_filemsglinks.msgid was INT (max 2147483647) while message ids are +-- snowflake int64 values (~1.58e18), so every INSERT into the link table +-- failed with MySQL error 1264 "Out of range value for column 'msgid'". +-- Under Kafka persistence that failure retried forever and stalled the +-- consumer partition, blocking persistence for every message queued +-- behind it. +-- Notes: im_filemsglinks is a global (non tenant-sharded) table, but in +-- tenant-scoped-DB deployments every tenant database has its own copy — +-- run this script once per database. +-- Re-running is safe (MODIFY to the same shape is idempotent). +-- NOT REVERSIBLE: converting back to INT would truncate snowflake ids. +-- Review and back up before execution. This script does not replay dead letters. + +-- 1. Preflight: confirm the live column shape before touching anything. +SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_TYPE, IS_NULLABLE +FROM information_schema.COLUMNS +WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'im_filemsglinks' + AND COLUMN_NAME = 'msgid'; + +-- 2. Upgrade the column (keeps the nullable semantics of the original schema). +ALTER TABLE `im_filemsglinks` + MODIFY COLUMN `msgid` BIGINT DEFAULT NULL; + +-- 3. Post-check: this query must return zero rows. +SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_TYPE +FROM information_schema.COLUMNS +WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'im_filemsglinks' + AND COLUMN_NAME = 'msgid' + AND DATA_TYPE <> 'bigint'; + +-- 4. Observe pending recovery work caused by the overflow (and, before the +-- startup schema-contract check existed, by any other drift). Replay only +-- after the post-check passes and the referenced files still exist. +SELECT status, last_reason, COUNT(*) AS entry_count, + MIN(created_at) AS first_created_at, MAX(created_at) AS last_created_at +FROM im_kafka_dead_letters +WHERE last_error LIKE '%Out of range value for column %' + OR last_error LIKE '%file lifecycle enqueue failed: not found%' +GROUP BY status, last_reason +ORDER BY status, last_reason; diff --git a/scripts/IM-租户业务库初始化.sql b/scripts/IM-租户业务库初始化.sql index ca10f77..335e316 100644 --- a/scripts/IM-租户业务库初始化.sql +++ b/scripts/IM-租户业务库初始化.sql @@ -467,7 +467,7 @@ CREATE TABLE IF NOT EXISTS `im_filemsglinks` ( `id` int NOT NULL AUTO_INCREMENT, `createdat` datetime(3) NOT NULL, `fileid` bigint NOT NULL, - `msgid` int DEFAULT NULL, + `msgid` bigint DEFAULT NULL, `topic` char(25) DEFAULT NULL, `userid` bigint DEFAULT NULL, PRIMARY KEY (`id`), diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 2214828..c7d458c 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -2259,57 +2259,120 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error { return err } +// schemaColumnContract describes the required shape of one column. +type schemaColumnContract struct { + dataType string + notNull bool + autoInc bool +} + +func (c schemaColumnContract) describe() string { + out := strings.ToUpper(c.dataType) + if c.notNull { + out += " NOT NULL" + } + if c.autoInc { + out += " AUTO_INCREMENT" + } + return out +} + +// TenantMessageSchemaValidate verifies the storage contracts asynchronous +// persistence depends on: message ids and topic sequence counters are written +// as explicit snowflake int64 values, and file-message links reference them. +// A column that is still INT overflows on the first snowflake write and, under +// Kafka persistence, stalls the consumer partition instead of failing fast. func (a *adapter) TenantMessageSchemaValidate(tenantIDs []t.TenantID) error { if len(tenantIDs) == 0 { return errors.New("mysql: no active tenants available for message schema validation") } + // Message ids are explicit snowflake values stored in the auto-increment + // id column; sequence counters and file ids are snowflake-derived too. + var ( + bigintPK = schemaColumnContract{dataType: "bigint", notNull: true, autoInc: true} + bigintReq = schemaColumnContract{dataType: "bigint", notNull: true} + bigintOpt = schemaColumnContract{dataType: "bigint"} + ) + type scopedCheck struct { + table string + column string + want schemaColumnContract + } + perTenant := func(tenantID t.TenantID, scopedDB bool) []scopedCheck { + msgTable, topicsTable, subsTable := tenantMessagesTable(tenantID), tenantTopicsTable(tenantID), tenantSubscriptionsTable(tenantID) + if scopedDB { + msgTable, topicsTable, subsTable = "im_messages", "im_topics", "im_subscriptions" + } + return []scopedCheck{ + {msgTable, "id", bigintPK}, + {topicsTable, "seqid", bigintReq}, + {subsTable, "recvseqid", bigintOpt}, + {subsTable, "readseqid", bigintOpt}, + // Global file tables exist once per database and are shared by all tenants. + {"im_filemsglinks", "fileid", bigintReq}, + {"im_filemsglinks", "msgid", bigintOpt}, + {"im_fileuploads", "id", bigintReq}, + } + } validated := make(map[*adapter]map[string]struct{}) + run := func(tenantAdp *adapter, chk scopedCheck) error { + key := chk.table + "." + chk.column + if _, ok := validated[tenantAdp][key]; ok { + return nil + } + if err := tenantAdp.validateColumnContract(chk.table, chk.column, chk.want); err != nil { + return err + } + if validated[tenantAdp] == nil { + validated[tenantAdp] = make(map[string]struct{}) + } + validated[tenantAdp][key] = struct{}{} + return nil + } for _, tenantID := range tenantIDs { tenantAdp, err := a.routedTenantAdapter(tenantID) if err != nil { return fmt.Errorf("mysql: resolve tenant %d for message schema validation: %w", tenantID, err) } - table := tenantMessagesTable(tenantID) - if tenantAdp.tenantScopedDB { - table = "im_messages" + for _, chk := range perTenant(tenantID, tenantAdp.tenantScopedDB) { + if err = run(tenantAdp, chk); err != nil { + return err + } } - if validated[tenantAdp] == nil { - validated[tenantAdp] = make(map[string]struct{}) - } - if _, ok := validated[tenantAdp][table]; ok { - continue - } - if err = tenantAdp.validateMessageIDColumn(table); err != nil { - return err - } - validated[tenantAdp][table] = struct{}{} } if !a.tenantScopedDB { - if err := a.validateMessageIDColumn("im_messages"); err != nil { - return err + for _, chk := range perTenant(0, true) { + if err := run(a, chk); err != nil { + return err + } } } return nil } -func (a *adapter) validateMessageIDColumn(table string) error { - if !regexp.MustCompile(`^im_(?:[0-9]+_)?messages$`).MatchString(table) { - return fmt.Errorf("mysql: unsafe message table name %q", table) +// validateColumnContract fails when the live column shape does not match the +// contract. The caller turns the error into a fatal startup failure so a +// drifted schema is caught before the first write instead of as a stuck +// persistence partition. +func (a *adapter) validateColumnContract(table, column string, want schemaColumnContract) error { + if !regexp.MustCompile(`^im_(?:[0-9]{3}_)?[a-z_]+$`).MatchString(table) { + return fmt.Errorf("mysql: unsafe table name %q for schema validation", table) } var dataType, nullable, extra string err := a.db.QueryRow(`SELECT DATA_TYPE,IS_NULLABLE,EXTRA FROM information_schema.COLUMNS - WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND COLUMN_NAME='id'`, table). + WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND COLUMN_NAME=?`, table, column). Scan(&dataType, &nullable, &extra) if err == sql.ErrNoRows { - return fmt.Errorf("mysql: required message table %s is missing", table) + return fmt.Errorf("mysql: required table %s is missing", table) } if err != nil { - return fmt.Errorf("mysql: inspect %s.id: %w", table, err) + return fmt.Errorf("mysql: inspect %s.%s: %w", table, column, err) } - if !strings.EqualFold(dataType, "bigint") || !strings.EqualFold(nullable, "NO") || - !strings.Contains(strings.ToLower(extra), "auto_increment") { - return fmt.Errorf("mysql: incompatible %s.id schema: type=%s nullable=%s extra=%s; want BIGINT NOT NULL AUTO_INCREMENT", - table, dataType, nullable, extra) + if !strings.EqualFold(dataType, want.dataType) || want.notNull != strings.EqualFold(nullable, "NO") || + (want.autoInc && !strings.Contains(strings.ToLower(extra), "auto_increment")) { + return fmt.Errorf("mysql: incompatible %s.%s schema: type=%s nullable=%s extra=%s; want %s "+ + "(apply scripts/IM-租户业务库初始化.sql equivalents or the matching BIGINT migration before starting)", + table, column, dataType, nullable, extra, want.describe()) } return nil }