限制MySQL备份元数据锁等待并覆盖取消与文件失败边界
This commit is contained in:
parent
79d70426fd
commit
73df576c1c
@ -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)
|
||||
|
||||
230
backend/internal/cli/mysql_backup_failure_integration_test.go
Normal file
230
backend/internal/cli/mysql_backup_failure_integration_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
@ -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():
|
||||
|
||||
Loading…
Reference in New Issue
Block a user