修复部署事务提交上下文

提交阶段使用独立的有界上下文,补充请求取消竞态回归测试。
This commit is contained in:
xuans 2026-10-01 03:57:38 +08:00
parent ad0b0c3f45
commit a14d040e3f
3 changed files with 71 additions and 11 deletions

View File

@ -231,13 +231,13 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
return result
}
}
persistTransaction := func(phase string) error {
persistTransaction := func(persistContext context.Context, phase string) error {
if !transactionEnabled {
return nil
}
journalRecord.Phase = phase
journalRecord.UpdatedAt = s.now().UTC()
return s.journal.UpdateDeploymentTransaction(ctx, *journalRecord)
return s.journal.UpdateDeploymentTransaction(persistContext, *journalRecord)
}
finishTransaction := func(status, message string) {
if !transactionEnabled {
@ -329,7 +329,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
return failed(fmt.Sprintf("上传远程证书失败:%v", err))
}
}
if err := persistTransaction(TransactionPhaseUploaded); err != nil {
if err := persistTransaction(ctx, TransactionPhaseUploaded); err != nil {
return failed(fmt.Sprintf("保存上传阶段失败:%v", err))
}
// 所有文件上传完成后才触碰当前版本,避免上传失败造成不必要的证书替换。
@ -345,7 +345,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
}
if transactionEnabled {
journalRecord.Backups = append(journalRecord.Backups, TransactionBackup{Original: file.path, Saved: backupPath, OriginalExists: exists})
if err := persistTransaction(TransactionPhaseReplacing); err != nil {
if err := persistTransaction(ctx, TransactionPhaseReplacing); err != nil {
return failed(fmt.Sprintf("保存替换阶段失败:%v", err))
}
}
@ -354,7 +354,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
if err := client.Rename(ctx, file.path, backupPath); err != nil {
return failed(fmt.Sprintf("备份远程证书失败:%v", err))
}
if err := persistTransaction(TransactionPhaseReplacing); err != nil {
if err := persistTransaction(ctx, TransactionPhaseReplacing); err != nil {
return failed(fmt.Sprintf("保存备份阶段失败:%v", err))
}
}
@ -362,19 +362,19 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
installed = append(installed, file.path)
if transactionEnabled {
journalRecord.InstalledPaths = append(journalRecord.InstalledPaths, file.path)
if err := persistTransaction(TransactionPhaseReplacing); err != nil {
if err := persistTransaction(ctx, 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 {
if err := persistTransaction(ctx, TransactionPhaseInstalled); err != nil {
return failed(fmt.Sprintf("保存已安装阶段失败:%v", err))
}
}
result.FailureCode = FailureValidation
if err := persistTransaction(TransactionPhaseInstalled); err != nil {
if err := persistTransaction(ctx, TransactionPhaseInstalled); err != nil {
return failed(fmt.Sprintf("保存校验前阶段失败:%v", err))
}
validationResult, err := client.Run(ctx, server.ValidationCommand)
@ -387,7 +387,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac
reloadAttempted = true
if transactionEnabled {
journalRecord.ReloadAttempted = true
if err := persistTransaction(TransactionPhaseReloadAttempted); err != nil {
if err := persistTransaction(ctx, TransactionPhaseReloadAttempted); err != nil {
return failed(fmt.Sprintf("保存重载前阶段失败:%v", err))
}
}
@ -429,7 +429,7 @@ 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 {
if err := persistTransaction(commitContext, "committing"); err != nil {
result.Status = "failed"
result.FailureCode = FailureCleanup
result.Message = fmt.Sprintf("保存提交阶段失败:%v", err)

View File

@ -65,8 +65,10 @@ func (s *Service) DeployVerified(ctx context.Context, targets []Target, artifact
report.Status = "failed"
return report, deployErr
}
commitContext, cancel := context.WithTimeout(context.WithoutCancel(ctx), deploymentRollbackTimeout)
defer cancel()
for _, transaction := range pending {
result := transaction.commit(ctx)
result := transaction.commit(commitContext)
report.Results[transaction.index] = transaction.finish(result.Status != "success")
}
summarizeReport(&report)

View File

@ -6,9 +6,67 @@ import (
"errors"
"reflect"
"strings"
"sync/atomic"
"testing"
)
type lateCancellationContext struct {
context.Context
errCalls atomic.Int32
}
func (c *lateCancellationContext) Done() <-chan struct{} {
return nil
}
func (c *lateCancellationContext) Err() error {
if c.errCalls.Add(1) > 2 {
return context.Canceled
}
return nil
}
type contextCheckingTransactionJournal struct {
memoryTransactionJournal
commitContextOK bool
}
func (j *contextCheckingTransactionJournal) UpdateDeploymentTransaction(ctx context.Context, item TransactionRecord) error {
if item.Phase == "committing" {
j.commitContextOK = ctx.Err() == nil
}
return j.memoryTransactionJournal.UpdateDeploymentTransaction(ctx, item)
}
func TestVerifiedDeploymentCommitUsesIndependentContextAfterCancellationRace(t *testing.T) {
client, server := newMemoryClient(), testServer("commit-context")
seedVerifiedFiles(client, server)
journal := &contextCheckingTransactionJournal{}
parent := &lateCancellationContext{Context: t.Context()}
ctx := WithTransactionContext(parent, TransactionContext{
TaskID: "task", AttemptID: "attempt", AttemptNumber: 1,
})
artifact := testArtifact(t)
service := NewServiceWithJournal(fixedClientFactory{client}, journal)
report, err := service.DeployVerified(ctx, []Target{{Server: server}}, artifact, Verification{
Verify: func(context.Context) error { return nil },
})
if err != nil || report.Status != "success" || !journal.commitContextOK {
t.Fatalf("unexpected deployment result: %+v, %v", report, err)
}
if parent.Err() != context.Canceled {
t.Fatal("parent context was not cancelled after verification")
}
if len(journal.finished) != 1 || journal.finished[0].status != TransactionStatusCommitted {
t.Fatalf("unexpected journal completion: %+v", journal.finished)
}
assertVerifiedLocks(t, client, server, false)
for path := range client.files {
if strings.Contains(path, ".scert-backup-") || strings.Contains(path, ".scert-tmp-") {
t.Fatalf("transaction artifact remains: %s", path)
}
}
}
func seedVerifiedFiles(client *memoryClient, server Server) map[string]memoryFile {
files := map[string]memoryFile{
server.CertificatePath: {data: []byte("原证书"), mode: 0o644},