完善回滚票据并发与失败重试
Some checks are pending
Security Checks / security (push) Waiting to run

This commit is contained in:
smail 2026-08-30 22:36:20 +08:00
parent d534f0187d
commit acf03222e9
5 changed files with 140 additions and 21 deletions

View File

@ -17,6 +17,7 @@ import (
var (
ErrGenerationConflict = errors.New("configuration generation conflict")
ErrRollbackInProgress = errors.New("configuration rollback is already in progress")
ErrNoBackup = errors.New("configuration backup not found")
ErrManagerStorage = errors.New("configuration manager storage unavailable")
)

View File

@ -96,6 +96,7 @@ type rollbackTicket struct {
SessionID string
TargetGeneration uint64
ExpiresAt time.Time
InFlight bool
}
type LegacyPluginSecret struct {
@ -524,31 +525,60 @@ func (m *ConfigManager) RollbackWithTicket(ctx context.Context, token string, ac
m.ticketMu.Unlock()
return nil, ErrAuthorizationRequired
}
delete(m.rollbackTickets, token)
if ticket.InFlight {
m.ticketMu.Unlock()
return nil, ErrRollbackInProgress
}
ticket.InFlight = true
m.rollbackTickets[token] = ticket
m.ticketMu.Unlock()
return m.rollback(ctx, ticket.TargetGeneration, actor, StepUpConfirmation{Confirmed: true, Method: "ticket"})
result, applied, err := m.rollbackWithState(ctx, ticket.TargetGeneration, actor, StepUpConfirmation{Confirmed: true, Method: "ticket"})
m.ticketMu.Lock()
if err == nil || applied || !rollbackTicketRetryable(err) {
delete(m.rollbackTickets, token)
} else if current, exists := m.rollbackTickets[token]; exists {
current.InFlight = false
m.rollbackTickets[token] = current
}
m.ticketMu.Unlock()
return result, err
}
func rollbackTicketRetryable(err error) bool {
if err == nil {
return false
}
return errors.Is(err, context.Canceled) ||
errors.Is(err, context.DeadlineExceeded) ||
errors.Is(err, ErrRuntimeSnapshotUnavailable)
}
func (m *ConfigManager) rollback(ctx context.Context, generation uint64, actor AuditActor, confirmation StepUpConfirmation) (*ApplyResult, error) {
result, _, err := m.rollbackWithState(ctx, generation, actor, confirmation)
return result, err
}
func (m *ConfigManager) rollbackWithState(ctx context.Context, generation uint64, actor AuditActor, confirmation StepUpConfirmation) (*ApplyResult, bool, error) {
if err := contextError(ctx); err != nil {
return nil, err
return nil, false, err
}
if m == nil {
return nil, ErrManagerStorage
return nil, false, ErrManagerStorage
}
if generation == 0 {
return nil, ErrGenerationConflict
return nil, false, ErrGenerationConflict
}
if err := m.authorize(ctx, actor, ConfigOperationRollback, SectionRoot); err != nil {
return nil, err
return nil, false, err
}
if confirmation.Method != "ticket" {
if err := m.confirm(ctx, actor, confirmation, ConfigOperationRollback, SectionRoot); err != nil {
return nil, err
return nil, false, err
}
}
if err := contextError(ctx); err != nil {
return nil, err
return nil, false, err
}
m.writeMu.Lock()
defer m.writeMu.Unlock()
@ -557,40 +587,40 @@ func (m *ConfigManager) rollback(ctx context.Context, generation uint64, actor A
pending := m.pending != nil
m.mu.RUnlock()
if pending || generation >= current.Generation {
return nil, ErrGenerationConflict
return nil, false, ErrGenerationConflict
}
candidate, err := m.transactions.LoadBackup(generation)
if err != nil {
return nil, err
return nil, false, err
}
if err := contextError(ctx); err != nil {
return nil, err
return nil, false, err
}
if err := ValidateRuntimeSecrets(m.vault, candidate); err != nil {
return nil, err
return nil, false, err
}
setCandidateGeneration(&candidate, current.Generation+1)
candidate.Root.UpdatedAt = time.Now().UTC()
if err := ValidateRuntimeConfig(candidate, m.options); err != nil {
return nil, err
return nil, false, err
}
if err := m.snapshots.Validate(candidate); err != nil {
return nil, err
return nil, false, err
}
if err := contextError(ctx); err != nil {
return nil, err
return nil, false, err
}
if _, _, err := m.transactions.CreateBackupIfDue(current, time.Now().UTC()); err != nil {
return nil, err
return nil, false, err
}
if err := contextError(ctx); err != nil {
return nil, err
return nil, false, err
}
if err := m.transactions.Commit(current, candidate); err != nil {
return nil, err
return nil, false, err
}
if err := m.snapshots.Publish(candidate); err != nil {
return nil, err
return nil, false, err
}
now := time.Now().UTC()
m.mu.Lock()
@ -615,9 +645,9 @@ func (m *ConfigManager) rollback(ctx context.Context, generation uint64, actor A
Result: "success",
CreatedAt: now,
}); err != nil {
return nil, err
return nil, true, err
}
return &ApplyResult{Snapshot: cloneRuntimeConfig(candidate), Generation: candidate.Generation, AppliedAt: now}, nil
return &ApplyResult{Snapshot: cloneRuntimeConfig(candidate), Generation: candidate.Generation, AppliedAt: now}, true, nil
}
func (m *ConfigManager) authorize(ctx context.Context, actor AuditActor, operation ConfigOperation, section Section) error {

View File

@ -7,6 +7,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
@ -1476,3 +1477,75 @@ func TestConfigManagerSnapshotResultsAreIndependentCopies(t *testing.T) {
t.Fatal("apply result exposed manager-owned active data")
}
}
type flakyRuntimeSnapshotStore struct {
mu sync.Mutex
current RuntimeConfig
failures int
}
func (s *flakyRuntimeSnapshotStore) Validate(RuntimeConfig) error {
return nil
}
func (s *flakyRuntimeSnapshotStore) Publish(candidate RuntimeConfig) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.failures > 0 {
s.failures--
return ErrRuntimeSnapshotUnavailable
}
s.current = cloneRuntimeConfig(candidate)
return nil
}
func (s *flakyRuntimeSnapshotStore) Snapshot() (RuntimeConfig, error) {
s.mu.Lock()
defer s.mu.Unlock()
return cloneRuntimeConfig(s.current), nil
}
func TestConfigManagerRollbackTicketCanRetryAfterSnapshotFailure(t *testing.T) {
root := t.TempDir()
writeRuntimeFixture(t, root, validRuntimeConfig())
manager, err := NewConfigManager(root, ConfigLoaderOptions{Validation: RuntimeValidationOptions{HasSuperAdmin: true, Production: true}})
if err != nil {
t.Fatal(err)
}
updatedEmail := manager.Snapshot().Email
updatedEmail.Policy.MaxMessageBytes = 8 << 20
payload, err := json.Marshal(updatedEmail)
if err != nil {
t.Fatal(err)
}
updated, err := manager.Update(context.Background(), UpdateRequest{Section: SectionEmail, ExpectedGeneration: 1, Payload: payload, Actor: authorizedConfigActor()})
if err != nil {
t.Fatal(err)
}
if _, err := manager.Apply(context.Background(), updated.Generation); err != nil {
t.Fatal(err)
}
manager.snapshots = &flakyRuntimeSnapshotStore{current: manager.Snapshot(), failures: 1}
actor := authorizedConfigActor()
ticket, err := manager.IssueRollbackTicket(context.Background(), 1, actor, StepUpConfirmation{Confirmed: true, Method: "webauthn"})
if err != nil {
t.Fatal(err)
}
if _, err := manager.RollbackWithTicket(context.Background(), ticket, actor); !errors.Is(err, ErrRuntimeSnapshotUnavailable) {
t.Fatalf("first rollback error = %v, want snapshot unavailable", err)
}
stored, exists := manager.rollbackTickets[ticket]
if !exists || stored.InFlight {
t.Fatalf("rollback ticket state after retryable failure = %+v, exists=%v", stored, exists)
}
result, err := manager.RollbackWithTicket(context.Background(), ticket, actor)
if err != nil {
t.Fatal(err)
}
if result.Generation != 3 || result.Snapshot.Email.Policy.MaxMessageBytes == 8<<20 {
t.Fatalf("retry rollback result = %+v", result)
}
if _, err := manager.RollbackWithTicket(context.Background(), ticket, actor); !errors.Is(err, ErrAuthorizationRequired) {
t.Fatalf("replayed retry ticket error = %v, want authorization required", err)
}
}

