修复SQLite异常退出后恢复被旧日志覆盖并验证业务数据恢复
This commit is contained in:
parent
5e5abf2dbf
commit
6db5d7b2a4
@ -460,6 +460,9 @@ func newRestoreCommandWithPostgresTools(factory postgresToolsFactory) *cobra.Com
|
||||
} else if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
|
||||
return fmt.Errorf("检查当前数据库失败：%w", statErr)
|
||||
}
|
||||
if err := prepareSQLiteRestoreTarget(cmd.Context(), databaseTarget); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// 当前账号用于恢复前快照、导入和回滚，不能因替换配置或主密钥而丢失。
|
||||
|
||||
204
backend/internal/cli/sqlite_business_fixture_test.go
Normal file
204
backend/internal/cli/sqlite_business_fixture_test.go
Normal file
@ -0,0 +1,204 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"crypto/ecdsa"
|
||||
"crypto/elliptic"
|
||||
"crypto/rand"
|
||||
"crypto/x509"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"encoding/pem"
|
||||
"math/big"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/secretstore"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
type sqliteBusinessFixture struct {
|
||||
config config.Config
|
||||
path string
|
||||
password string
|
||||
secrets map[string]map[string]string
|
||||
certificate certificate.SavedResource
|
||||
}
|
||||
|
||||
func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture {
|
||||
t.Helper()
|
||||
fixture := sqliteBusinessFixture{config: config.Default(t.TempDir()), password: rand.Text()}
|
||||
cfg := &fixture.config
|
||||
for _, directory := range []string{cfg.Paths.Config, cfg.Paths.Data, cfg.Paths.Certificate, cfg.Paths.Backup} {
|
||||
if err := os.MkdirAll(directory, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
cfg.DNS.OwnerUserID = "owner"
|
||||
cfg.DNS.Provider = "alidns"
|
||||
cfg.DNS.CredentialRef = "secret://dns/business"
|
||||
fixture.path = filepath.Join(cfg.Paths.Config, "scert.json")
|
||||
if err := cfg.Save(fixture.path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sshKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sshBlock, err := ssh.MarshalPrivateKey(sshKey, "隔离恢复测试")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fixture.secrets = map[string]map[string]string{
|
||||
cfg.DNS.CredentialRef: {"access_key_id": rand.Text(), "access_key_secret": rand.Text()},
|
||||
"secret://ssh/business": {"private_key": string(pem.EncodeToMemory(sshBlock))},
|
||||
}
|
||||
store, err := secretstore.Open(cfg.Paths.Config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for reference, values := range fixture.secrets {
|
||||
if err := store.Put(t.Context(), reference, values); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
fixture.certificate = saveSQLiteBusinessCertificate(t, *cfg)
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte(fixture.password), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if err := database.Migrate(t.Context(), db.DB, "sqlite"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
statements := []struct {
|
||||
query string
|
||||
args []any
|
||||
}{
|
||||
{`INSERT INTO settings (key, value) VALUES ('business-recovery', ?)`, []any{`{"label":"中文恢复配置","enabled":true}`}},
|
||||
{`INSERT INTO users (id, username, email, password_hash, role, groups_json) VALUES ('owner', 'recovery-owner', 'owner@example.test', ?, 'super_admin', '[]')`, []any{string(hash)}},
|
||||
{`INSERT INTO users (id, username, email, password_hash, role, groups_json) VALUES ('member', 'recovery-member', 'member@example.test', ?, 'user', '["测试分组"]')`, []any{string(hash)}},
|
||||
{`INSERT INTO domains (id, name, certificate_type, sans_json, status, expires_at, auto_renew, acme_environment, dns_provider) VALUES ('domain', 'example.test', 'single', '["example.test"]', 'issued', ?, 1, 'staging', 'alidns')`, []any{fixture.certificate.NotAfter}},
|
||||
{`INSERT INTO servers (id, name, address, port, username, private_key_ref, host_key_fingerprint, certificate_path, private_key_path, certificate_chain_path, web_server, reload_command_json, validation_command_json) VALUES ('server', '恢复测试服务器', '192.0.2.10', 22, 'deploy', 'secret://ssh/business', 'test-only-fingerprint', '/test/cert.pem', '/test/key.pem', '/test/chain.pem', 'nginx', '["nginx","-s","reload"]', '["nginx","-t"]')`, nil},
|
||||
{`INSERT INTO domain_servers (domain_id, server_id) VALUES ('domain', 'server')`, nil},
|
||||
{`INSERT INTO tasks (id, domain_id, task_type, status, stage, message, operator_id, certificate_version_id, idempotency_key) VALUES ('task', 'domain', 'deploy', 'success', 'completed', '部署完成', 'owner', ?, 'recovery-idempotency')`, []any{fixture.certificate.Version}},
|
||||
{`INSERT INTO task_attempts (id, task_id, attempt_number, stage, status, message) VALUES ('attempt', 'task', 1, 'deploy', 'success', '证书已推送')`, nil},
|
||||
{`INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('deploy', 'domain', 'domain', 'success', '恢复测试审计')`, nil},
|
||||
}
|
||||
for i, statement := range statements {
|
||||
if _, err := db.DB.ExecContext(t.Context(), statement.query, statement.args...); err != nil {
|
||||
t.Fatalf("写入业务夹具第 %d 条失败：%v", i+1, err)
|
||||
}
|
||||
}
|
||||
return fixture
|
||||
}
|
||||
|
||||
func saveSQLiteBusinessCertificate(t *testing.T, cfg config.Config) certificate.SavedResource {
|
||||
t.Helper()
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
template := &x509.Certificate{SerialNumber: big.NewInt(17), DNSNames: []string{"example.test"}, NotBefore: now.Add(-time.Hour), NotAfter: now.Add(24 * time.Hour), KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}}
|
||||
der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keyDER, err := x509.MarshalPKCS8PrivateKey(key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resource := certificate.Resource{Domains: []string{"example.test"}, Certificate: pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), PrivateKey: pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER})}
|
||||
store := certificate.FileStore{Root: cfg.Paths.Certificate, BackupRoot: cfg.Paths.Backup}
|
||||
saved, err := store.Save(t.Context(), "example.test", resource)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return saved
|
||||
}
|
||||
|
||||
// 比较全部现有表和模式定义，不以行数相同代替关联、字段及迁移数据一致。
|
||||
func snapshotSQLiteBusiness(t *testing.T, cfg config.Config) map[string]string {
|
||||
t.Helper()
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if err := database.Migrate(t.Context(), db.DB, "sqlite"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tables, err := db.DB.QueryContext(t.Context(), "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var names []string
|
||||
for tables.Next() {
|
||||
var name string
|
||||
if err := tables.Scan(&name); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
names = append(names, name)
|
||||
}
|
||||
if err := tables.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tables.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result := make(map[string]string)
|
||||
result["schema"] = snapshotSQLiteRows(t, db.DB, "SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name")
|
||||
for _, name := range names {
|
||||
result[name] = snapshotSQLiteRows(t, db.DB, `SELECT * FROM "`+strings.ReplaceAll(name, `"`, `""`)+`" ORDER BY 1`)
|
||||
}
|
||||
for _, name := range []string{"settings", "users", "domains", "servers", "domain_servers", "tasks", "task_attempts", "audit_logs", "schema_migrations", "sqlite_sequence"} {
|
||||
if result[name] == "" || result[name] == "null" {
|
||||
t.Fatalf("业务夹具缺少表数据：%s", name)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func snapshotSQLiteRows(t *testing.T, db *sql.DB, query string) string {
|
||||
t.Helper()
|
||||
rows, err := db.QueryContext(t.Context(), query)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
columns, err := rows.Columns()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var values [][]any
|
||||
for rows.Next() {
|
||||
value := make([]any, len(columns))
|
||||
dest := make([]any, len(columns))
|
||||
for i := range dest {
|
||||
dest[i] = &value[i]
|
||||
}
|
||||
if err := rows.Scan(dest...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
values = append(values, value)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := json.Marshal(values)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
183
backend/internal/cli/sqlite_business_restore_test.go
Normal file
183
backend/internal/cli/sqlite_business_restore_test.go
Normal file
@ -0,0 +1,183 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/repository"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/secretstore"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
func TestSQLiteBusinessBackupRestore(t *testing.T) {
|
||||
fixture := newSQLiteBusinessFixture(t)
|
||||
cfg := fixture.config
|
||||
previous := configPath
|
||||
configPath = fixture.path
|
||||
t.Cleanup(func() { configPath = previous })
|
||||
expected := snapshotSQLiteBusiness(t, cfg)
|
||||
input := filepath.Join(t.TempDir(), "business-backup")
|
||||
backup := newBackupCommand()
|
||||
var output strings.Builder
|
||||
backup.SetOut(&output)
|
||||
backup.SetErr(&output)
|
||||
backup.SetArgs([]string{"--output", input})
|
||||
if err := backup.ExecuteContext(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := validateBackupManifest(input); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// 备份后改变数据库、配置、加密凭据和证书，确保恢复不是空操作。
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
for _, query := range []string{
|
||||
"UPDATE settings SET value = '变更后'",
|
||||
"UPDATE users SET status = 'disabled', groups_json = '[]'",
|
||||
"UPDATE domains SET status = 'failed'",
|
||||
"UPDATE servers SET name = '变更后服务器'",
|
||||
"DELETE FROM domain_servers",
|
||||
"UPDATE tasks SET message = '变更后任务'",
|
||||
"DELETE FROM task_attempts",
|
||||
"INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('after-backup', 'test', 'test', 'success', '不应进入恢复结果')",
|
||||
} {
|
||||
if _, err := db.DB.ExecContext(t.Context(), query); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
changed := cfg
|
||||
changed.Server.Port = 19005
|
||||
if err := changed.Save(fixture.path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, err := secretstore.Open(cfg.Paths.Config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for reference := range fixture.secrets {
|
||||
if err := store.Put(t.Context(), reference, map[string]string{"changed": "不应保留"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := os.WriteFile(fixture.certificate.CertificatePath, []byte("已替换的证书"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
for _, round := range []string{"first_restore", "repeat_restore"} {
|
||||
t.Run(round, func(t *testing.T) {
|
||||
restore := newRestoreCommand()
|
||||
restore.SetOut(&output)
|
||||
restore.SetErr(&output)
|
||||
restore.SetArgs([]string{"--input", input})
|
||||
if err := restore.ExecuteContext(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
actual := snapshotSQLiteBusiness(t, cfg)
|
||||
if !reflect.DeepEqual(actual, expected) {
|
||||
for table, want := range expected {
|
||||
if actual[table] != want {
|
||||
t.Errorf("恢复后的表或模式不一致：%s", table)
|
||||
}
|
||||
}
|
||||
if len(actual) != len(expected) {
|
||||
t.Error("恢复后的表清单数量不一致")
|
||||
}
|
||||
}
|
||||
assertSQLiteBusinessUsable(t, fixture)
|
||||
// 同一份备份可重复恢复，读取和恢复均不能修改归档清单。
|
||||
if err := validateBackupManifest(input); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func assertSQLiteBusinessUsable(t *testing.T, fixture sqliteBusinessFixture) {
|
||||
t.Helper()
|
||||
cfg, err := config.Load(fixture.path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.Server.Port != fixture.config.Server.Port || cfg.DNS != fixture.config.DNS {
|
||||
t.Fatal("服务配置或 DNS 归属未恢复")
|
||||
}
|
||||
store, err := secretstore.Open(cfg.Paths.Config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for reference, want := range fixture.secrets {
|
||||
actual, err := store.Get(t.Context(), reference)
|
||||
if err != nil || !reflect.DeepEqual(actual, want) {
|
||||
t.Fatalf("恢复后的加密凭据无法正确读取：%v", err)
|
||||
}
|
||||
if key := actual["private_key"]; key != "" {
|
||||
if _, err := ssh.ParsePrivateKey([]byte(key)); err != nil {
|
||||
t.Fatalf("恢复后的 SSH 私钥无法解析：%v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := tls.LoadX509KeyPair(fixture.certificate.CertificatePath, fixture.certificate.PrivateKeyPath); err != nil {
|
||||
t.Fatalf("恢复后的证书与私钥不可用：%v", err)
|
||||
}
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
users := repository.NewSQLUserRepository(db.DB)
|
||||
user, err := users.FindByIdentifier(t.Context(), "recovery-owner")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if user.Status != "active" || user.Role != "super_admin" || bcrypt.CompareHashAndPassword([]byte(user.PasswordHash), []byte(fixture.password)) != nil {
|
||||
t.Fatal("恢复后的用户密码或角色不可用")
|
||||
}
|
||||
member, err := users.GetUserByID(t.Context(), "member")
|
||||
if err != nil || !reflect.DeepEqual(member.Groups, []string{"测试分组"}) {
|
||||
t.Fatalf("用户分组未恢复：%v", err)
|
||||
}
|
||||
domains := repository.NewSQLDomainRepository(db.DB)
|
||||
item, err := domains.GetByID(t.Context(), "domain")
|
||||
if err != nil || item.Status != "issued" || item.ExpiresAt == nil || !item.ExpiresAt.Equal(fixture.certificate.NotAfter) {
|
||||
t.Fatalf("域名和到期时间未恢复：%v", err)
|
||||
}
|
||||
servers := repository.NewSQLServerRepository(db.DB)
|
||||
server, err := servers.GetServerByID(t.Context(), "server")
|
||||
if err != nil || server.PrivateKeyRef != "secret://ssh/business" {
|
||||
t.Fatalf("服务器凭据引用未恢复：%v", err)
|
||||
}
|
||||
tasks := repository.NewSQLTaskRepository(db.DB)
|
||||
task, err := tasks.GetTask(t.Context(), "task")
|
||||
if err != nil || task.DomainID != item.ID || task.OperatorID != user.ID || task.CertificateVersionID != fixture.certificate.Version {
|
||||
t.Fatalf("任务资源关联未恢复：%v", err)
|
||||
}
|
||||
attempts, err := tasks.ListAttempts(t.Context(), task.ID)
|
||||
if err != nil || len(attempts) != 1 || attempts[0].Status != "success" {
|
||||
t.Fatalf("任务尝试未恢复：%v", err)
|
||||
}
|
||||
var bindings int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM domain_servers WHERE domain_id = ? AND server_id = ?", item.ID, server.ID).Scan(&bindings); err != nil || bindings != 1 {
|
||||
t.Fatalf("域名部署服务器关联未恢复：%v", err)
|
||||
}
|
||||
result, err := db.DB.ExecContext(t.Context(), "INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('after-restore', 'test', 'test', 'success', '恢复后写入')")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if id, err := result.LastInsertId(); err != nil || id != 2 {
|
||||
t.Fatalf("审计自增序列未恢复：编号=%d，错误=%v", id, err)
|
||||
}
|
||||
}
|
||||
78
backend/internal/cli/sqlite_restore.go
Normal file
78
backend/internal/cli/sqlite_restore.go
Normal file
@ -0,0 +1,78 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// 调用方必须持有服务维护锁，并停止所有外部数据库访问。
|
||||
// 不能直接替换主文件后再处理旧 WAL，否则旧事务会覆盖恢复的数据。
|
||||
func prepareSQLiteRestoreTarget(ctx context.Context, path string) (err error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
info, statErr := os.Lstat(path)
|
||||
exists := statErr == nil
|
||||
if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
|
||||
return fmt.Errorf("检查 SQLite 恢复目标失败：%w", statErr)
|
||||
}
|
||||
if exists && !info.Mode().IsRegular() {
|
||||
return errors.New("SQLite 恢复目标必须是普通文件")
|
||||
}
|
||||
for _, suffix := range []string{"-wal", "-shm", "-journal"} {
|
||||
sidecar, sidecarErr := os.Lstat(path + suffix)
|
||||
if errors.Is(sidecarErr, os.ErrNotExist) {
|
||||
continue
|
||||
}
|
||||
if sidecarErr != nil {
|
||||
return fmt.Errorf("检查 SQLite 日志文件失败：%w", sidecarErr)
|
||||
}
|
||||
if !exists || !sidecar.Mode().IsRegular() {
|
||||
return fmt.Errorf("SQLite 日志状态不安全，拒绝恢复：%s", suffix)
|
||||
}
|
||||
}
|
||||
if !exists {
|
||||
return nil
|
||||
}
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("打开 SQLite 恢复目标失败：%w", err)
|
||||
}
|
||||
db.SetMaxOpenConns(1)
|
||||
defer func() { err = errors.Join(err, db.Close()) }()
|
||||
// 维护窗口仍有外部事务时立即失败，不等待后继续替换文件。
|
||||
if _, err := db.ExecContext(ctx, "PRAGMA busy_timeout = 0"); err != nil {
|
||||
return fmt.Errorf("配置 SQLite 恢复检查失败：%w", err)
|
||||
}
|
||||
var busy, logged, checkpointed int
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &logged, &checkpointed); err != nil {
|
||||
return fmt.Errorf("合并 SQLite 恢复前日志失败：%w", err)
|
||||
}
|
||||
if busy != 0 || logged != checkpointed {
|
||||
return errors.New("SQLite 仍被其他连接占用，拒绝恢复；请停止所有数据库访问")
|
||||
}
|
||||
// 由 SQLite 自身完成恢复及日志模式切换，不手动删除日志文件。
|
||||
var mode string
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA journal_mode = DELETE").Scan(&mode); err != nil {
|
||||
return fmt.Errorf("关闭 SQLite 恢复前日志失败，请停止所有数据库访问：%w", err)
|
||||
}
|
||||
if !strings.EqualFold(mode, "delete") {
|
||||
return fmt.Errorf("SQLite 恢复前日志模式未切换完成：%s", mode)
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
return fmt.Errorf("关闭 SQLite 恢复目标失败：%w", err)
|
||||
}
|
||||
for _, suffix := range []string{"-wal", "-shm", "-journal"} {
|
||||
if _, err := os.Lstat(path + suffix); !errors.Is(err, os.ErrNotExist) {
|
||||
if err != nil {
|
||||
return fmt.Errorf("复核 SQLite 日志文件失败：%w", err)
|
||||
}
|
||||
return fmt.Errorf("SQLite 日志仍存在，拒绝替换数据库：%s", suffix)
|
||||
}
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
110
backend/internal/cli/sqlite_restore_safety_test.go
Normal file
110
backend/internal/cli/sqlite_restore_safety_test.go
Normal file
@ -0,0 +1,110 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
)
|
||||
|
||||
func TestSQLiteRestoreRejectsActiveTransaction(t *testing.T) {
|
||||
for _, write := range []bool{false, true} {
|
||||
name := "reader"
|
||||
if write {
|
||||
name = "writer"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
cfg, path, input := prepareRestoreCommandFixture(t, false)
|
||||
before, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
tx, err := db.DB.BeginTx(t.Context(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if write {
|
||||
_, err = tx.ExecContext(t.Context(), "UPDATE audit_logs SET message = '占用中的事务'")
|
||||
} else {
|
||||
var count int
|
||||
err = tx.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs").Scan(&count)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
previous := configPath
|
||||
configPath = path
|
||||
t.Cleanup(func() { configPath = previous })
|
||||
command := newRestoreCommand()
|
||||
var output strings.Builder
|
||||
command.SetOut(&output)
|
||||
command.SetErr(&output)
|
||||
command.SetArgs([]string{"--input", input})
|
||||
if err := command.ExecuteContext(t.Context()); err == nil {
|
||||
t.Fatal("存在外部事务时仍执行了恢复")
|
||||
}
|
||||
after, err := os.ReadFile(path)
|
||||
if err != nil || string(after) != string(before) {
|
||||
t.Fatalf("预检失败修改了配置：%v", err)
|
||||
}
|
||||
if err := tx.Rollback(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count, err := auditActionCount(cfg.Database.DSN, "current"); err != nil || count != 1 {
|
||||
t.Fatalf("预检失败修改了原有数据：数量=%d，错误=%v", count, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLiteRestoreTargetSidecarSafety(t *testing.T) {
|
||||
for _, suffix := range []string{"-wal", "-shm", "-journal"} {
|
||||
t.Run("orphan"+suffix, func(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "absent.db")
|
||||
if err := os.WriteFile(path+suffix, []byte("保留的日志"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := prepareSQLiteRestoreTarget(t.Context(), path); err == nil {
|
||||
t.Fatal("孤立日志未被拒绝")
|
||||
}
|
||||
if data, err := os.ReadFile(path + suffix); err != nil || string(data) != "保留的日志" {
|
||||
t.Fatalf("孤立日志被修改：%v", err)
|
||||
}
|
||||
if _, err := os.Stat(path); !os.IsNotExist(err) {
|
||||
t.Fatalf("错误创建了数据库：%v", err)
|
||||
}
|
||||
})
|
||||
t.Run("directory"+suffix, func(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "current.db")
|
||||
if err := createSQLiteTestDatabase(path, "current"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Mkdir(path+suffix, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := prepareSQLiteRestoreTarget(t.Context(), path); err == nil {
|
||||
t.Fatal("非普通日志路径未被拒绝")
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("canceled", func(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "absent.db")
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
cancel()
|
||||
if err := prepareSQLiteRestoreTarget(ctx, path); err != context.Canceled {
|
||||
t.Fatalf("未保留取消错误：%v", err)
|
||||
}
|
||||
if _, err := os.Stat(path); !os.IsNotExist(err) {
|
||||
t.Fatalf("取消后创建了数据库：%v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
87
backend/internal/cli/sqlite_restore_wal_test.go
Normal file
87
backend/internal/cli/sqlite_restore_wal_test.go
Normal file
@ -0,0 +1,87 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
)
|
||||
|
||||
func TestSQLiteRestoreAfterUncleanShutdown(t *testing.T) {
|
||||
if path := os.Getenv("SCERT_SQLITE_RESTORE_CRASH_DSN"); path != "" {
|
||||
cfg := config.Default("").Database
|
||||
cfg.DSN = path
|
||||
db, err := database.Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), "UPDATE audit_logs SET action = 'wal-current' WHERE action = 'current'"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 故意不关闭数据库，模拟已提交 WAL 尚未检查点时进程退出。
|
||||
os.Exit(0)
|
||||
}
|
||||
for _, rollback := range []bool{false, true} {
|
||||
name := "restore"
|
||||
if rollback {
|
||||
name = "rollback_after_asset_failure"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
cfg, currentPath, input := prepareRestoreCommandFixture(t, rollback)
|
||||
executable, err := os.Executable()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
|
||||
defer cancel()
|
||||
child := exec.CommandContext(ctx, executable, "-test.run=^TestSQLiteRestoreAfterUncleanShutdown$", "-test.timeout=20s")
|
||||
child.Env = append(os.Environ(), "SCERT_SQLITE_RESTORE_CRASH_DSN="+cfg.Database.DSN)
|
||||
if output, err := child.CombinedOutput(); err != nil {
|
||||
t.Fatalf("异常退出夹具创建失败：%v\n%s", err, output)
|
||||
}
|
||||
if info, err := os.Stat(cfg.Database.DSN + "-wal"); err != nil || info.Size() == 0 {
|
||||
t.Fatalf("夹具未留下真实 WAL：%v", err)
|
||||
}
|
||||
previous := configPath
|
||||
configPath = currentPath
|
||||
t.Cleanup(func() { configPath = previous })
|
||||
command := newRestoreCommand()
|
||||
var output strings.Builder
|
||||
command.SetOut(&output)
|
||||
command.SetErr(&output)
|
||||
command.SetArgs([]string{"--input", input})
|
||||
err = command.ExecuteContext(t.Context())
|
||||
snapshots, snapshotErr := filepath.Glob(cfg.Database.DSN + ".before-restore-*")
|
||||
if snapshotErr != nil || len(snapshots) != 1 {
|
||||
t.Fatalf("恢复前快照缺失：数量=%d，错误=%v", len(snapshots), snapshotErr)
|
||||
}
|
||||
if count, snapshotErr := auditActionCount(snapshots[0], "wal-current"); snapshotErr != nil || count != 1 {
|
||||
t.Fatalf("恢复前快照遗漏已提交的 WAL 数据：数量=%d，错误=%v", count, snapshotErr)
|
||||
}
|
||||
want := "restored"
|
||||
if rollback {
|
||||
want = "wal-current"
|
||||
if err == nil || !strings.Contains(err.Error(), "证书恢复目标必须是普通目录") {
|
||||
t.Fatalf("未触发预期的资产恢复失败：%v", err)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Fatalf("恢复失败：%v", err)
|
||||
}
|
||||
for _, action := range []string{"restored", "current", "wal-current"} {
|
||||
expected := 0
|
||||
if action == want {
|
||||
expected = 1
|
||||
}
|
||||
if count, err := auditActionCount(cfg.Database.DSN, action); err != nil || count != expected {
|
||||
t.Errorf("异常退出后的恢复内容不正确：动作=%s，数量=%d，期望=%d，错误=%v", action, count, expected, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user