修复MySQL跨表备份一致性并补充实库回归测试
This commit is contained in:
parent
95bae2240b
commit
8d742fe4e6
@ -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 {
|
||||
|
||||
138
backend/internal/cli/mysql_snapshot_integration_test.go
Normal file
138
backend/internal/cli/mysql_snapshot_integration_test.go
Normal file
@ -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:
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user