《后端架构设计·分布式与数据一致性》7.2 Saga：补偿、语义锁与不可补偿步骤

// file: internal/platform/txn/saga/saga.go
// tier: L1
// verify: go test ./internal/platform/txn/saga/ -count=1
package saga

import (
	"context"
	"errors"
	"fmt"
	"sync"
)

// ErrLocked is returned when another transaction holds the aggregate lock.
var ErrLocked = errors.New("saga: aggregate locked")

// Action is one forward or compensation step.
type Action func(context.Context) error

// Step is one saga step. A nil Compensate marks an irreversible step.
type Step struct {
	Name       string
	Do         Action
	Compensate Action
	Retryable  bool
}

// State is the terminal state of a saga run.
type State string

const (
	StateCompleted   State = "completed"
	StateCompensated State = "compensated"
	StateStuck       State = "stuck"
)

// Result reports what happened to each step.
type Result struct {
	State       State
	Completed   []string
	Compensated []string
	FailedStep  string
	Err         error
}

// Locks is an in-memory semantic lock table keyed by aggregate ID.
type Locks struct {
	mu   sync.Mutex
	held map[string]string
}

// NewLocks creates an empty lock table.
func NewLocks() *Locks { return &Locks{held: make(map[string]string)} }

// Acquire marks the aggregate as being modified by txID.
func (l *Locks) Acquire(aggregateID, txID string) error {
	l.mu.Lock()
	defer l.mu.Unlock()
	if owner, ok := l.held[aggregateID]; ok && owner != txID {
		return fmt.Errorf("saga: acquire %s: %w", aggregateID, ErrLocked)
	}
	l.held[aggregateID] = txID
	return nil
}

// Release removes the lock only if txID still owns it.
func (l *Locks) Release(aggregateID, txID string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.held[aggregateID] == txID {
		delete(l.held, aggregateID)
	}
}

// Held reports the owner of an aggregate lock.
func (l *Locks) Held(aggregateID string) (string, bool) {
	l.mu.Lock()
	defer l.mu.Unlock()
	owner, ok := l.held[aggregateID]
	return owner, ok
}

// Validate enforces the structural rule: after an irreversible step no
// compensable step may follow, and the irreversible step must be retryable
// because there is no undo to fall back on.
func Validate(steps []Step) error {
	seen := make(map[string]bool, len(steps))
	irreversible := false
	for i, s := range steps {
		if s.Name == "" {
			return fmt.Errorf("saga: step %d has empty name", i)
		}
		if seen[s.Name] {
			return fmt.Errorf("saga: duplicate step %q", s.Name)
		}
		seen[s.Name] = true
		if s.Do == nil {
			return fmt.Errorf("saga: step %q has no action", s.Name)
		}
		if s.Compensate == nil {
			irreversible = true
			if !s.Retryable {
				return fmt.Errorf("saga: irreversible step %q must be retryable", s.Name)
			}
		} else if irreversible {
			return fmt.Errorf("saga: compensable step %q follows irreversible step", s.Name)
		}
	}
	return nil
}

// Orchestrator runs steps sequentially, retries retryable steps, and
// compensates completed steps in reverse order on failure.
type Orchestrator struct {
	Steps    []Step
	Locks    *Locks
	Attempts int
}

// Run executes a saga for aggregateID using txID as the lock owner.
func (o *Orchestrator) Run(ctx context.Context, aggregateID, txID string) Result {
	if o.Locks != nil {
		if err := o.Locks.Acquire(aggregateID, txID); err != nil {
			return Result{State: StateStuck, Err: err}
		}
		defer o.Locks.Release(aggregateID, txID)
	}
	res := Result{State: StateCompleted}
	for _, s := range o.Steps {
		if err := o.runStep(ctx, s); err != nil {
			res.FailedStep = s.Name
			res.Err = err
			res.Compensated = o.compensate(ctx, res.Completed)
			res.State = StateStuck
			if len(res.Compensated) == len(res.Completed) {
				res.State = StateCompensated
			}
			return res
		}
		res.Completed = append(res.Completed, s.Name)
	}
	return res
}

func (o *Orchestrator) runStep(ctx context.Context, s Step) error {
	attempts := 1
	if s.Retryable && o.Attempts > 1 {
		attempts = o.Attempts
	}
	var err error
	for i := 0; i < attempts; i++ {
		if err = s.Do(ctx); err == nil {
			return nil
		}
	}
	return err
}

func (o *Orchestrator) compensate(ctx context.Context, completed []string) []string {
	done := make([]string, 0, len(completed))
	for i := len(completed) - 1; i >= 0; i-- {
		name := completed[i]
		for _, s := range o.Steps {
			if s.Name != name || s.Compensate == nil {
				continue
			}
			if err := s.Compensate(ctx); err == nil {
				done = append(done, name)
			}
			break
		}
	}
	return done
}

$ go test ./internal/platform/txn/saga/ -count=1 -v
=== RUN   TestSagaHappyPathReleasesLock
--- PASS: TestSagaHappyPathReleasesLock (0.00s)
=== RUN   TestFailureCompensatesInReverseOrder
--- PASS: TestFailureCompensatesInReverseOrder (0.00s)
=== RUN   TestFailedCompensationLeavesSagaStuck
--- PASS: TestFailedCompensationLeavesSagaStuck (0.00s)
=== RUN   TestRetryableStepIsRetried
--- PASS: TestRetryableStepIsRetried (0.00s)
=== RUN   TestSemanticLockRejectsConcurrentTransaction
--- PASS: TestSemanticLockRejectsConcurrentTransaction (0.00s)
=== RUN   TestValidateRejectsUnsafeStepOrder
=== RUN   TestValidateRejectsUnsafeStepOrder/compensable_then_irreversible
=== RUN   TestValidateRejectsUnsafeStepOrder/compensable_after_irreversible
=== RUN   TestValidateRejectsUnsafeStepOrder/irreversible_not_retryable
=== RUN   TestValidateRejectsUnsafeStepOrder/duplicate_names
=== RUN   TestValidateRejectsUnsafeStepOrder/empty_name
--- PASS: TestValidateRejectsUnsafeStepOrder (0.00s)
    --- PASS: TestValidateRejectsUnsafeStepOrder/compensable_then_irreversible (0.00s)
    --- PASS: TestValidateRejectsUnsafeStepOrder/compensable_after_irreversible (0.00s)
    --- PASS: TestValidateRejectsUnsafeStepOrder/irreversible_not_retryable (0.00s)
    --- PASS: TestValidateRejectsUnsafeStepOrder/duplicate_names (0.00s)
    --- PASS: TestValidateRejectsUnsafeStepOrder/empty_name (0.00s)
PASS
ok  	archlab/internal/platform/txn/saga	1.491s