From 73df576c1c43bfcbccf2d6778b3c6f85a39f499d Mon Sep 17 00:00:00 2001 From: xuans Date: Sun, 27 Sep 2026 19:54:57 +0800 Subject: [PATCH] =?UTF-8?q?=E9=99=90=E5=88=B6MySQL=E5=A4=87=E4=BB=BD?= =?UTF-8?q?=E5=85=83=E6=95=B0=E6=8D=AE=E9=94=81=E7=AD=89=E5=BE=85=E5=B9=B6?= =?UTF-8?q?=E8=A6=86=E7=9B=96=E5=8F=96=E6=B6=88=E4=B8=8E=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/cli/mysql_backup.go | 5 + .../mysql_backup_failure_integration_test.go | 230 ++++++++++++++++++ .../cli/mysql_snapshot_integration_test.go | 28 ++- 3 files changed, 251 insertions(+), 12 deletions(-) create mode 100644 backend/internal/cli/mysql_backup_failure_integration_test.go diff --git a/backend/internal/cli/mysql_backup.go b/backend/internal/cli/mysql_backup.go index b56b526..8bdd7cf 100644 --- a/backend/internal/cli/mysql_backup.go +++ b/backend/internal/cli/mysql_backup.go @@ -50,6 +50,11 @@ func backupMySQLDatabase(ctx context.Context, cfg config.DatabaseConfig, output return fmt.Errorf("创建 MySQL 一致性备份事务失败：%w", err) } defer tx.Rollback() + // 客户端取消会关闭连接，但服务端等待元数据锁时未必立即感知断开。 + // 仅限制本次备份会话的单次锁等待；不限制总导出时间，也不修改全局配置。 + if _, err := tx.ExecContext(ctx, "SET SESSION lock_wait_timeout = 5"); err != nil { + return fmt.Errorf("设置 MySQL 备份元数据锁等待上限失败：%w", err) + } 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) diff --git a/backend/internal/cli/mysql_backup_failure_integration_test.go b/backend/internal/cli/mysql_backup_failure_integration_test.go new file mode 100644 index 0000000..b1e88dd --- /dev/null +++ b/backend/internal/cli/mysql_backup_failure_integration_test.go @@ -0,0 +1,230 @@ +package cli + +import ( + "bytes" + "context" + "errors" + "fmt" + "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 TestMySQLRealBackupFailureBoundaries(t *testing.T) { + server := mysqltest.New(t) + var originalGlobalTimeout int + if err := server.Admin().QueryRowContext(t.Context(), "SELECT @@GLOBAL.lock_wait_timeout").Scan(&originalGlobalTimeout); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var currentGlobalTimeout int + if err := server.Admin().QueryRowContext(ctx, "SELECT @@GLOBAL.lock_wait_timeout").Scan(¤tGlobalTimeout); err != nil || currentGlobalTimeout != originalGlobalTimeout { + t.Errorf("备份不能修改数据库全局锁等待配置：%v", err) + } + }) + for _, scenario := range []struct { + name string + ddl string + }{ + {name: "reject_nontransactional_table", ddl: "CREATE TABLE %s (id INT PRIMARY KEY) ENGINE=MyISAM"}, + {name: "reject_view", ddl: "CREATE VIEW %s AS SELECT 1 AS id"}, + } { + t.Run(scenario.name, func(t *testing.T) { + cfg := server.Database(t) + parsed, err := mysql.ParseDSN(cfg.DSN) + if err != nil { + t.Fatal("解析隔离数据库连接失败") + } + qualified := quoteMySQLIdentifier(parsed.DBName) + ".`unsupported_object`" + if _, err := server.Admin().ExecContext(t.Context(), fmt.Sprintf(scenario.ddl, qualified)); err != nil { + t.Fatal(err) + } + output := filepath.Join(server.Root(), scenario.name+".sql") + original := []byte("原有备份不能因失败被替换\n") + writeMySQLBackupFixture(t, output, original) + err = backupMySQLDatabase(t.Context(), cfg, output) + if err == nil || !strings.Contains(err.Error(), "仅支持 InnoDB 基础表") { + t.Fatalf("非事务对象不能被当作一致性备份成功导出：%v", err) + } + assertMySQLBackupFile(t, output, original) + assertMySQLBackupAbsent(t, output+".tmp") + }) + } + + for _, scenario := range []struct { + name string + cancel bool + timeout bool + }{ + {name: "cancel_after_first_table", cancel: true}, + {name: "query_failure_after_first_table"}, + {name: "metadata_lock_timeout", timeout: true}, + } { + for _, existing := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/existing_output_%t", scenario.name, existing), func(t *testing.T) { + cfg := server.Database(t) + 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) ENGINE=InnoDB"); err != nil { + t.Fatal(err) + } + if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO "+table+" VALUES (1)"); 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() + if _, err := locker.ExecContext(ctx, "UNLOCK TABLES"); err != nil { + t.Error(err) + } + }() + output := filepath.Join(server.Root(), fmt.Sprintf("%s-%t.sql", scenario.name, existing)) + original := []byte("已完成的历史备份\n") + if existing { + writeMySQLBackupFixture(t, output, original) + } + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + done := make(chan struct{}) + var backupErr error + go func() { + defer close(done) + backupErr = backupMySQLDatabase(ctx, cfg, output) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("备份协程未按期结束") + } + }) + connectionID := waitForMySQLBackupLock(t, ctx, server.Admin(), parsed.DBName) + content, err := os.ReadFile(output + ".tmp") + if err != nil || !bytes.Contains(content, []byte("INSERT INTO `a_snapshot`")) { + t.Fatal("失败必须注入在首表已经写入备份、第二张表等待锁之后") + } + if scenario.cancel { + cancel() + } else if !scenario.timeout { + // 只终止本次已确认等待锁的备份语句，不终止连接或共享数据库。 + if _, err := server.Admin().ExecContext(ctx, fmt.Sprintf("KILL QUERY %d", connectionID)); err != nil { + t.Fatal(err) + } + } + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("备份失败后未及时退出") + } + if scenario.cancel { + if !errors.Is(backupErr, context.Canceled) { + t.Fatalf("备份未保留取消错误：%v", backupErr) + } + } else { + var sqlErr *mysql.MySQLError + expectedCode := uint16(1317) + if scenario.timeout { + expectedCode = 1205 + } + if !errors.As(backupErr, &sqlErr) || sqlErr.Number != expectedCode { + t.Fatalf("备份未保留真实语句中断错误：%v", backupErr) + } + } + if existing { + assertMySQLBackupFile(t, output, original) + } else { + assertMySQLBackupAbsent(t, output) + } + assertMySQLBackupAbsent(t, output+".tmp") + // 备份事务曾读取首表；失败后必须释放元数据锁，不能阻塞后续维护。 + verifyCtx, verifyCancel := context.WithTimeout(t.Context(), 10*time.Second) + defer verifyCancel() + if _, err := db.DB.ExecContext(verifyCtx, "ALTER TABLE a_snapshot ADD COLUMN released INT NULL"); err != nil { + t.Fatalf("备份失败后未释放首表元数据锁：%v", err) + } + }) + } + } + + t.Run("temporary_file_conflict", func(t *testing.T) { + cfg := server.Database(t) + output := filepath.Join(server.Root(), "conflict.sql") + original := []byte("保留原有完整备份\n") + temporary := []byte("其他备份操作持有的临时文件\n") + writeMySQLBackupFixture(t, output, original) + writeMySQLBackupFixture(t, output+".tmp", temporary) + if err := backupMySQLDatabase(t.Context(), cfg, output); !errors.Is(err, os.ErrExist) { + t.Fatalf("临时文件冲突必须拒绝：%v", err) + } + assertMySQLBackupFile(t, output, original) + assertMySQLBackupFile(t, output+".tmp", temporary) + }) + + t.Run("publish_failure", func(t *testing.T) { + cfg := server.Database(t) + output := filepath.Join(server.Root(), "occupied-output") + if err := os.Mkdir(output, 0o700); err != nil { + t.Fatal(err) + } + marker := filepath.Join(output, "marker.txt") + original := []byte("已有目录与内容必须保留\n") + writeMySQLBackupFixture(t, marker, original) + err := backupMySQLDatabase(t.Context(), cfg, output) + var linkErr *os.LinkError + if !errors.As(err, &linkErr) || linkErr.Op != "rename" { + t.Fatalf("必须在文件发布阶段失败：%v", err) + } + assertMySQLBackupFile(t, marker, original) + assertMySQLBackupAbsent(t, output+".tmp") + }) +} + +func writeMySQLBackupFixture(t *testing.T, path string, content []byte) { + t.Helper() + if err := os.WriteFile(path, content, 0o600); err != nil { + t.Fatal(err) + } +} + +func assertMySQLBackupFile(t *testing.T, path string, expected []byte) { + t.Helper() + content, err := os.ReadFile(path) + if err != nil || !bytes.Equal(content, expected) { + t.Fatalf("已有备份或其他操作的文件被修改：%s，%v", path, err) + } +} + +func assertMySQLBackupAbsent(t *testing.T, path string) { + t.Helper() + if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("不应遗留不完整备份或本次临时文件：%s，%v", path, err) + } +} diff --git a/backend/internal/cli/mysql_snapshot_integration_test.go b/backend/internal/cli/mysql_snapshot_integration_test.go index 2c38b3d..8d4a2ac 100644 --- a/backend/internal/cli/mysql_snapshot_integration_test.go +++ b/backend/internal/cli/mysql_snapshot_integration_test.go @@ -3,6 +3,7 @@ package cli import ( "context" "database/sql" + "errors" "os" "path/filepath" "strings" @@ -61,8 +62,12 @@ func TestMySQLRealBackupSnapshot(t *testing.T) { 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) }() + done := make(chan struct{}) + var backupErr error + go func() { + defer close(done) + backupErr = backupMySQLDatabase(ctx, cfg, output) + }() t.Cleanup(func() { cancel() select { @@ -88,10 +93,9 @@ func TestMySQLRealBackupSnapshot(t *testing.T) { t.Fatal(err) } select { - case err := <-done: - done <- err - if err != nil { - t.Fatal(err) + case <-done: + if backupErr != nil { + t.Fatal(backupErr) } case <-ctx.Done(): t.Fatal("真实备份未按期完成") @@ -116,18 +120,18 @@ func TestMySQLRealBackupSnapshot(t *testing.T) { } } -func waitForMySQLBackupLock(t *testing.T, ctx context.Context, admin *sql.DB, name string) { +func waitForMySQLBackupLock(t *testing.T, ctx context.Context, admin *sql.DB, name string) uint64 { 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 { + var connectionID uint64 + err := admin.QueryRowContext(ctx, "SELECT ID 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(&connectionID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { t.Fatal(err) } - if count > 0 { - return + if err == nil { + return connectionID } select { case <-ctx.Done():