《MMO 服务器开发》10.1 跨服务事务与补偿

// file: internal/saga/saga.go
// tier: L1
// verify: go test ./internal/saga/

package saga

// Status is the terminal state of one saga instance.
type Status int

const (
	StatusCommitted  Status = iota // every forward step succeeded
	StatusRolledBack               // a step failed; all completed steps compensated
	StatusManual                   // some compensation kept failing; needs a human
)

// Step is one forward action and the action that undoes it. Compensate must be
// idempotent: the orchestrator may call it more than once when retrying.
type Step struct {
	Name       string
	Do         func() error
	Compensate func() error
}

// Orchestrator drives an ordered list of steps and rolls back on failure.
type Orchestrator struct {
	// MaxCompRetry is the number of extra attempts after the first failed
	// compensation. MaxCompRetry=1 means at most two tries per step.
	MaxCompRetry int
}

// Result records what happened so callers (and on-call humans) can audit it.
type Result struct {
	Status      Status
	FailedStep  string   // the forward step that failed, "" if none
	Compensated []string // steps compensated, newest-first
	ManualSteps []string // steps whose compensation never succeeded
}

// Run executes steps in order; on the first failure it compensates the already
// completed steps in reverse order so the world returns to its initial state.
func (o *Orchestrator) Run(steps []Step) Result {
	done := make([]int, 0, len(steps))
	for i, s := range steps {
		if err := s.Do(); err != nil {
			return o.compensate(steps, done, s.Name)
		}
		done = append(done, i)
	}
	return Result{Status: StatusCommitted}
}

// compensate undoes done (indices into steps) newest-first. A step still failing
// after MaxCompRetry retries is flagged for a human, but the loop keeps going:
// one stuck step must not strand the others.
func (o *Orchestrator) compensate(steps []Step, done []int, failed string) Result {
	r := Result{Status: StatusRolledBack, FailedStep: failed}
	for j := len(done) - 1; j >= 0; j-- {
		st := steps[done[j]]
		ok := false
		for attempt := 0; attempt <= o.MaxCompRetry; attempt++ {
			if err := st.Compensate(); err == nil {
				ok = true
				break
			}
		}
		if ok {
			r.Compensated = append(r.Compensated, st.Name)
		} else {
			r.ManualSteps = append(r.ManualSteps, st.Name)
		}
	}
	if len(r.ManualSteps) > 0 {
		r.Status = StatusManual
	}
	return r
}

// file: internal/saga/services.go
// tier: L1
// verify: go test ./internal/saga/

package saga

import "fmt"

// Account is the asset view a cross-server saga mutates.
type Account struct {
	Points int64            // 跨服点券
	Items  map[uint32]int64 // ItemID -> count
}

// World stands in for the three services a cross-server signup touches: the
// point wallet, the bag, and the settlement ledger. Fail / FailComp inject
// forward / compensation failures and exist only to drive tests.
type World struct {
	Accounts map[uint64]*Account
	Ledger   []string // settled order ids, append-only
	Fail     map[string]bool
	FailComp map[string]bool
}

// NewWorld returns an empty world with injectable failure maps.
func NewWorld() *World {
	return &World{Accounts: map[uint64]*Account{}, Fail: map[string]bool{}, FailComp: map[string]bool{}}
}

func (w *World) account(pid uint64) *Account {
	a := w.Accounts[pid]
	if a == nil {
		a = &Account{Items: map[uint32]int64{}}
		w.Accounts[pid] = a
	}
	return a
}

// PointsService is the point-wallet service (may live on another shard).
type PointsService struct{ w *World }

// Deduct removes n points; it fails if the player cannot afford it.
func (p PointsService) Deduct(pid uint64, n int64) error {
	a := p.w.account(pid)
	if a.Points < n {
		return fmt.Errorf("points: player %d has %d, need %d", pid, a.Points, n)
	}
	a.Points -= n
	return nil
}

// Refund returns n points; idempotent because the saga may retry it.
func (p PointsService) Refund(pid uint64, n int64) error { p.w.account(pid).Points += n; return nil }

// RewardService grants items into the bag.
type RewardService struct{ w *World }

func (r RewardService) Grant(pid uint64, itemID uint32, n int64) error {
	r.w.account(pid).Items[itemID] += n
	return nil
}

// Revoke removes n of itemID; the inverse of Grant.
func (r RewardService) Revoke(pid uint64, itemID uint32, n int64) error {
	a := r.w.account(pid)
	a.Items[itemID] -= n
	if a.Items[itemID] <= 0 {
		delete(a.Items, itemID)
	}
	return nil
}

