开发检查点：完成可信恢复证据与部署回归
补充部署事务恢复证据、锁归属和文件摘要校验，接入站点恢复基线，更新三数据库迁移与回归测试。
This commit is contained in:
parent
fb8ddd5002
commit
0e48dc9231
@ -105,6 +105,24 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
|
||||
taskRepo := repository.NewSQLTaskRepository(db.DB, runtimeDatabaseConfig.Driver)
|
||||
transactionRepo := repository.NewSQLDeploymentTransactionRepository(db.DB, runtimeDatabaseConfig.Driver)
|
||||
deploymentService := deployment.NewServiceWithJournal(deployment.SSHFactory{Secrets: secretStore}, transactionRepo)
|
||||
remoteValidator := certificate.RemoteValidator{}
|
||||
deploymentService.VerifyRecoveryTarget = func(ctx context.Context, server deployment.Server, baseline deployment.RecoveryBaseline, committed bool) error {
|
||||
expected := baseline.PreviousFingerprint
|
||||
if committed {
|
||||
expected = baseline.ExpectedFingerprint
|
||||
}
|
||||
if expected == "" || baseline.Endpoint == "" || len(baseline.Domains) == 0 {
|
||||
return fmt.Errorf("恢复站点基线不完整")
|
||||
}
|
||||
validation, err := remoteValidator.ValidateHTTPSAt(ctx, baseline.Endpoint, server.Address, baseline.Domains)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if validation.FingerprintSHA256 != expected {
|
||||
return fmt.Errorf("站点证书指纹不符合事务基线")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if err := deploymentService.RecoverPending(ctx, serverRepo.GetServerByID); err != nil {
|
||||
logger.Error("恢复中断部署事务失败，已保留现场等待人工核查", "error", err)
|
||||
}
|
||||
@ -128,7 +146,6 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
|
||||
certificates.SetDNSAuthorizationChecker(aliyunAuthChecker)
|
||||
|
||||
notifier := notification.New(cfg.Notification, secretStore)
|
||||
remoteValidator := certificate.RemoteValidator{}
|
||||
workflow := &CertificateWorkflow{
|
||||
Certificates: certificates,
|
||||
Domains: repo,
|
||||
|
||||
@ -290,6 +290,10 @@ func (w *CertificateWorkflow) deploy(ctx context.Context, result certificate.Iss
|
||||
server: target.Server,
|
||||
baselineFingerprint: baselineFingerprint,
|
||||
})
|
||||
targets[len(verificationTargets)-1].Baseline = deployment.RecoveryBaseline{
|
||||
Endpoint: endpoint, Domains: append([]string(nil), result.Domains...),
|
||||
PreviousFingerprint: baselineFingerprint, ExpectedFingerprint: expectedFingerprint,
|
||||
}
|
||||
}
|
||||
verification := deployment.Verification{
|
||||
Verify: func(verificationContext context.Context) error {
|
||||
|
||||
@ -19,6 +19,7 @@ import (
|
||||
//go:embed migrations/sqlite/0015_task_idempotency.sql migrations/mysql/0015_task_idempotency.sql migrations/postgres/0015_task_idempotency.sql
|
||||
//go:embed migrations/sqlite/0016_task_deployment_results.sql migrations/mysql/0016_task_deployment_results.sql migrations/postgres/0016_task_deployment_results.sql
|
||||
//go:embed migrations/sqlite/0017_deployment_transactions.sql migrations/mysql/0017_deployment_transactions.sql migrations/postgres/0017_deployment_transactions.sql
|
||||
//go:embed migrations/sqlite/0018_recovery_evidence.sql migrations/mysql/0018_recovery_evidence.sql migrations/postgres/0018_recovery_evidence.sql
|
||||
var migrationFiles embed.FS
|
||||
|
||||
func Migrate(ctx context.Context, db *sql.DB, driver string) error {
|
||||
@ -43,7 +44,7 @@ func Migrate(ctx context.Context, db *sql.DB, driver string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
const latestMigrationVersion = 17
|
||||
const latestMigrationVersion = 18
|
||||
|
||||
// currentMigrationVersion 在执行业务迁移前拒绝新版数据库及不连续的迁移历史。
|
||||
func currentMigrationVersion(ctx context.Context, db *sql.DB) (int, error) {
|
||||
@ -229,6 +230,8 @@ func migrationName(version int) string {
|
||||
return "task_deployment_results"
|
||||
case 17:
|
||||
return "deployment_transactions"
|
||||
case 18:
|
||||
return "recovery_evidence"
|
||||
default:
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
@ -0,0 +1 @@
|
||||
ALTER TABLE deployment_transactions ADD COLUMN recovery_evidence_json LONGTEXT NULL;
|
||||
@ -0,0 +1 @@
|
||||
ALTER TABLE deployment_transactions ADD COLUMN recovery_evidence_json TEXT;
|
||||
@ -0,0 +1 @@
|
||||
ALTER TABLE deployment_transactions ADD COLUMN recovery_evidence_json TEXT;
|
||||
133
backend/internal/deployment/recovery_evidence.go
Normal file
133
backend/internal/deployment/recovery_evidence.go
Normal file
@ -0,0 +1,133 @@
|
||||
package deployment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
)
|
||||
|
||||
// RecoveryBaseline 只保存站点身份，不保存证书内容、私钥或登录凭据。
|
||||
type RecoveryBaseline struct {
|
||||
Endpoint string `json:"endpoint"`
|
||||
Domains []string `json:"domains"`
|
||||
PreviousFingerprint string `json:"previous_fingerprint"`
|
||||
ExpectedFingerprint string `json:"expected_fingerprint"`
|
||||
}
|
||||
|
||||
// RecoveryEvidence 保存恢复所需的不可变身份与已核验决策。
|
||||
type RecoveryEvidence struct {
|
||||
Version int `json:"version"`
|
||||
LockToken string `json:"lock_token"`
|
||||
Address string `json:"address"`
|
||||
Port int `json:"port"`
|
||||
HostKeyFingerprint string `json:"host_key_fingerprint"`
|
||||
ValidationCommand []string `json:"validation_command"`
|
||||
ReloadCommand []string `json:"reload_command"`
|
||||
Files map[string]string `json:"files"`
|
||||
Baseline RecoveryBaseline `json:"baseline"`
|
||||
}
|
||||
|
||||
// EvidenceClient 通过原子符号链接保存锁归属，文件摘要不会将私钥传回业务层。
|
||||
type EvidenceClient interface {
|
||||
CreateOwnedLock(context.Context, string, string) error
|
||||
ReadLockOwner(context.Context, string) (string, error)
|
||||
FileDigest(context.Context, string) (string, error)
|
||||
}
|
||||
|
||||
func contentDigest(data []byte) string {
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func validDigest(value string) bool {
|
||||
decoded, err := hex.DecodeString(value)
|
||||
return err == nil && len(decoded) == sha256.Size
|
||||
}
|
||||
|
||||
func checkOwnedLock(ctx context.Context, client Client, path, token string, allowMissing bool) error {
|
||||
reader, ok := client.(EvidenceClient)
|
||||
if !ok || !validDigest(token) {
|
||||
return errors.New("缺少可验证的部署锁归属信息")
|
||||
}
|
||||
owner, err := reader.ReadLockOwner(ctx, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if owner == "" && allowMissing {
|
||||
return nil
|
||||
}
|
||||
if owner != token {
|
||||
return errors.New("部署锁不属于本次事务，已保留现场")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkFileDigest(ctx context.Context, client Client, path, expected string) error {
|
||||
reader, ok := client.(EvidenceClient)
|
||||
if !ok || !validDigest(expected) {
|
||||
return errors.New("缺少可信文件摘要")
|
||||
}
|
||||
actual, err := reader.FileDigest(ctx, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if actual != expected {
|
||||
return fmt.Errorf("文件身份与事务记录不一致：%s", path)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateRecoveryEvidence(record TransactionRecord, server Server) error {
|
||||
e := record.Evidence
|
||||
if e == nil || e.Version != 1 || !validDigest(e.LockToken) {
|
||||
return errors.New("历史事务缺少可信恢复证据，需核查现场，不自动接管")
|
||||
}
|
||||
if e.Address != server.Address || e.Port != server.Port || e.HostKeyFingerprint != server.HostKeyFingerprint ||
|
||||
!reflect.DeepEqual(e.ValidationCommand, server.ValidationCommand) || !reflect.DeepEqual(e.ReloadCommand, server.ReloadCommand) {
|
||||
return errors.New("服务器身份或恢复命令已变更，不自动接管历史事务")
|
||||
}
|
||||
if len(record.Paths) != 3 || len(record.LockPaths) != 3 || len(e.Files) != 3 {
|
||||
return errors.New("部署事务目标集合不完整")
|
||||
}
|
||||
paths := make(map[string]bool)
|
||||
for _, path := range record.Paths {
|
||||
digest := e.Files[path]
|
||||
if paths[path] || RemotePath(path) != path || len(path) < 2 || path[0] != '/' ||
|
||||
!validDigest(digest) {
|
||||
return errors.New("部署事务路径或文件摘要无效")
|
||||
}
|
||||
paths[path] = true
|
||||
}
|
||||
locks := make(map[string]bool)
|
||||
for _, lock := range record.LockPaths {
|
||||
matched := false
|
||||
for path := range paths {
|
||||
matched = matched || lock == path+".scert-lock"
|
||||
}
|
||||
if !matched || locks[lock] {
|
||||
return errors.New("部署事务锁集合无效")
|
||||
}
|
||||
locks[lock] = true
|
||||
}
|
||||
for _, temporary := range record.TemporaryPaths {
|
||||
matched := false
|
||||
for path := range paths {
|
||||
matched = matched || temporary == path+".scert-tmp-"+record.ID
|
||||
}
|
||||
if !matched {
|
||||
return errors.New("部署事务临时路径无效")
|
||||
}
|
||||
}
|
||||
seen := make(map[string]bool)
|
||||
for _, backup := range record.Backups {
|
||||
if !paths[backup.Original] || seen[backup.Original] || backup.Saved != backup.Original+".scert-backup-"+record.ID ||
|
||||
(backup.OriginalExists && !validDigest(backup.OriginalDigest)) {
|
||||
return errors.New("部署事务备份身份无效")
|
||||
}
|
||||
seen[backup.Original] = true
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -13,7 +13,8 @@ import (
|
||||
)
|
||||
|
||||
type Target struct {
|
||||
Server Server
|
||||
Server Server
|
||||
Baseline RecoveryBaseline
|
||||
}
|
||||
|
||||
type Artifact struct {
|
||||
@ -51,9 +52,10 @@ type Report struct {
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
factory ClientFactory
|
||||
journal TransactionJournal
|
||||
now func() time.Time
|
||||
factory ClientFactory
|
||||
journal TransactionJournal
|
||||
now func() time.Time
|
||||
VerifyRecoveryTarget func(context.Context, Server, RecoveryBaseline, bool) error
|
||||
}
|
||||
|
||||
func NewService(factory ClientFactory) *Service {
|
||||
@ -91,7 +93,7 @@ func (s *Service) deploy(ctx context.Context, targets []Target, artifact Artifac
|
||||
return report
|
||||
}
|
||||
for index, target := range targets {
|
||||
result := s.deployOne(ctx, target.Server, artifact, pending, index)
|
||||
result := s.deployOne(ctx, target.Server, artifact, pending, index, target.Baseline)
|
||||
report.Results = append(report.Results, result)
|
||||
if result.Status == "success" {
|
||||
report.Success++
|
||||
@ -134,7 +136,7 @@ func loadArtifact(saved SavedFiles) (Artifact, error) {
|
||||
return Artifact{Certificate: certificateData, FullChain: fullChainData, PrivateKey: privateKeyData}, nil
|
||||
}
|
||||
|
||||
func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifact, pending *[]*pendingDeployment, index int) (result Result) {
|
||||
func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifact, pending *[]*pendingDeployment, index int, baseline RecoveryBaseline) (result Result) {
|
||||
ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
|
||||
defer cancel()
|
||||
server = server.Normalize()
|
||||
@ -181,6 +183,21 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
lockPaths = append(lockPaths, file.path+".scert-lock")
|
||||
}
|
||||
sort.Strings(lockPaths)
|
||||
backupID := s.backupID()
|
||||
result.BackupID = backupID
|
||||
journalRecord, transactionEnabled := s.newTransactionRecord(ctx, server, paths, lockPaths, backupID, index, baseline)
|
||||
if transactionEnabled {
|
||||
if _, ok := client.(EvidenceClient); !ok {
|
||||
result.FailureCode = FailureUnavailable
|
||||
result.Message = "当前客户端不支持部署恢复证据"
|
||||
return result
|
||||
}
|
||||
if err := s.journal.CreateDeploymentTransaction(ctx, *journalRecord); err != nil {
|
||||
result.FailureCode = FailureUnavailable
|
||||
result.Message = fmt.Sprintf("保存部署事务现场失败：%v", err)
|
||||
return result
|
||||
}
|
||||
}
|
||||
heldLocks := make([]string, 0, len(lockPaths))
|
||||
retainLocks := false
|
||||
releaseLocks := func() {
|
||||
@ -197,6 +214,15 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
}
|
||||
released := true
|
||||
for index := len(heldLocks) - 1; index >= 0; index-- {
|
||||
if transactionEnabled {
|
||||
if err := checkOwnedLock(releaseCtx, client, heldLocks[index], journalRecord.Evidence.LockToken, true); err != nil {
|
||||
released = false
|
||||
result.Status = "failed"
|
||||
result.FailureCode = FailureLockRelease
|
||||
result.Message += fmt.Sprintf("；部署锁归属核验失败：%v", err)
|
||||
continue
|
||||
}
|
||||
}
|
||||
if err := client.Unlock(releaseCtx, heldLocks[index]); err != nil {
|
||||
released = false
|
||||
result.Status = "failed"
|
||||
@ -220,15 +246,12 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
return result
|
||||
}
|
||||
heldLocks = append(heldLocks, lockPath)
|
||||
}
|
||||
backupID := s.backupID()
|
||||
result.BackupID = backupID
|
||||
journalRecord, transactionEnabled := s.newTransactionRecord(ctx, server, paths, lockPaths, backupID, index)
|
||||
if transactionEnabled {
|
||||
if err := s.journal.CreateDeploymentTransaction(ctx, *journalRecord); err != nil {
|
||||
result.FailureCode = FailureUnavailable
|
||||
result.Message = fmt.Sprintf("保存部署事务现场失败：%v", err)
|
||||
return result
|
||||
if transactionEnabled {
|
||||
if err := client.(EvidenceClient).CreateOwnedLock(ctx, lockPath, journalRecord.Evidence.LockToken); err != nil {
|
||||
result.FailureCode = FailureUnavailable
|
||||
result.Message = fmt.Sprintf("保存部署锁归属失败：%v", err)
|
||||
return result
|
||||
}
|
||||
}
|
||||
}
|
||||
persistTransaction := func(persistContext context.Context, phase string) error {
|
||||
@ -249,6 +272,11 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
result.Message += fmt.Sprintf("；保存部署事务最终状态失败：%v", err)
|
||||
}
|
||||
}
|
||||
if err := persistTransaction(ctx, TransactionPhaseLocked); err != nil {
|
||||
result.FailureCode = FailureUnavailable
|
||||
result.Message = fmt.Sprintf("保存部署锁阶段失败：%v", err)
|
||||
return result
|
||||
}
|
||||
backups := make([]backupFile, 0, len(paths))
|
||||
temporary := make([]string, 0, len(paths))
|
||||
installed := make([]string, 0, len(paths))
|
||||
@ -258,6 +286,9 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
if err := reconnectDeploymentClient(rollbackContext, s.factory, server, &client); err != nil {
|
||||
return fmt.Errorf("重新连接服务器以回滚失败：%w", err)
|
||||
}
|
||||
if transactionEnabled {
|
||||
return restoreTransactionFiles(rollbackContext, client, *journalRecord)
|
||||
}
|
||||
var rollbackErrors []error
|
||||
for _, path := range temporary {
|
||||
if err := client.Remove(rollbackContext, path); err != nil {
|
||||
@ -300,10 +331,18 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
if !retainLocks {
|
||||
result.FailureCode = FailureRuntimeUnknown
|
||||
}
|
||||
retainLocks = true
|
||||
// 命令可能已部分执行；文件回滚并不证明运行中的服务已经恢复。
|
||||
message += "；服务运行状态未确认，请检查服务并在确认旧配置后重新加载"
|
||||
}
|
||||
result.Message = message
|
||||
if !retainLocks {
|
||||
if err := persistTransaction(context.WithoutCancel(ctx), TransactionPhaseRollbackVerified); err != nil {
|
||||
retainLocks = true
|
||||
result.FailureCode = FailureRecoveryUnknown
|
||||
result.Message += fmt.Sprintf("；保存回滚核验阶段失败：%v", err)
|
||||
}
|
||||
}
|
||||
releaseLocks()
|
||||
if retainLocks || result.FailureCode == FailureLockRelease {
|
||||
finishTransaction(TransactionStatusRecoveryRequired, result.Message)
|
||||
@ -317,9 +356,6 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
temporaryPath := file.path + ".scert-tmp-" + backupID
|
||||
// 上传失败也可能留下半份私钥，必须在开始上传前登记清理路径。
|
||||
temporary = append(temporary, temporaryPath)
|
||||
if transactionEnabled {
|
||||
journalRecord.TemporaryPaths = append(journalRecord.TemporaryPaths, temporaryPath)
|
||||
}
|
||||
if err := client.Upload(ctx, temporaryPath, file.data, file.mode); err != nil {
|
||||
var notCreated *uploadNotCreatedError
|
||||
if errors.As(err, ¬Created) {
|
||||
@ -344,7 +380,14 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
return failed(fmt.Sprintf("备份目标已存在或无法检查，已停止替换：%s：%v", backupPath, checkErr))
|
||||
}
|
||||
if transactionEnabled {
|
||||
journalRecord.Backups = append(journalRecord.Backups, TransactionBackup{Original: file.path, Saved: backupPath, OriginalExists: exists})
|
||||
originalDigest := ""
|
||||
if exists {
|
||||
originalDigest, err = client.(EvidenceClient).FileDigest(ctx, file.path)
|
||||
if err != nil {
|
||||
return failed(fmt.Sprintf("读取旧证书文件摘要失败：%v", err))
|
||||
}
|
||||
}
|
||||
journalRecord.Backups = append(journalRecord.Backups, TransactionBackup{Original: file.path, Saved: backupPath, OriginalExists: exists, OriginalDigest: originalDigest})
|
||||
if err := persistTransaction(ctx, TransactionPhaseReplacing); err != nil {
|
||||
return failed(fmt.Sprintf("保存替换阶段失败：%v", err))
|
||||
}
|
||||
@ -430,7 +473,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
return result
|
||||
}
|
||||
transaction.commit = func(commitContext context.Context) Result {
|
||||
if err := persistTransaction(commitContext, "committing"); err != nil {
|
||||
if err := persistTransaction(commitContext, TransactionPhaseCommitting); err != nil {
|
||||
result.Status = "failed"
|
||||
result.FailureCode = FailureCleanup
|
||||
result.Message = fmt.Sprintf("保存提交阶段失败：%v", err)
|
||||
@ -443,6 +486,11 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
return result
|
||||
}
|
||||
result.Message = "证书部署、配置校验并 reload 成功"
|
||||
if err := persistTransaction(commitContext, TransactionPhaseCommitVerified); err != nil {
|
||||
result.Status = "failed"
|
||||
result.FailureCode = FailureCleanup
|
||||
result.Message = fmt.Sprintf("保存提交核验阶段失败：%v", err)
|
||||
}
|
||||
return result
|
||||
}
|
||||
transaction.finish = func(retain bool) Result {
|
||||
@ -450,6 +498,13 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
return result
|
||||
}
|
||||
finished = true
|
||||
if !retain && result.Status == "failed" && result.FailureCode == FailureRecoveryPending {
|
||||
if err := persistTransaction(context.WithoutCancel(ctx), TransactionPhaseRollbackVerified); err != nil {
|
||||
retain = true
|
||||
result.FailureCode = FailureRecoveryUnknown
|
||||
result.Message = fmt.Sprintf("保存回滚核验阶段失败：%v", err)
|
||||
}
|
||||
}
|
||||
if !closed {
|
||||
retainLocks = retain
|
||||
releaseLocks()
|
||||
@ -494,7 +549,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
|
||||
return result
|
||||
}
|
||||
|
||||
func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths []remoteFile, lockPaths []string, id string, index int) (*TransactionRecord, bool) {
|
||||
func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths []remoteFile, lockPaths []string, id string, index int, baseline RecoveryBaseline) (*TransactionRecord, bool) {
|
||||
transactionContext, ok := TransactionContextFromContext(ctx)
|
||||
if !ok || s == nil || s.journal == nil {
|
||||
return nil, false
|
||||
@ -509,12 +564,22 @@ func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths
|
||||
if s.now != nil {
|
||||
now = s.now().UTC()
|
||||
}
|
||||
files := make(map[string]string, len(paths))
|
||||
for _, file := range paths {
|
||||
files[file.path] = contentDigest(file.data)
|
||||
}
|
||||
return &TransactionRecord{
|
||||
ID: id, TaskID: transactionContext.TaskID, AttemptID: transactionContext.AttemptID,
|
||||
AttemptNumber: transactionContext.AttemptNumber, ResultIndex: index,
|
||||
ServerID: server.ID, ServerName: server.Name, Paths: pathValues,
|
||||
TemporaryPaths: temporaryPaths, LockPaths: append([]string(nil), lockPaths...),
|
||||
Phase: TransactionPhaseLocked, Status: TransactionStatusPending, UpdatedAt: now,
|
||||
Phase: TransactionPhaseAcquiring, Status: TransactionStatusPending, UpdatedAt: now,
|
||||
Evidence: &RecoveryEvidence{
|
||||
Version: 1, LockToken: contentDigest([]byte("scert-lock:" + id)),
|
||||
Address: server.Address, Port: server.Port, HostKeyFingerprint: server.HostKeyFingerprint,
|
||||
ValidationCommand: append([]string(nil), server.ValidationCommand...),
|
||||
ReloadCommand: append([]string(nil), server.ReloadCommand...), Files: files, Baseline: baseline,
|
||||
},
|
||||
}, true
|
||||
}
|
||||
|
||||
@ -532,24 +597,29 @@ func (s *Service) RecoverPending(ctx context.Context, resolve func(context.Conte
|
||||
}
|
||||
var recoveryErrors []error
|
||||
for _, record := range records {
|
||||
if err := s.recoverTransaction(ctx, record, resolve); err != nil {
|
||||
if err := s.recoverTransaction(ctx, &record, resolve); err != nil {
|
||||
recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 恢复失败：%w", record.ID, err))
|
||||
record.Status = TransactionStatusRecoveryRequired
|
||||
record.Phase = TransactionStatusRecoveryRequired
|
||||
record.Message = err.Error()
|
||||
if updateErr := s.journal.UpdateDeploymentTransaction(context.WithoutCancel(ctx), record); updateErr != nil {
|
||||
recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 保存人工核查状态失败：%w", record.ID, updateErr))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := s.journal.FinishDeploymentTransaction(context.WithoutCancel(ctx), record.ID, TransactionStatusRolledBack, "服务启动已完成中断部署事务回滚"); err != nil {
|
||||
status := TransactionStatusRolledBack
|
||||
message := "服务启动已完成中断部署事务回滚"
|
||||
if record.Phase == TransactionPhaseCommitting || record.Phase == TransactionPhaseCommitVerified {
|
||||
status = TransactionStatusCommitted
|
||||
message = "服务启动已完成中断部署事务提交清理"
|
||||
}
|
||||
if err := s.journal.FinishDeploymentTransaction(context.WithoutCancel(ctx), record.ID, status, message); err != nil {
|
||||
recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 保存恢复结果失败：%w", record.ID, err))
|
||||
}
|
||||
}
|
||||
return errors.Join(recoveryErrors...)
|
||||
}
|
||||
|
||||
func (s *Service) recoverTransaction(ctx context.Context, record TransactionRecord, resolve func(context.Context, string) (Server, error)) error {
|
||||
func (s *Service) recoverTransaction(ctx context.Context, record *TransactionRecord, resolve func(context.Context, string) (Server, error)) error {
|
||||
server, err := resolve(ctx, record.ServerID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取服务器配置失败：%w", err)
|
||||
@ -565,56 +635,17 @@ func (s *Service) recoverTransaction(ctx context.Context, record TransactionReco
|
||||
return fmt.Errorf("连接服务器失败：%w", err)
|
||||
}
|
||||
defer client.Close()
|
||||
for _, lockPath := range record.LockPaths {
|
||||
exists, checkErr := client.Exists(recoveryContext, lockPath)
|
||||
if checkErr != nil || !exists {
|
||||
if checkErr == nil {
|
||||
checkErr = errors.New("部署锁不存在")
|
||||
}
|
||||
return fmt.Errorf("无法证明事务仍拥有部署锁 %s：%w", lockPath, checkErr)
|
||||
}
|
||||
if err := validateRecoveryEvidence(*record, server); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, temporaryPath := range record.TemporaryPaths {
|
||||
if exists, checkErr := client.Exists(recoveryContext, temporaryPath); checkErr != nil {
|
||||
return fmt.Errorf("检查临时文件失败：%w", checkErr)
|
||||
} else if exists {
|
||||
if err := client.Remove(recoveryContext, temporaryPath); err != nil {
|
||||
return fmt.Errorf("删除临时文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
if record.Phase == TransactionPhaseAcquiring || record.Phase == TransactionPhaseRollbackVerified || record.Phase == TransactionPhaseCommitVerified {
|
||||
return releaseTransactionLocks(recoveryContext, client, *record, true)
|
||||
}
|
||||
for index := len(record.Backups) - 1; index >= 0; index-- {
|
||||
backup := record.Backups[index]
|
||||
backupExists, checkErr := client.Exists(recoveryContext, backup.Saved)
|
||||
if checkErr != nil {
|
||||
return fmt.Errorf("检查备份文件失败：%w", checkErr)
|
||||
}
|
||||
targetExists, checkErr := client.Exists(recoveryContext, backup.Original)
|
||||
if checkErr != nil {
|
||||
return fmt.Errorf("检查目标文件失败：%w", checkErr)
|
||||
}
|
||||
if backup.OriginalExists {
|
||||
if !backupExists {
|
||||
if targetExists {
|
||||
continue
|
||||
}
|
||||
return fmt.Errorf("备份文件和目标文件均不存在：%s", backup.Original)
|
||||
}
|
||||
if targetExists {
|
||||
if err := client.Remove(recoveryContext, backup.Original); err != nil {
|
||||
return fmt.Errorf("删除待恢复目标文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
if err := client.Rename(recoveryContext, backup.Saved, backup.Original); err != nil {
|
||||
return fmt.Errorf("恢复备份文件失败：%w", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if targetExists {
|
||||
if err := client.Remove(recoveryContext, backup.Original); err != nil {
|
||||
return fmt.Errorf("删除新增目标文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
if record.Phase == TransactionPhaseCommitting {
|
||||
return s.recoverCommittedTransaction(recoveryContext, record, server, client)
|
||||
}
|
||||
if err := restoreTransactionFiles(recoveryContext, client, *record); err != nil {
|
||||
return err
|
||||
}
|
||||
if record.ReloadAttempted {
|
||||
for _, command := range [][]string{server.ValidationCommand, server.ReloadCommand} {
|
||||
@ -626,15 +657,132 @@ func (s *Service) recoverTransaction(ctx context.Context, record TransactionReco
|
||||
return fmt.Errorf("恢复服务失败：%w", runErr)
|
||||
}
|
||||
}
|
||||
if s.VerifyRecoveryTarget == nil || record.Evidence.Baseline.PreviousFingerprint == "" {
|
||||
return errors.New("缺少可信站点恢复基线，已保留现场等待人工核查")
|
||||
}
|
||||
if err := s.VerifyRecoveryTarget(recoveryContext, server, record.Evidence.Baseline, false); err != nil {
|
||||
return fmt.Errorf("恢复站点核验失败：%w", err)
|
||||
}
|
||||
}
|
||||
if err := s.persistRecoveryPhase(recoveryContext, record, TransactionPhaseRollbackVerified); err != nil {
|
||||
return err
|
||||
}
|
||||
return releaseTransactionLocks(recoveryContext, client, *record, true)
|
||||
}
|
||||
|
||||
func (s *Service) recoverCommittedTransaction(ctx context.Context, record *TransactionRecord, server Server, client Client) error {
|
||||
for _, lockPath := range record.LockPaths {
|
||||
if err := checkOwnedLock(ctx, client, lockPath, record.Evidence.LockToken, false); err != nil {
|
||||
return fmt.Errorf("无法证明提交事务仍拥有部署锁 %s：%w", lockPath, err)
|
||||
}
|
||||
}
|
||||
for _, path := range record.Paths {
|
||||
if err := checkFileDigest(ctx, client, path, record.Evidence.Files[path]); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if s.VerifyRecoveryTarget == nil || record.Evidence.Baseline.ExpectedFingerprint == "" {
|
||||
return errors.New("缺少可信提交站点基线，已保留现场等待人工核查")
|
||||
}
|
||||
if err := s.VerifyRecoveryTarget(ctx, server, record.Evidence.Baseline, true); err != nil {
|
||||
return fmt.Errorf("提交站点核验失败：%w", err)
|
||||
}
|
||||
for _, temporary := range record.TemporaryPaths {
|
||||
if err := client.Remove(ctx, temporary); err != nil {
|
||||
return fmt.Errorf("清理提交临时文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
for _, backup := range record.Backups {
|
||||
if exists, err := client.Exists(ctx, backup.Saved); err != nil {
|
||||
return err
|
||||
} else if exists {
|
||||
if err := checkFileDigest(ctx, client, backup.Saved, backup.OriginalDigest); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := client.Remove(ctx, backup.Saved); err != nil {
|
||||
return fmt.Errorf("清理提交备份失败：%w", err)
|
||||
}
|
||||
}
|
||||
if err := s.persistRecoveryPhase(ctx, record, TransactionPhaseCommitVerified); err != nil {
|
||||
return err
|
||||
}
|
||||
return releaseTransactionLocks(ctx, client, *record, true)
|
||||
}
|
||||
|
||||
func restoreTransactionFiles(ctx context.Context, client Client, record TransactionRecord) error {
|
||||
for _, temporaryPath := range record.TemporaryPaths {
|
||||
if exists, err := client.Exists(ctx, temporaryPath); err != nil {
|
||||
return fmt.Errorf("检查临时文件失败：%w", err)
|
||||
} else if exists {
|
||||
if err := client.Remove(ctx, temporaryPath); err != nil {
|
||||
return fmt.Errorf("删除临时文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
for index := len(record.Backups) - 1; index >= 0; index-- {
|
||||
backup := record.Backups[index]
|
||||
backupExists, err := client.Exists(ctx, backup.Saved)
|
||||
if err != nil {
|
||||
return fmt.Errorf("检查备份文件失败：%w", err)
|
||||
}
|
||||
targetExists, err := client.Exists(ctx, backup.Original)
|
||||
if err != nil {
|
||||
return fmt.Errorf("检查目标文件失败：%w", err)
|
||||
}
|
||||
if backup.OriginalExists {
|
||||
if !backupExists {
|
||||
if !targetExists {
|
||||
return fmt.Errorf("备份文件和目标文件均不存在：%s", backup.Original)
|
||||
}
|
||||
if err := checkFileDigest(ctx, client, backup.Original, backup.OriginalDigest); err != nil {
|
||||
return fmt.Errorf("无法确认目标文件已恢复：%w", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if targetExists {
|
||||
if err := client.Remove(ctx, backup.Original); err != nil {
|
||||
return fmt.Errorf("删除待恢复目标文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
if err := client.Rename(ctx, backup.Saved, backup.Original); err != nil {
|
||||
return fmt.Errorf("恢复备份文件失败：%w", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if targetExists {
|
||||
if err := client.Remove(ctx, backup.Original); err != nil {
|
||||
return fmt.Errorf("删除新增目标文件失败：%w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func releaseTransactionLocks(ctx context.Context, client Client, record TransactionRecord, verifyOwner bool) error {
|
||||
for index := len(record.LockPaths) - 1; index >= 0; index-- {
|
||||
if err := client.Unlock(recoveryContext, record.LockPaths[index]); err != nil {
|
||||
lockPath := record.LockPaths[index]
|
||||
if verifyOwner {
|
||||
if err := checkOwnedLock(ctx, client, lockPath, record.Evidence.LockToken, false); err != nil {
|
||||
return fmt.Errorf("无法证明事务仍拥有部署锁 %s：%w", lockPath, err)
|
||||
}
|
||||
}
|
||||
if err := client.Unlock(ctx, lockPath); err != nil {
|
||||
return fmt.Errorf("释放部署锁失败：%w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) persistRecoveryPhase(ctx context.Context, record *TransactionRecord, phase string) error {
|
||||
record.Phase = phase
|
||||
record.UpdatedAt = time.Now().UTC()
|
||||
if s.now != nil {
|
||||
record.UpdatedAt = s.now().UTC()
|
||||
}
|
||||
return s.journal.UpdateDeploymentTransaction(ctx, *record)
|
||||
}
|
||||
|
||||
func cleanupBackups(ctx context.Context, client Client, backups []backupFile, paths []remoteFile) error {
|
||||
seen := make(map[string]struct{}, len(backups))
|
||||
for _, item := range backups {
|
||||
|
||||
@ -63,10 +63,35 @@ func (c *memoryClient) Lock(_ context.Context, path string) error {
|
||||
}
|
||||
|
||||
func (c *memoryClient) Unlock(_ context.Context, path string) error {
|
||||
delete(c.files, path+"/owner")
|
||||
delete(c.files, path)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *memoryClient) CreateOwnedLock(_ context.Context, path, token string) error {
|
||||
if _, exists := c.files[path]; !exists {
|
||||
return errors.New("模拟锁不存在")
|
||||
}
|
||||
c.files[path+"/owner"] = memoryFile{data: []byte(token), mode: 0o600}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *memoryClient) ReadLockOwner(_ context.Context, path string) (string, error) {
|
||||
file, ok := c.files[path+"/owner"]
|
||||
if !ok {
|
||||
return "", nil
|
||||
}
|
||||
return string(file.data), nil
|
||||
}
|
||||
|
||||
func (c *memoryClient) FileDigest(_ context.Context, path string) (string, error) {
|
||||
file, ok := c.files[path]
|
||||
if !ok {
|
||||
return "", errors.New("模拟文件不存在")
|
||||
}
|
||||
return contentDigest(file.data), nil
|
||||
}
|
||||
|
||||
func (c *memoryClient) Exists(_ context.Context, path string) (bool, error) {
|
||||
_, ok := c.files[path]
|
||||
return ok, nil
|
||||
|
||||
@ -2,7 +2,9 @@ package deployment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@ -302,12 +304,65 @@ func (c *sshClientAdapter) Lock(ctx context.Context, path string) error {
|
||||
}
|
||||
|
||||
func (c *sshClientAdapter) Unlock(ctx context.Context, path string) error {
|
||||
return c.fileOperation(ctx, func() error { return c.sftp.RemoveDirectory(path) })
|
||||
return c.fileOperation(ctx, func() error {
|
||||
if err := c.sftp.Remove(path + "/owner"); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return err
|
||||
}
|
||||
return c.sftp.RemoveDirectory(path)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *sshClientAdapter) CreateOwnedLock(ctx context.Context, path, token string) error {
|
||||
if !validDigest(token) {
|
||||
return errors.New("部署锁令牌格式无效")
|
||||
}
|
||||
return c.Upload(ctx, path+"/owner", []byte(token), 0o600)
|
||||
}
|
||||
|
||||
func (c *sshClientAdapter) ReadLockOwner(ctx context.Context, path string) (string, error) {
|
||||
return sshOperation(ctx, c, func() (string, error) {
|
||||
file, err := c.sftp.Open(path + "/owner")
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer file.Close()
|
||||
data, err := io.ReadAll(file)
|
||||
return strings.TrimSpace(string(data)), err
|
||||
})
|
||||
}
|
||||
|
||||
func (c *sshClientAdapter) FileDigest(ctx context.Context, path string) (string, error) {
|
||||
return sshOperation(ctx, c, func() (string, error) {
|
||||
info, err := c.sftp.Lstat(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !info.Mode().IsRegular() || info.Size() > 16*1024*1024 {
|
||||
return "", errors.New("证书文件类型或大小无效")
|
||||
}
|
||||
file, err := c.sftp.Open(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer file.Close()
|
||||
hash := sha256.New()
|
||||
count, err := io.Copy(hash, io.LimitReader(file, 16*1024*1024+1))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if count > 16*1024*1024 {
|
||||
return "", errors.New("证书文件超过摘要读取上限")
|
||||
}
|
||||
return hex.EncodeToString(hash.Sum(nil)), nil
|
||||
})
|
||||
}
|
||||
|
||||
func (c *sshClientAdapter) Exists(ctx context.Context, path string) (bool, error) {
|
||||
return sshOperation(ctx, c, func() (bool, error) {
|
||||
_, err := c.sftp.Stat(path)
|
||||
_, err := c.sftp.Lstat(path)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
@ -11,6 +11,7 @@ type TransactionBackup struct {
|
||||
Original string `json:"original"`
|
||||
Saved string `json:"saved"`
|
||||
OriginalExists bool `json:"original_exists"`
|
||||
OriginalDigest string `json:"original_digest,omitempty"`
|
||||
}
|
||||
|
||||
// TransactionRecord 保存远程部署事务的可恢复现场，不包含证书内容或凭据。
|
||||
@ -32,6 +33,7 @@ type TransactionRecord struct {
|
||||
Status string
|
||||
Message string
|
||||
UpdatedAt time.Time
|
||||
Evidence *RecoveryEvidence
|
||||
}
|
||||
|
||||
// TransactionJournal 持久化部署现场，保证进程重启后仍能识别未提交事务。
|
||||
@ -48,13 +50,17 @@ const (
|
||||
TransactionStatusRolledBack = "rolled_back"
|
||||
TransactionStatusRecoveryRequired = "recovery_required"
|
||||
|
||||
TransactionPhaseLocked = "locked"
|
||||
TransactionPhaseUploaded = "uploaded"
|
||||
TransactionPhaseReplacing = "replacing"
|
||||
TransactionPhaseInstalled = "installed"
|
||||
TransactionPhaseReloadAttempted = "reload_attempted"
|
||||
TransactionPhaseCommitted = "committed"
|
||||
TransactionPhaseRolledBack = "rolled_back"
|
||||
TransactionPhaseLocked = "locked"
|
||||
TransactionPhaseAcquiring = "acquiring"
|
||||
TransactionPhaseRollbackVerified = "rollback_verified"
|
||||
TransactionPhaseCommitVerified = "commit_verified"
|
||||
TransactionPhaseCommitting = "committing"
|
||||
TransactionPhaseUploaded = "uploaded"
|
||||
TransactionPhaseReplacing = "replacing"
|
||||
TransactionPhaseInstalled = "installed"
|
||||
TransactionPhaseReloadAttempted = "reload_attempted"
|
||||
TransactionPhaseCommitted = "committed"
|
||||
TransactionPhaseRolledBack = "rolled_back"
|
||||
)
|
||||
|
||||
type transactionContextKey struct{}
|
||||
|
||||
@ -40,31 +40,37 @@ func (j *memoryTransactionJournal) ListPendingDeploymentTransactions(context.Con
|
||||
|
||||
func TestRecoverPendingRollsBackOwnedDeployment(t *testing.T) {
|
||||
server := testServer("recovery")
|
||||
server = server.Normalize()
|
||||
client := newMemoryClient()
|
||||
id := "recovery-transaction"
|
||||
paths := []string{server.CertificatePath, server.CertificateChainPath, server.PrivateKeyPath}
|
||||
locks := make([]string, 0, len(paths))
|
||||
backups := make([]TransactionBackup, 0, len(paths))
|
||||
temporary := make([]string, 0, len(paths))
|
||||
files := make(map[string]string, len(paths))
|
||||
for _, path := range paths {
|
||||
lockPath := path + ".scert-lock"
|
||||
backupPath := path + ".scert-backup-" + id
|
||||
temporaryPath := path + ".scert-tmp-" + id
|
||||
locks = append(locks, lockPath)
|
||||
backups = append(backups, TransactionBackup{Original: path, Saved: backupPath, OriginalExists: true})
|
||||
backups = append(backups, TransactionBackup{Original: path, Saved: backupPath, OriginalExists: true, OriginalDigest: contentDigest([]byte("old"))})
|
||||
temporary = append(temporary, temporaryPath)
|
||||
client.files[lockPath] = memoryFile{}
|
||||
client.files[lockPath+"/owner"] = memoryFile{data: []byte(contentDigest([]byte("scert-lock:" + id))), mode: 0o600}
|
||||
client.files[backupPath] = memoryFile{data: []byte("old"), mode: 0o600}
|
||||
client.files[path] = memoryFile{data: []byte("new"), mode: 0o600}
|
||||
client.files[temporaryPath] = memoryFile{data: []byte("partial"), mode: 0o600}
|
||||
files[path] = contentDigest([]byte("old"))
|
||||
}
|
||||
journal := &memoryTransactionJournal{pending: []TransactionRecord{{
|
||||
ID: id, TaskID: "task", AttemptID: "attempt", AttemptNumber: 1, ResultIndex: 0,
|
||||
ServerID: server.ID, ServerName: server.Name, Paths: paths, TemporaryPaths: temporary,
|
||||
InstalledPaths: paths, Backups: backups, LockPaths: locks, ReloadAttempted: true,
|
||||
Phase: TransactionPhaseReloadAttempted, Status: TransactionStatusPending,
|
||||
Evidence: &RecoveryEvidence{Version: 1, LockToken: contentDigest([]byte("scert-lock:" + id)), Address: server.Address, Port: server.Port, HostKeyFingerprint: server.HostKeyFingerprint, ValidationCommand: append([]string(nil), server.ValidationCommand...), ReloadCommand: append([]string(nil), server.ReloadCommand...), Files: files, Baseline: RecoveryBaseline{PreviousFingerprint: "old", Endpoint: "https://example.com", Domains: []string{"example.com"}}},
|
||||
}}}
|
||||
service := NewServiceWithJournal(memoryFactory{clients: map[string]*memoryClient{server.ID: client}}, journal)
|
||||
service.VerifyRecoveryTarget = func(context.Context, Server, RecoveryBaseline, bool) error { return nil }
|
||||
if err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { return server, nil }); err != nil {
|
||||
t.Fatalf("事务恢复失败：%v", err)
|
||||
}
|
||||
|
||||
@ -41,11 +41,11 @@ func (r *SQLDeploymentTransactionRepository) CreateDeploymentTransaction(ctx con
|
||||
(id, task_id, attempt_id, attempt_number, result_index, server_id, server_name,
|
||||
certificate_path, certificate_chain_path, private_key_path, temporary_paths_json,
|
||||
installed_paths_json, backups_json, lock_paths_json, reload_attempted, phase, status,
|
||||
message, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
|
||||
message, updated_at, recovery_evidence_json)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
|
||||
item.ID, item.TaskID, item.AttemptID, item.AttemptNumber, item.ResultIndex, item.ServerID, item.ServerName,
|
||||
values.paths[0], values.paths[1], values.paths[2], values.temporary, values.installed, values.backups,
|
||||
values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt)
|
||||
values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, values.evidence)
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存部署事务现场失败：%w", err)
|
||||
}
|
||||
@ -66,9 +66,9 @@ func (r *SQLDeploymentTransactionRepository) UpdateDeploymentTransaction(ctx con
|
||||
result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET
|
||||
certificate_path = ?, certificate_chain_path = ?, private_key_path = ?,
|
||||
temporary_paths_json = ?, installed_paths_json = ?, backups_json = ?, lock_paths_json = ?,
|
||||
reload_attempted = ?, phase = ?, status = ?, message = ?, updated_at = ? WHERE id = ?`),
|
||||
reload_attempted = ?, phase = ?, status = ?, message = ?, updated_at = ?, recovery_evidence_json = ? WHERE id = ?`),
|
||||
values.paths[0], values.paths[1], values.paths[2], values.temporary, values.installed, values.backups,
|
||||
values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, item.ID)
|
||||
values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, values.evidence, item.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新部署事务现场失败：%w", err)
|
||||
}
|
||||
@ -91,8 +91,8 @@ func (r *SQLDeploymentTransactionRepository) FinishDeploymentTransaction(ctx con
|
||||
if id == "" || status == "" {
|
||||
return errors.New("部署事务状态参数不能为空")
|
||||
}
|
||||
result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET status = ?, phase = ?, message = ?, updated_at = ? WHERE id = ?`),
|
||||
status, status, message, time.Now().UTC(), id)
|
||||
result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET status = ?, phase = CASE WHEN ? = 'recovery_required' THEN phase ELSE ? END, message = ?, updated_at = ? WHERE id = ?`),
|
||||
status, status, status, message, time.Now().UTC(), id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("完成部署事务记录失败：%w", err)
|
||||
}
|
||||
@ -113,7 +113,7 @@ func (r *SQLDeploymentTransactionRepository) ListPendingDeploymentTransactions(c
|
||||
rows, err := r.db.QueryContext(ctx, r.query(`SELECT id, task_id, attempt_id, attempt_number, result_index,
|
||||
server_id, server_name, certificate_path, certificate_chain_path, private_key_path,
|
||||
temporary_paths_json, installed_paths_json, backups_json, lock_paths_json,
|
||||
reload_attempted, phase, status, message, updated_at
|
||||
reload_attempted, phase, status, message, updated_at, recovery_evidence_json
|
||||
FROM deployment_transactions WHERE status IN ('pending', 'recovery_required') ORDER BY updated_at, id`))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("查询待恢复部署事务失败：%w", err)
|
||||
@ -140,6 +140,7 @@ type deploymentTransactionValues struct {
|
||||
backups string
|
||||
locks string
|
||||
updatedAt time.Time
|
||||
evidence string
|
||||
}
|
||||
|
||||
func validateDeploymentTransaction(item deployment.TransactionRecord) error {
|
||||
@ -159,6 +160,10 @@ func validateDeploymentTransaction(item deployment.TransactionRecord) error {
|
||||
}
|
||||
|
||||
func encodeDeploymentTransaction(item deployment.TransactionRecord) (deploymentTransactionValues, error) {
|
||||
evidence, err := json.Marshal(item.Evidence)
|
||||
if err != nil {
|
||||
return deploymentTransactionValues{}, fmt.Errorf("编码部署恢复证据失败：%w", err)
|
||||
}
|
||||
paths := [3]string{strings.TrimSpace(item.Paths[0]), strings.TrimSpace(item.Paths[1]), strings.TrimSpace(item.Paths[2])}
|
||||
temporary, err := json.Marshal(item.TemporaryPaths)
|
||||
if err != nil {
|
||||
@ -183,7 +188,7 @@ func encodeDeploymentTransaction(item deployment.TransactionRecord) (deploymentT
|
||||
updatedAt = updatedAt.UTC()
|
||||
}
|
||||
return deploymentTransactionValues{
|
||||
paths: paths, temporary: string(temporary), installed: string(installed), backups: string(backups), locks: string(locks), updatedAt: updatedAt,
|
||||
paths: paths, temporary: string(temporary), installed: string(installed), backups: string(backups), locks: string(locks), updatedAt: updatedAt, evidence: string(evidence),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@ -196,12 +201,18 @@ func scanDeploymentTransaction(row deploymentTransactionScanner) (deployment.Tra
|
||||
var paths [3]string
|
||||
var temporary, installed, backups, locks string
|
||||
var reloadAttempted int
|
||||
var evidence sql.NullString
|
||||
if err := row.Scan(&item.ID, &item.TaskID, &item.AttemptID, &item.AttemptNumber, &item.ResultIndex,
|
||||
&item.ServerID, &item.ServerName, &paths[0], &paths[1], &paths[2], &temporary, &installed, &backups, &locks,
|
||||
&reloadAttempted, &item.Phase, &item.Status, &item.Message, &item.UpdatedAt); err != nil {
|
||||
&reloadAttempted, &item.Phase, &item.Status, &item.Message, &item.UpdatedAt, &evidence); err != nil {
|
||||
return deployment.TransactionRecord{}, fmt.Errorf("读取部署事务失败：%w", err)
|
||||
}
|
||||
item.Paths = paths[:]
|
||||
if evidence.Valid && evidence.String != "" {
|
||||
if err := json.Unmarshal([]byte(evidence.String), &item.Evidence); err != nil {
|
||||
return deployment.TransactionRecord{}, fmt.Errorf("解析部署恢复证据失败：%w", err)
|
||||
}
|
||||
}
|
||||
item.ReloadAttempted = reloadAttempted != 0
|
||||
if err := json.Unmarshal([]byte(temporary), &item.TemporaryPaths); err != nil {
|
||||
return deployment.TransactionRecord{}, fmt.Errorf("解析部署临时路径失败：%w", err)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user