fix: 将im_filemsglinks.msgid字段从INT改为BIGINT适配雪花ID
修复msgid为INT类型无法容纳雪花ID的问题,避免插入时数值溢出导致Kafka消费分区停滞 同时新增数据库 schema 校验逻辑,提前发现不符合要求的字段类型
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user