View File

@ -53,6 +53,10 @@ func configErrorDetails(err error) (string, int, string, map[string]string) {
code = "CONFIG_MANAGER_UNAVAILABLE"
status = http.StatusServiceUnavailable
message = "配置服务暂时不可用,请稍后重试"
case errors.Is(err, config.ErrRollbackInProgress):
code = "CONFIG_ROLLBACK_IN_PROGRESS"
status = http.StatusConflict
message = "配置回滚正在执行,请稍后查看结果"
case errors.Is(err, config.ErrGenerationConflict):
code = "CONFIG_GENERATION_CONFLICT"
status = http.StatusConflict

View File

@ -2,6 +2,7 @@ package controllers
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
@ -167,3 +168,13 @@ func TestConfigErrorDetailsMapsDatabaseMigrationFailures(t *testing.T) {
})
}
}
func TestConfigErrorDetailsMapsRollbackInProgress(t *testing.T) {
code, status, message, _ := configErrorDetails(config.ErrRollbackInProgress)
if code != "CONFIG_ROLLBACK_IN_PROGRESS" || status != http.StatusConflict {
t.Fatalf("rollback in progress mapping = %q/%d, want conflict", code, status)
}
if message == "" || strings.Contains(message, "rollback in progress") {
t.Fatalf("rollback in progress message is not a Chinese public message: %q", message)
}
}