diff --git a/backend/internal/cli/mysql_backup.go b/backend/internal/cli/mysql_backup.go index 78d3374..b56b526 100644 --- a/backend/internal/cli/mysql_backup.go +++ b/backend/internal/cli/mysql_backup.go @@ -43,22 +43,34 @@ func backupMySQLDatabase(ctx context.Context, cfg config.DatabaseConfig, output _ = os.Remove(temporaryPath) } }() + // 所有表的数据必须来自同一个可重复读事务，不能按表从连接池独立读取。 + // 元数据变更不属于 MVCC 快照，运维期间仍需禁止外部并发 DDL。 + tx, err := db.DB.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true}) + if err != nil { + return fmt.Errorf("创建 MySQL 一致性备份事务失败：%w", err) + } + defer tx.Rollback() if _, err := fmt.Fprintf(file, "-- Scert MySQL 数据备份\n-- 生成时间：%s\n\nSET FOREIGN_KEY_CHECKS=0;\n\n", time.Now().UTC().Format(time.RFC3339)); err != nil { return fmt.Errorf("写入 MySQL 备份头失败：%w", err) } - rows, err := db.DB.QueryContext(ctx, "SHOW TABLES") + rows, err := tx.QueryContext(ctx, "SELECT TABLE_NAME, ENGINE, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() ORDER BY TABLE_NAME") if err != nil { return fmt.Errorf("读取 MySQL 表列表失败：%w", err) } var tableNames []string for rows.Next() { - var tableName string - if err := rows.Scan(&tableName); err != nil { + var tableName, tableType string + var engine sql.NullString + if err := rows.Scan(&tableName, &engine, &tableType); err != nil { _ = rows.Close() return fmt.Errorf("读取 MySQL 表名失败：%w", err) } + if tableType != "BASE TABLE" || !engine.Valid || !strings.EqualFold(engine.String, "InnoDB") { + _ = rows.Close() + return fmt.Errorf("MySQL 一致性备份当前仅支持 InnoDB 基础表，不能静默导出其他对象：%s", tableName) + } tableNames = append(tableNames, tableName) } if err := rows.Err(); err != nil { @@ -70,19 +82,28 @@ func backupMySQLDatabase(ctx context.Context, cfg config.DatabaseConfig, output } for _, tableName := range tableNames { - if err := dumpMySQLTable(ctx, db.DB, file, tableName); err != nil { + if err := dumpMySQLTable(ctx, tx, file, tableName); err != nil { return err } } if _, err := fmt.Fprint(file, "SET FOREIGN_KEY_CHECKS=1;\n"); err != nil { return fmt.Errorf("写入 MySQL 备份尾失败：%w", err) } + if err := tx.Commit(); err != nil { + return fmt.Errorf("结束 MySQL 一致性备份事务失败：%w", err) + } + if err := file.Sync(); err != nil { + return fmt.Errorf("同步 MySQL 备份文件失败：%w", err) + } if err := file.Close(); err != nil { return fmt.Errorf("关闭 MySQL 备份文件失败：%w", err) } if err := os.Chmod(temporaryPath, 0o600); err != nil { return fmt.Errorf("设置 MySQL 备份文件权限失败：%w", err) } + if err := ctx.Err(); err != nil { + return err + } if err := os.Rename(temporaryPath, output); err != nil { return fmt.Errorf("保存 MySQL 备份文件失败：%w", err) } @@ -121,7 +142,7 @@ func createMySQLRestoreBackup(ctx context.Context, cfg config.Config, connection return rollbackPath, nil } -func dumpMySQLTable(ctx context.Context, db *sql.DB, output *os.File, tableName string) error { +func dumpMySQLTable(ctx context.Context, db *sql.Tx, output *os.File, tableName string) error { quotedTable := quoteMySQLIdentifier(tableName) var name, createSQL string if err := db.QueryRowContext(ctx, "SHOW CREATE TABLE "+quotedTable).Scan(&name, &createSQL); err != nil { diff --git a/backend/internal/cli/mysql_snapshot_integration_test.go b/backend/internal/cli/mysql_snapshot_integration_test.go new file mode 100644 index 0000000..2c38b3d --- /dev/null +++ b/backend/internal/cli/mysql_snapshot_integration_test.go @@ -0,0 +1,138 @@ +package cli + +import ( + "context" + "database/sql" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/go-sql-driver/mysql" + + "git.pcux.cn/xuans/Scert/backend/internal/database" + "git.pcux.cn/xuans/Scert/backend/internal/testsupport/mysqltest" +) + +func TestMySQLRealBackupSnapshot(t *testing.T) { + server := mysqltest.New(t) + for _, scenario := range []struct { + name string + idle int + }{ + {name: "without_idle_connections", idle: 0}, + {name: "with_idle_connections", idle: 1}, + } { + t.Run(scenario.name, func(t *testing.T) { + cfg := server.Database(t) + cfg.MaxIdleConns = scenario.idle + db, err := database.Open(t.Context(), cfg) + if err != nil { + t.Fatal(err) + } + defer db.Close() + for _, table := range []string{"a_snapshot", "b_snapshot"} { + if _, err := db.DB.ExecContext(t.Context(), "CREATE TABLE "+table+" (id INT PRIMARY KEY, epoch INT NOT NULL) ENGINE=InnoDB"); err != nil { + t.Fatal(err) + } + if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO "+table+" VALUES (1, 0)"); err != nil { + t.Fatal(err) + } + } + parsed, err := mysql.ParseDSN(cfg.DSN) + if err != nil { + t.Fatal("解析隔离数据库连接失败") + } + locker, err := server.Admin().Conn(t.Context()) + if err != nil { + t.Fatal(err) + } + defer locker.Close() + qualified := quoteMySQLIdentifier(parsed.DBName) + ".`b_snapshot`" + if _, err := locker.ExecContext(t.Context(), "LOCK TABLES "+qualified+" WRITE"); err != nil { + t.Fatal(err) + } + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _ = locker.ExecContext(ctx, "UNLOCK TABLES") + }() + output := filepath.Join(server.Root(), scenario.name+".sql") + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- backupMySQLDatabase(ctx, cfg, output) }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("备份协程未按期结束") + } + }) + waitForMySQLBackupLock(t, ctx, server.Admin(), parsed.DBName) + content, err := os.ReadFile(output + ".tmp") + if err != nil || !strings.Contains(string(content), "INSERT INTO `a_snapshot`") { + t.Fatal("未到达首表已导出、第二张表等待锁的测试边界") + } + // 真实状态依次为 (0,0)、(1,0)、(1,1)，从未出现 (0,1)。 + // 更新第一张表后，再在持锁连接更新第二张表，避免依赖调度概率。 + if _, err := db.DB.ExecContext(ctx, "UPDATE a_snapshot SET epoch = 1"); err != nil { + t.Fatal(err) + } + if _, err := locker.ExecContext(ctx, "UPDATE "+qualified+" SET epoch = 1"); err != nil { + t.Fatal(err) + } + if _, err := locker.ExecContext(ctx, "UNLOCK TABLES"); err != nil { + t.Fatal(err) + } + select { + case err := <-done: + done <- err + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("真实备份未按期完成") + } + restored := server.Database(t) + if err := restoreMySQL(t.Context(), restored, output); err != nil { + t.Fatal(err) + } + verify, err := database.Open(t.Context(), restored) + if err != nil { + t.Fatal(err) + } + defer verify.Close() + var first, second int + if err := verify.DB.QueryRowContext(t.Context(), "SELECT a.epoch, b.epoch FROM a_snapshot a JOIN b_snapshot b ON a.id = b.id").Scan(&first, &second); err != nil { + t.Fatal(err) + } + if first != 0 || second != 0 { + t.Fatalf("跨表备份不是同一读取快照：首表=%d，次表=%d", first, second) + } + }) + } +} + +func waitForMySQLBackupLock(t *testing.T, ctx context.Context, admin *sql.DB, name string) { + t.Helper() + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + var count int + err := admin.QueryRowContext(ctx, "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE DB = ? AND STATE LIKE '%lock%' AND (INFO = 'SHOW CREATE TABLE `b_snapshot`' OR INFO = 'SELECT * FROM `b_snapshot`')", name).Scan(&count) + if err != nil { + t.Fatal(err) + } + if count > 0 { + return + } + select { + case <-ctx.Done(): + t.Fatal("未观察到备份在第二张表等待真实数据库锁") + case <-ticker.C: + } + } +}