修复MySQL恢复会话切换与精确数值备份错误
固定恢复连接，按DECIMAL列类型保留精确文本。新增双版本实库备份恢复、错误停止、取消以及正式命令失败回滚回归；不将同进程命令测试视为独立发行验收。
This commit is contained in:
parent
ac205e2a06
commit
7bfd400345
@ -140,6 +140,10 @@ func dumpMySQLTable(ctx context.Context, db *sql.DB, output *os.File, tableName
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取 MySQL 表字段失败（%s）：%w", tableName, err)
|
||||
}
|
||||
columnTypes, err := rows.ColumnTypes()
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取 MySQL 表字段类型失败（%s）：%w", tableName, err)
|
||||
}
|
||||
for rows.Next() {
|
||||
values := make([]any, len(columns))
|
||||
pointers := make([]any, len(columns))
|
||||
@ -151,7 +155,7 @@ func dumpMySQLTable(ctx context.Context, db *sql.DB, output *os.File, tableName
|
||||
}
|
||||
literals := make([]string, len(values))
|
||||
for index, value := range values {
|
||||
literals[index] = mysqlSQLLiteral(value)
|
||||
literals[index] = mysqlColumnSQLLiteral(value, columnTypes[index].DatabaseTypeName())
|
||||
}
|
||||
if _, err := fmt.Fprintf(output, "INSERT INTO %s (%s) VALUES (%s);\n", quotedTable, quoteMySQLColumns(columns), strings.Join(literals, ", ")); err != nil {
|
||||
return fmt.Errorf("写入 MySQL 表数据失败（%s）：%w", tableName, err)
|
||||
@ -176,6 +180,20 @@ func quoteMySQLColumns(columns []string) string {
|
||||
return strings.Join(quoted, ", ")
|
||||
}
|
||||
|
||||
func mysqlColumnSQLLiteral(value any, columnType string) string {
|
||||
// 驱动将 DECIMAL 返回为文本字节；十六进制字面量在数值列中会被当作整数。
|
||||
// 使用文本字面量保留任意精度，不经浮点转换；真正的二进制列仍按原样导出。
|
||||
if columnType == "DECIMAL" {
|
||||
switch typed := value.(type) {
|
||||
case sql.RawBytes:
|
||||
return mysqlStringLiteral(string(typed))
|
||||
case []byte:
|
||||
return mysqlStringLiteral(string(typed))
|
||||
}
|
||||
}
|
||||
return mysqlSQLLiteral(value)
|
||||
}
|
||||
|
||||
func mysqlSQLLiteral(value any) string {
|
||||
switch typed := value.(type) {
|
||||
case nil:
|
||||
@ -243,8 +261,14 @@ func restoreMySQL(ctx context.Context, cfg config.DatabaseConfig, input string)
|
||||
return fmt.Errorf("打开 MySQL 恢复连接失败：%w", err)
|
||||
}
|
||||
defer db.Close()
|
||||
// 会话变量与外键开关必须留在同一条连接上；失败后返回，不切换连接重放。
|
||||
connection, err := db.DB.Conn(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取 MySQL 恢复会话失败：%w", err)
|
||||
}
|
||||
defer connection.Close()
|
||||
for _, statement := range splitMySQLStatements(string(script)) {
|
||||
if _, err := db.DB.ExecContext(ctx, statement); err != nil {
|
||||
if _, err := connection.ExecContext(ctx, statement); err != nil {
|
||||
return fmt.Errorf("执行 MySQL 恢复语句失败：%w", err)
|
||||
}
|
||||
}
|
||||
|
||||
@ -40,6 +40,28 @@ func TestQuoteMySQLColumns(t *testing.T) {
|
||||
t.Fatalf("MySQL 字段列表不正确：%s", value)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMySQLColumnSQLLiteral(t *testing.T) {
|
||||
for _, item := range []struct {
|
||||
name string
|
||||
value any
|
||||
columnType string
|
||||
want string
|
||||
}{
|
||||
{name: "decimal", value: []byte("123456789.123456"), columnType: "DECIMAL", want: "'123456789.123456'"},
|
||||
{name: "precision", value: sql.RawBytes("-12345678901234567890123456789012345.123456789012345678901234567890"), columnType: "DECIMAL", want: "'-12345678901234567890123456789012345.123456789012345678901234567890'"},
|
||||
{name: "null", value: nil, columnType: "DECIMAL", want: "NULL"},
|
||||
{name: "string", value: "0.000001", columnType: "DECIMAL", want: "'0.000001'"},
|
||||
{name: "binary", value: []byte("12"), columnType: "BLOB", want: "X'3132'"},
|
||||
} {
|
||||
t.Run(item.name, func(t *testing.T) {
|
||||
if got := mysqlColumnSQLLiteral(item.value, item.columnType); got != item.want {
|
||||
t.Fatalf("字段字面量不正确：%s", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSplitMySQLStatementsKeepsQuotedSemicolons(t *testing.T) {
|
||||
script := "-- 注释\nSET FOREIGN_KEY_CHECKS=0;\nINSERT INTO `items` (`value`) VALUES ('a; b'), ('it''s');\n"
|
||||
statements := splitMySQLStatements(script)
|
||||
|
||||
198
backend/internal/cli/mysql_commands_integration_test.go
Normal file
198
backend/internal/cli/mysql_commands_integration_test.go
Normal file
@ -0,0 +1,198 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/go-sql-driver/mysql"
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/secretstore"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/testsupport/mysqltest"
|
||||
)
|
||||
|
||||
// 在同一进程调用正式命令；测试 TLS 注册名不能用于独立发行二进制验收。
|
||||
func TestMySQLRealBackupRestoreCommands(t *testing.T) {
|
||||
server := mysqltest.New(t)
|
||||
for _, scenario := range []string{"success", "sql_failure", "certificate_failure"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
connection := server.Database(t)
|
||||
connection.MaxIdleConns = 0
|
||||
db, err := database.Open(t.Context(), connection)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
if err := database.Migrate(t.Context(), db.DB, "mysql"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
exec := func(statement string) {
|
||||
t.Helper()
|
||||
if _, err := db.DB.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
exec("INSERT INTO settings (`key`, value) VALUES ('integration', 'backup')")
|
||||
exec("INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('backup', 'test', '1', 'ok', 'backup')")
|
||||
root := filepath.Join(server.Root(), scenario)
|
||||
cfg := config.Default(root)
|
||||
cfg.Paths = config.PathsConfig{
|
||||
Install: root, Config: filepath.Join(root, "conf"), Data: filepath.Join(root, "data"),
|
||||
Certificate: filepath.Join(root, "ssl"), Log: filepath.Join(root, "log"), Backup: filepath.Join(root, "snapshots"),
|
||||
}
|
||||
parsed, err := mysql.ParseDSN(connection.DSN)
|
||||
if err != nil {
|
||||
t.Fatal("解析隔离数据库连接失败")
|
||||
}
|
||||
username, password := parsed.User, parsed.Passwd
|
||||
parsed.User, parsed.Passwd = "", ""
|
||||
cfg.Database = connection
|
||||
cfg.Database.DSN = parsed.FormatDSN()
|
||||
cfg.Database.CredentialRef = "secret://database/maintenance"
|
||||
current := filepath.Join(cfg.Paths.Config, "scert.json")
|
||||
if err := cfg.Save(current); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, err := secretstore.Open(cfg.Paths.Config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Put(t.Context(), cfg.Database.CredentialRef, map[string]string{"username": username, "password": password}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
write := func(path string, content []byte) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, content, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
read := func(path string) []byte {
|
||||
t.Helper()
|
||||
content, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return content
|
||||
}
|
||||
certificate := filepath.Join(cfg.Paths.Certificate, "integration.pem")
|
||||
write(certificate, read(filepath.Join(server.Root(), "server.crt")))
|
||||
previous := configPath
|
||||
configPath = current
|
||||
t.Cleanup(func() { configPath = previous })
|
||||
run := func(command *cobra.Command, args ...string) (string, error) {
|
||||
t.Helper()
|
||||
var output bytes.Buffer
|
||||
command.SetArgs(args)
|
||||
command.SetContext(t.Context())
|
||||
command.SetOut(&output)
|
||||
command.SetErr(&output)
|
||||
err := command.Execute()
|
||||
if strings.Contains(output.String(), password) || (err != nil && strings.Contains(err.Error(), password)) {
|
||||
t.Fatal("命令输出不得泄漏数据库密码")
|
||||
}
|
||||
return output.String(), err
|
||||
}
|
||||
input := filepath.Join(root, "backup")
|
||||
if _, err := run(newBackupCommand(), "--output", input); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := validateBackupManifest(input); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
exec("UPDATE settings SET value = 'current' WHERE `key` = 'integration'")
|
||||
exec("INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('current', 'test', '2', 'ok', 'current')")
|
||||
cfg.Server.Port++
|
||||
if err := cfg.Save(current); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Put(t.Context(), "secret://integration/marker", map[string]string{"value": "current"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
write(certificate, []byte("current-certificate"))
|
||||
before := make(map[string][]byte)
|
||||
for _, name := range []string{"scert.json", "master.key", "secrets.json"} {
|
||||
before[name] = read(filepath.Join(cfg.Paths.Config, name))
|
||||
}
|
||||
if scenario == "sql_failure" {
|
||||
// 保持清单有效，以确认错误发生在实际导入阶段而非哈希校验阶段。
|
||||
path := filepath.Join(input, "scert.sql")
|
||||
write(path, append(read(path), []byte("\nINSERT INTO missing_restore_table VALUES (1);\n")...))
|
||||
manifest, err := buildBackupManifestExcludingManifest(input)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
encoded, err := json.Marshal(manifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
write(filepath.Join(input, "manifest.json"), encoded)
|
||||
if err := validateBackupManifest(input); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if scenario == "certificate_failure" {
|
||||
if err := os.Rename(cfg.Paths.Certificate, cfg.Paths.Certificate+"-preserved"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
write(cfg.Paths.Certificate, []byte("blocked-target"))
|
||||
}
|
||||
output, err := run(newRestoreCommand(), "--input", input)
|
||||
want, count := "backup", 1
|
||||
if scenario == "success" {
|
||||
if err != nil || !strings.Contains(output, "恢复已完成") {
|
||||
t.Fatalf("正式命令恢复失败：%v", err)
|
||||
}
|
||||
for name := range before {
|
||||
if !bytes.Equal(read(filepath.Join(cfg.Paths.Config, name)), read(filepath.Join(input, name))) {
|
||||
t.Fatalf("未恢复备份配置：%s", name)
|
||||
}
|
||||
}
|
||||
if !bytes.Equal(read(certificate), read(filepath.Join(input, "certificates", "integration.pem"))) {
|
||||
t.Fatal("未恢复证书文件")
|
||||
}
|
||||
} else {
|
||||
want, count = "current", 2
|
||||
phase := "恢复 MySQL 数据库失败"
|
||||
if scenario == "certificate_failure" {
|
||||
phase = "恢复敏感配置或证书失败"
|
||||
}
|
||||
if err == nil || !strings.Contains(err.Error(), phase) || strings.Contains(output, "恢复已完成") {
|
||||
t.Fatalf("未在指定阶段失败：%v", err)
|
||||
}
|
||||
for name, content := range before {
|
||||
if !bytes.Equal(read(filepath.Join(cfg.Paths.Config, name)), content) {
|
||||
t.Fatalf("失败后未恢复当前配置：%s", name)
|
||||
}
|
||||
}
|
||||
preserved := certificate
|
||||
if scenario == "certificate_failure" {
|
||||
preserved = filepath.Join(cfg.Paths.Certificate+"-preserved", "integration.pem")
|
||||
}
|
||||
if string(read(preserved)) != "current-certificate" {
|
||||
t.Fatal("失败后原证书内容被修改")
|
||||
}
|
||||
}
|
||||
var value string
|
||||
var rows int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT value FROM settings WHERE `key` = 'integration'").Scan(&value); err != nil || value != want {
|
||||
t.Fatalf("恢复或回滚后的业务数据错误：%v", err)
|
||||
}
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs").Scan(&rows); err != nil || rows != count {
|
||||
t.Fatalf("恢复或回滚后的审计记录错误：%v", err)
|
||||
}
|
||||
snapshots, err := filepath.Glob(filepath.Join(cfg.Paths.Backup, "scert-before-restore-*", "scert.sql"))
|
||||
if err != nil || len(snapshots) != 1 {
|
||||
t.Fatal("必须保留恢复前的数据库快照")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
128
backend/internal/cli/mysql_integration_test.go
Normal file
128
backend/internal/cli/mysql_integration_test.go
Normal file
@ -0,0 +1,128 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/testsupport/mysqltest"
|
||||
)
|
||||
|
||||
func TestMySQLRealBackupRestore(t *testing.T) {
|
||||
server := mysqltest.New(t)
|
||||
t.Run("session_identity", func(t *testing.T) {
|
||||
cfg := server.Database(t)
|
||||
cfg.MaxIdleConns = 0
|
||||
input := filepath.Join(server.Root(), "session.sql")
|
||||
script := "CREATE TABLE session_probe (id BIGINT NOT NULL, original_id BIGINT NOT NULL); SET @restore_connection = CONNECTION_ID(); INSERT INTO session_probe VALUES (CONNECTION_ID(), @restore_connection);"
|
||||
if err := os.WriteFile(input, []byte(script), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := restoreMySQL(t.Context(), cfg, input); err != nil {
|
||||
t.Fatalf("恢复必须保留会话变量：%v", err)
|
||||
}
|
||||
db, err := database.Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
var same bool
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT id = original_id FROM session_probe").Scan(&same); err != nil || !same {
|
||||
t.Fatalf("恢复过程发生连接切换：%v", err)
|
||||
}
|
||||
})
|
||||
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)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
exec := func(statement string, args ...any) sql.Result {
|
||||
t.Helper()
|
||||
result, err := db.DB.ExecContext(t.Context(), statement, args...)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return result
|
||||
}
|
||||
// 子表排在父表前面，恢复需要在同一会话内暂时关闭外键检查。
|
||||
exec("CREATE TABLE z_parent (id BIGINT PRIMARY KEY AUTO_INCREMENT, label VARCHAR(100) NOT NULL) ENGINE=InnoDB")
|
||||
exec("CREATE TABLE a_child (id BIGINT PRIMARY KEY, parent_id BIGINT NOT NULL, payload BLOB, amount DECIMAL(20,6), optional_value TEXT NULL, created_at DATETIME(6), FOREIGN KEY (parent_id) REFERENCES z_parent(id)) ENGINE=InnoDB")
|
||||
label := "证书恢复；单引号'与反斜线\\\n"
|
||||
payload := []byte{0, 1, 39, 59, 92, 255}
|
||||
exec("INSERT INTO z_parent (id, label) VALUES (7, ?)", label)
|
||||
exec("INSERT INTO a_child VALUES (1, 7, ?, 123456789.123456, NULL, '2026-09-27 12:34:56.123456')", payload)
|
||||
root := filepath.Join(server.Root(), scenario.name)
|
||||
if err := os.MkdirAll(root, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
backup := filepath.Join(root, "backup.sql")
|
||||
if err := backupMySQLDatabase(t.Context(), cfg, backup); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
exec("DELETE FROM a_child")
|
||||
exec("DELETE FROM z_parent")
|
||||
exec("INSERT INTO z_parent (id, label) VALUES (8, 'changed')")
|
||||
exec("INSERT INTO a_child (id, parent_id) VALUES (2, 8)")
|
||||
if err := restoreMySQL(t.Context(), cfg, backup); err != nil {
|
||||
t.Fatalf("恢复真实备份失败：%v", err)
|
||||
}
|
||||
var gotLabel, amount, created string
|
||||
var gotPayload []byte
|
||||
var optional sql.NullString
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT p.label, c.payload, CAST(c.amount AS CHAR), c.optional_value, DATE_FORMAT(c.created_at, '%Y-%m-%d %H:%i:%s.%f') FROM a_child c JOIN z_parent p ON p.id = c.parent_id WHERE c.id = 1 AND p.id = 7").Scan(&gotLabel, &gotPayload, &amount, &optional, &created); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if gotLabel != label || !bytes.Equal(gotPayload, payload) || amount != "123456789.123456" || optional.Valid || created != "2026-09-27 12:34:56.123456" {
|
||||
t.Fatal("恢复后的文本、二进制、精确数值、空值或时间不一致")
|
||||
}
|
||||
id, err := exec("INSERT INTO z_parent (label) VALUES ('next')").LastInsertId()
|
||||
if err != nil || id != 8 {
|
||||
t.Fatalf("恢复后的自增序列不正确：%d，%v", id, err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO a_child (id, parent_id) VALUES (99, 999)"); err == nil {
|
||||
t.Fatal("恢复后仍须强制检查外键约束")
|
||||
}
|
||||
|
||||
t.Run("stop_on_error", func(t *testing.T) {
|
||||
input := filepath.Join(root, "invalid.sql")
|
||||
// 第一条写入已经成功时，不能隐式重放或继续执行失败后的语句。
|
||||
script := "SET FOREIGN_KEY_CHECKS=0; INSERT INTO z_parent (id, label) VALUES (20, 'before'); INSERT INTO absent_table VALUES (1); INSERT INTO z_parent (id, label) VALUES (21, 'after');"
|
||||
if err := os.WriteFile(input, []byte(script), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := restoreMySQL(t.Context(), cfg, input); err == nil {
|
||||
t.Fatal("无效恢复语句应返回错误")
|
||||
}
|
||||
var count int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM z_parent WHERE id IN (20, 21)").Scan(&count); err != nil || count != 1 {
|
||||
t.Fatalf("失败后不应继续执行或隐式重放：%d，%v", count, err)
|
||||
}
|
||||
})
|
||||
t.Run("cancel_before_restore", func(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
cancel()
|
||||
if err := restoreMySQL(ctx, cfg, backup); err == nil {
|
||||
t.Fatal("已取消的恢复不得执行")
|
||||
}
|
||||
var count int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM z_parent WHERE id = 20").Scan(&count); err != nil || count != 1 {
|
||||
t.Fatalf("取消恢复不应修改数据：%d，%v", count, err)
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user