From 918e3f6d253d433920b5dcd2be36fa70f911e4b6 Mon Sep 17 00:00:00 2001 From: xuans Date: Thu, 1 Oct 2026 02:36:58 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E9=83=A8=E7=BD=B2=E4=BA=8B?= =?UTF-8?q?=E5=8A=A1=E6=8C=81=E4=B9=85=E5=8C=96=E4=B8=8E=E4=B8=AD=E6=96=AD?= =?UTF-8?q?=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 接入任务上下文、部署事务日志、启动恢复流程及恢复回归测试，保留无法证明归属时的人工核查状态。 --- backend/internal/app/app.go | 7 +- backend/internal/app/certificate_tasks.go | 3 + backend/internal/deployment/service.go | 250 +++++++++++++++++- backend/internal/deployment/transaction.go | 20 +- .../deployment/transaction_recovery_test.go | 105 ++++++++ backend/internal/httpapi/task_api.go | 3 + 6 files changed, 380 insertions(+), 8 deletions(-) create mode 100644 backend/internal/deployment/transaction_recovery_test.go diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 6877e70..29dbc42 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -103,6 +103,11 @@ 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) + if err := deploymentService.RecoverPending(ctx, serverRepo.GetServerByID); err != nil { + logger.Error("恢复中断部署事务失败，已保留现场等待人工核查", "error", err) + } if recovered, err := taskRepo.RecoverInterruptedTasks(ctx, time.Now().UTC()); err != nil { _ = db.Close() _ = logger.Close() @@ -121,7 +126,7 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string certificates := certificate.NewService(cfg, repo, secretStore, dnsprovider.NewLegoFactory(), accounts) aliyunAuthChecker := dnsprovider.NewAliyunAuthorizationChecker() certificates.SetDNSAuthorizationChecker(aliyunAuthChecker) - deploymentService := deployment.NewService(deployment.SSHFactory{Secrets: secretStore}) + notifier := notification.New(cfg.Notification, secretStore) remoteValidator := certificate.RemoteValidator{} workflow := &CertificateWorkflow{ diff --git a/backend/internal/app/certificate_tasks.go b/backend/internal/app/certificate_tasks.go index d83c2d9..8e7df6e 100644 --- a/backend/internal/app/certificate_tasks.go +++ b/backend/internal/app/certificate_tasks.go @@ -147,6 +147,9 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s return result, err } issueCtx, cancel := context.WithTimeout(taskCtx, 30*time.Minute) + issueCtx = deployment.WithTransactionContext(issueCtx, deployment.TransactionContext{ + TaskID: task.ID, AttemptID: attempt.ID, AttemptNumber: attempt.AttemptNumber, + }) result, resultErr = execute(issueCtx, id) cancel() if recordErr := w.tasks.RecordTaskDeploymentResults(taskCtx, task.ID, attempt.ID, number, taskDeploymentResults(result.Deployment)); recordErr != nil { diff --git a/backend/internal/deployment/service.go b/backend/internal/deployment/service.go index 7685b0f..2690fd3 100644 --- a/backend/internal/deployment/service.go +++ b/backend/internal/deployment/service.go @@ -52,6 +52,7 @@ type Report struct { type Service struct { factory ClientFactory + journal TransactionJournal now func() time.Time } @@ -59,6 +60,10 @@ func NewService(factory ClientFactory) *Service { return &Service{factory: factory, now: time.Now} } +func NewServiceWithJournal(factory ClientFactory, journal TransactionJournal) *Service { + return &Service{factory: factory, journal: journal, now: time.Now} +} + func (s *Service) Deploy(ctx context.Context, targets []Target, artifact Artifact) Report { return s.deploy(ctx, targets, artifact, nil) } @@ -190,13 +195,18 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac result.Message += fmt.Sprintf("；释放部署锁失败，需人工检查：%v", err) return } + released := true for index := len(heldLocks) - 1; index >= 0; index-- { if err := client.Unlock(releaseCtx, heldLocks[index]); err != nil { + released = false result.Status = "failed" result.FailureCode = FailureLockRelease result.Message += fmt.Sprintf("；释放部署锁 %s 失败，需人工检查：%v", heldLocks[index], err) } } + if released { + heldLocks = nil + } } defer func() { if !handedOff { @@ -213,6 +223,32 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac } 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 + } + } + persistTransaction := func(phase string) error { + if !transactionEnabled { + return nil + } + journalRecord.Phase = phase + journalRecord.UpdatedAt = s.now().UTC() + return s.journal.UpdateDeploymentTransaction(ctx, *journalRecord) + } + finishTransaction := func(status, message string) { + if !transactionEnabled { + return + } + if err := s.journal.FinishDeploymentTransaction(context.WithoutCancel(ctx), journalRecord.ID, status, message); err != nil { + result.Status = "failed" + result.FailureCode = FailureRecoveryUnknown + result.Message += fmt.Sprintf("；保存部署事务最终状态失败：%v", err) + } + } backups := make([]backupFile, 0, len(paths)) temporary := make([]string, 0, len(paths)) installed := make([]string, 0, len(paths)) @@ -268,6 +304,12 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac message += "；服务运行状态未确认，请检查服务并在确认旧配置后重新加载" } result.Message = message + releaseLocks() + if retainLocks || result.FailureCode == FailureLockRelease { + finishTransaction(TransactionStatusRecoveryRequired, result.Message) + } else { + finishTransaction(TransactionStatusRolledBack, result.Message) + } return result } result.FailureCode = FailureUpload @@ -275,6 +317,9 @@ 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) { @@ -284,6 +329,9 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return failed(fmt.Sprintf("上传远程证书失败：%v", err)) } } + if err := persistTransaction(TransactionPhaseUploaded); err != nil { + return failed(fmt.Sprintf("保存上传阶段失败：%v", err)) + } // 所有文件上传完成后才触碰当前版本，避免上传失败造成不必要的证书替换。 result.FailureCode = FailureReplace for _, file := range paths { @@ -291,23 +339,44 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac if err != nil { return failed(fmt.Sprintf("检查远程文件失败：%v", err)) } - if exists { - backupPath := file.path + ".scert-backup-" + backupID - if exists, err := client.Exists(ctx, backupPath); err != nil || exists { - return failed(fmt.Sprintf("备份目标已存在或无法检查，已停止替换：%s：%v", backupPath, err)) + backupPath := file.path + ".scert-backup-" + backupID + if backupExists, checkErr := client.Exists(ctx, backupPath); checkErr != nil || backupExists { + return failed(fmt.Sprintf("备份目标已存在或无法检查，已停止替换：%s：%v", backupPath, checkErr)) + } + if transactionEnabled { + journalRecord.Backups = append(journalRecord.Backups, TransactionBackup{Original: file.path, Saved: backupPath, OriginalExists: exists}) + if err := persistTransaction(TransactionPhaseReplacing); err != nil { + return failed(fmt.Sprintf("保存替换阶段失败：%v", err)) } + } + if exists { backups = append(backups, backupFile{original: file.path, saved: backupPath}) if err := client.Rename(ctx, file.path, backupPath); err != nil { return failed(fmt.Sprintf("备份远程证书失败：%v", err)) } + if err := persistTransaction(TransactionPhaseReplacing); err != nil { + return failed(fmt.Sprintf("保存备份阶段失败：%v", err)) + } } temporaryPath := file.path + ".scert-tmp-" + backupID installed = append(installed, file.path) + if transactionEnabled { + journalRecord.InstalledPaths = append(journalRecord.InstalledPaths, file.path) + if err := persistTransaction(TransactionPhaseReplacing); err != nil { + return failed(fmt.Sprintf("保存安装阶段失败：%v", err)) + } + } if err := client.Rename(ctx, temporaryPath, file.path); err != nil { return failed(fmt.Sprintf("替换远程证书失败：%v", err)) } + if err := persistTransaction(TransactionPhaseInstalled); err != nil { + return failed(fmt.Sprintf("保存已安装阶段失败：%v", err)) + } } result.FailureCode = FailureValidation + if err := persistTransaction(TransactionPhaseInstalled); err != nil { + return failed(fmt.Sprintf("保存校验前阶段失败：%v", err)) + } validationResult, err := client.Run(ctx, server.ValidationCommand) if err != nil || validationResult.ExitCode != 0 { if err == nil { @@ -316,6 +385,12 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return failed(fmt.Sprintf("配置校验失败：%v", err)) } reloadAttempted = true + if transactionEnabled { + journalRecord.ReloadAttempted = true + if err := persistTransaction(TransactionPhaseReloadAttempted); err != nil { + return failed(fmt.Sprintf("保存重载前阶段失败：%v", err)) + } + } commandResult, err := client.Run(ctx, server.ReloadCommand) if err != nil || commandResult.ExitCode != 0 { if err == nil { @@ -354,6 +429,12 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return result } transaction.commit = func(commitContext context.Context) Result { + if err := persistTransaction("committing"); err != nil { + result.Status = "failed" + result.FailureCode = FailureCleanup + result.Message = fmt.Sprintf("保存提交阶段失败：%v", err) + return result + } if err := cleanupBackups(commitContext, client, backups, paths); err != nil { result.Status = "failed" result.FailureCode = FailureCleanup @@ -370,8 +451,18 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac _ = client.Close() closed = true } + status := TransactionStatusCommitted + if retain || result.Status != "success" { + status = TransactionStatusRecoveryRequired + } else if result.Message != "" && result.Status == "failed" { + status = TransactionStatusRolledBack + } + finishTransaction(status, result.Message) return result } + if transactionEnabled { + journalRecord.Phase = TransactionPhaseReloadAttempted + } handedOff = true *pending = append(*pending, transaction) return result @@ -379,16 +470,165 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac if err := cleanupBackups(ctx, client, backups, paths); err != nil { result.Status = "failed" result.FailureCode = FailureCleanup - result.Message = fmt.Sprintf("\u8bc1\u4e66\u5df2\u90e8\u7f72\u5e76 reload \u6210\u529f\uff0c\u4f46\u65e7\u8bc1\u4e66\u5907\u4efd\u6e05\u7406\u5931\u8d25\uff1a%v", err) + result.Message = fmt.Sprintf("证书已部署并 reload 成功，但旧证书备份清理失败：%v", err) + releaseLocks() + finishTransaction(TransactionStatusRecoveryRequired, result.Message) return result } result.Status = "success" result.FailureCode = "" result.Message = "证书部署、配置校验并 reload 成功" result.BackupID = backupID + releaseLocks() + if result.FailureCode == FailureLockRelease { + finishTransaction(TransactionStatusRecoveryRequired, result.Message) + } else { + finishTransaction(TransactionStatusCommitted, result.Message) + } return result } +func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths []remoteFile, lockPaths []string, id string, index int) (*TransactionRecord, bool) { + transactionContext, ok := TransactionContextFromContext(ctx) + if !ok || s == nil || s.journal == nil { + return nil, false + } + pathValues := make([]string, 0, len(paths)) + temporaryPaths := make([]string, 0, len(paths)) + for _, file := range paths { + pathValues = append(pathValues, string(file.path)) + temporaryPaths = append(temporaryPaths, string(file.path)+".scert-tmp-"+id) + } + now := time.Now().UTC() + if s.now != nil { + now = s.now().UTC() + } + 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, + }, true +} + +// RecoverPending 在服务启动时只处理能够证明属于本次事务的现场；无法证明时保留现场并要求人工核查。 +func (s *Service) RecoverPending(ctx context.Context, resolve func(context.Context, string) (Server, error)) error { + if s == nil || s.journal == nil { + return nil + } + if resolve == nil { + return errors.New("部署事务恢复缺少服务器解析器") + } + records, err := s.journal.ListPendingDeploymentTransactions(ctx) + if err != nil { + return fmt.Errorf("读取待恢复部署事务失败：%w", err) + } + var recoveryErrors []error + for _, record := range records { + 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 { + 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 { + server, err := resolve(ctx, record.ServerID) + if err != nil { + return fmt.Errorf("读取服务器配置失败：%w", err) + } + server = server.Normalize() + if err := server.Validate(); err != nil { + return fmt.Errorf("服务器配置无效：%w", err) + } + recoveryContext, cancel := context.WithTimeout(context.WithoutCancel(ctx), deploymentRollbackTimeout) + defer cancel() + client, err := s.factory.Connect(recoveryContext, server) + if err != nil { + 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) + } + } + 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) + } + } + } + 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.ReloadAttempted { + for _, command := range [][]string{server.ValidationCommand, server.ReloadCommand} { + result, runErr := client.Run(recoveryContext, command) + if runErr != nil || result.ExitCode != 0 { + if runErr == nil { + runErr = fmt.Errorf("退出码 %d", result.ExitCode) + } + return fmt.Errorf("恢复服务失败：%w", runErr) + } + } + } + for index := len(record.LockPaths) - 1; index >= 0; index-- { + if err := client.Unlock(recoveryContext, record.LockPaths[index]); err != nil { + return fmt.Errorf("释放部署锁失败：%w", err) + } + } + return nil +} + func cleanupBackups(ctx context.Context, client Client, backups []backupFile, paths []remoteFile) error { seen := make(map[string]struct{}, len(backups)) for _, item := range backups { diff --git a/backend/internal/deployment/transaction.go b/backend/internal/deployment/transaction.go index 2f9cd24..708b2f2 100644 --- a/backend/internal/deployment/transaction.go +++ b/backend/internal/deployment/transaction.go @@ -8,8 +8,9 @@ import ( // TransactionBackup 描述一次部署事务保存的旧文件。 type TransactionBackup struct { - Original string `json:"original"` - Saved string `json:"saved"` + Original string `json:"original"` + Saved string `json:"saved"` + OriginalExists bool `json:"original_exists"` } // TransactionRecord 保存远程部署事务的可恢复现场，不包含证书内容或凭据。 @@ -41,6 +42,21 @@ type TransactionJournal interface { ListPendingDeploymentTransactions(context.Context) ([]TransactionRecord, error) } +const ( + TransactionStatusPending = "pending" + TransactionStatusCommitted = "committed" + TransactionStatusRolledBack = "rolled_back" + TransactionStatusRecoveryRequired = "recovery_required" + + TransactionPhaseLocked = "locked" + TransactionPhaseUploaded = "uploaded" + TransactionPhaseReplacing = "replacing" + TransactionPhaseInstalled = "installed" + TransactionPhaseReloadAttempted = "reload_attempted" + TransactionPhaseCommitted = "committed" + TransactionPhaseRolledBack = "rolled_back" +) + type transactionContextKey struct{} // TransactionContext 保存任务与尝试标识，供部署层建立可追溯事务记录。 diff --git a/backend/internal/deployment/transaction_recovery_test.go b/backend/internal/deployment/transaction_recovery_test.go new file mode 100644 index 0000000..75779e3 --- /dev/null +++ b/backend/internal/deployment/transaction_recovery_test.go @@ -0,0 +1,105 @@ +package deployment + +import ( + "context" + "testing" +) + +type memoryTransactionJournal struct { + pending []TransactionRecord + finished []struct { + id string + status string + message string + } + updated []TransactionRecord +} + +func (j *memoryTransactionJournal) CreateDeploymentTransaction(_ context.Context, item TransactionRecord) error { + j.pending = append(j.pending, item) + return nil +} + +func (j *memoryTransactionJournal) UpdateDeploymentTransaction(_ context.Context, item TransactionRecord) error { + j.updated = append(j.updated, item) + return nil +} + +func (j *memoryTransactionJournal) FinishDeploymentTransaction(_ context.Context, id, status, message string) error { + j.finished = append(j.finished, struct { + id string + status string + message string + }{id: id, status: status, message: message}) + return nil +} + +func (j *memoryTransactionJournal) ListPendingDeploymentTransactions(context.Context) ([]TransactionRecord, error) { + return append([]TransactionRecord(nil), j.pending...), nil +} + +func TestRecoverPendingRollsBackOwnedDeployment(t *testing.T) { + server := testServer("recovery") + 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)) + 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}) + temporary = append(temporary, temporaryPath) + client.files[lockPath] = memoryFile{} + 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} + } + 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, + }}} + service := NewServiceWithJournal(memoryFactory{clients: map[string]*memoryClient{server.ID: client}}, journal) + if err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { return server, nil }); err != nil { + t.Fatalf("事务恢复失败：%v", err) + } + for _, path := range paths { + file, ok := client.files[path] + if !ok || string(file.data) != "old" { + t.Fatalf("目标文件未恢复：%s，文件：%+v", path, client.files) + } + if _, ok := client.files[path+".scert-lock"]; ok { + t.Fatalf("恢复成功后不应残留锁：%s", path) + } + if _, ok := client.files[path+".scert-tmp-"+id]; ok { + t.Fatalf("恢复成功后不应残留临时文件：%s", path) + } + } + if len(client.commands) != 2 || len(journal.finished) != 1 || journal.finished[0].status != TransactionStatusRolledBack { + t.Fatalf("恢复收尾记录不正确：命令=%d，完成记录=%+v", len(client.commands), journal.finished) + } +} + +func TestRecoverPendingKeepsUnprovableState(t *testing.T) { + server := testServer("unprovable") + client := newMemoryClient() + journal := &memoryTransactionJournal{pending: []TransactionRecord{{ + ID: "unprovable-transaction", TaskID: "task", AttemptID: "attempt", AttemptNumber: 1, + ResultIndex: 0, ServerID: server.ID, ServerName: server.Name, + Paths: []string{server.CertificatePath, server.CertificateChainPath, server.PrivateKeyPath}, + LockPaths: []string{server.PrivateKeyPath + ".scert-lock"}, Status: TransactionStatusPending, + Phase: TransactionPhaseInstalled, + }}} + service := NewServiceWithJournal(memoryFactory{clients: map[string]*memoryClient{server.ID: client}}, journal) + if err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { return server, nil }); err == nil { + t.Fatal("无法证明锁归属时必须返回人工核查错误") + } + if len(journal.updated) != 1 || journal.updated[0].Status != TransactionStatusRecoveryRequired { + t.Fatalf("未记录人工核查状态：%+v", journal.updated) + } +} diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index 249fc87..69db7e3 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -138,6 +138,9 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string) { executionCtx, cancel := context.WithTimeout(c.Request.Context(), certificateTaskTimeout) + executionCtx = deployment.WithTransactionContext(executionCtx, deployment.TransactionContext{ + TaskID: task.ID, AttemptID: attempt.ID, AttemptNumber: attempt.AttemptNumber, + }) s.registerTaskCancel(task.ID, cancel) defer func() { cancel()