// LedgerService records the cross-server settlement line.
type LedgerService struct{ w *World }

func (l LedgerService) Append(orderID string) error {
	l.w.Ledger = append(l.w.Ledger, orderID)
	return nil
}

// RemoveLast drops the top line, undoing the matching Append.
func (l LedgerService) RemoveLast(orderID string) error {
	n := len(l.w.Ledger)
	if n == 0 || l.w.Ledger[n-1] != orderID {
		return fmt.Errorf("ledger: top is not %q", orderID)
	}
	l.w.Ledger = l.w.Ledger[:n-1]
	return nil
}

// CrossSignup builds the three-step saga: charge points, grant the reward, then
// record the settlement line. Each step carries its inverse.
func CrossSignup(w *World, pid uint64, cost int64, itemID uint32, reward int64, orderID string) []Step {
	points, rewardSvc, ledger := PointsService{w}, RewardService{w}, LedgerService{w}
	wrap := func(name string, fn func() error) func() error {
		return func() error {
			if w.Fail[name] {
				return fmt.Errorf("injected failure at %s", name)
			}
			return fn()
		}
	}
	wrapComp := func(name string, fn func() error) func() error {
		return func() error {
			if w.FailComp[name] {
				return fmt.Errorf("injected compensation failure at %s", name)
			}
			return fn()
		}
	}
	return []Step{
		{Name: "charge", Do: wrap("charge", func() error { return points.Deduct(pid, cost) }),
			Compensate: wrapComp("charge", func() error { return points.Refund(pid, cost) })},
		{Name: "grant", Do: wrap("grant", func() error { return rewardSvc.Grant(pid, itemID, reward) }),
			Compensate: wrapComp("grant", func() error { return rewardSvc.Revoke(pid, itemID, reward) })},
		{Name: "settle", Do: wrap("settle", func() error { return ledger.Append(orderID) }),
			Compensate: wrapComp("settle", func() error { return ledger.RemoveLast(orderID) })},
	}
}

// file: internal/saga/saga_test.go
// tier: L1
// verify: go test ./internal/saga/ -run TestSaga -v

package saga

import "testing"

func TestSaga(t *testing.T) {
	o := &Orchestrator{MaxCompRetry: 1}

	// (1) all steps succeed: committed, assets move exactly once.
	w := NewWorld()
	w.Accounts[1] = &Account{Points: 100, Items: map[uint32]int64{}}
	if res := o.Run(CrossSignup(w, 1, 30, 7001, 5, "order-A")); res.Status != StatusCommitted {
		t.Fatalf("committed: status=%v", res.Status)
	}
	if a := w.Accounts[1]; a.Points != 70 || a.Items[7001] != 5 || len(w.Ledger) != 1 {
		t.Fatalf("committed: %+v ledger=%v", a, w.Ledger)
	}

	// (2) last step fails: roll back newest-first, world back to the start.
	w2 := NewWorld()
	w2.Accounts[2] = &Account{Points: 100, Items: map[uint32]int64{}}
	w2.Fail["settle"] = true
	res := o.Run(CrossSignup(w2, 2, 30, 7001, 5, "order-B"))
	if res.Status != StatusRolledBack || res.FailedStep != "settle" {
		t.Fatalf("rollback: %+v", res)
	}
	if len(res.Compensated) != 2 || res.Compensated[0] != "grant" || res.Compensated[1] != "charge" {
		t.Fatalf("rollback order=%v want [grant charge]", res.Compensated)
	}
	if a := w2.Accounts[2]; a.Points != 100 || a.Items[7001] != 0 || len(w2.Ledger) != 0 {
		t.Fatalf("rollback not clean: %+v ledger=%v", a, w2.Ledger)
	}

	// (3) compensation itself fails: retried, then flagged for a human while
	// the other steps still roll back.
	w3 := NewWorld()
	w3.Accounts[3] = &Account{Points: 100, Items: map[uint32]int64{}}
	w3.Fail["settle"] = true
	w3.FailComp["grant"] = true
	res3 := o.Run(CrossSignup(w3, 3, 30, 7001, 5, "order-C"))
	if res3.Status != StatusManual || len(res3.ManualSteps) != 1 || res3.ManualSteps[0] != "grant" {
		t.Fatalf("manual: %+v", res3)
	}
	if len(res3.Compensated) != 1 || res3.Compensated[0] != "charge" || w3.Accounts[3].Points != 100 {
		t.Fatalf("manual: refund should still happen: %+v", res3)
	}
}