《后端架构设计·分布式与数据一致性》9.3 冲正、补偿与资金类操作的最终一致

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

import (
	"errors"
	"fmt"
	"sort"
	"sync"
)

// Posting is one leg of a double-entry transaction. Debits are positive,
// credits are negative, and every transaction must sum to zero.
type Posting struct {
	Account     string
	AmountCents int64
}

// Entry is one immutable posted transaction.
type Entry struct {
	TxnID      string
	Postings   []Posting
	ReversalOf string // empty unless this entry reverses another one
}

var (
	ErrUnbalanced      = errors.New("ledger: transaction does not sum to zero")
	ErrDuplicateTxn    = errors.New("ledger: transaction id already posted")
	ErrAlreadyReversed = errors.New("ledger: transaction already reversed")
	ErrUnknownTxn      = errors.New("ledger: unknown transaction")
	ErrInvalidPosting  = errors.New("ledger: invalid posting")
)

// Ledger is an append-only double-entry book. A mistake is corrected by
// reversing it, never by editing the original entry.
type Ledger struct {
	mu         sync.Mutex
	entries    map[string]Entry
	order      []string
	balances   map[string]int64
	reversedBy map[string]string
}

// New returns an empty ledger.
func New() *Ledger {
	return &Ledger{
		entries:    make(map[string]Entry),
		balances:   make(map[string]int64),
		reversedBy: make(map[string]string),
	}
}

func validate(e Entry) error {
	if e.TxnID == "" {
		return fmt.Errorf("ledger: empty txn id: %w", ErrInvalidPosting)
	}
	if len(e.Postings) < 2 {
		return fmt.Errorf("ledger: txn %s has %d postings, want at least 2: %w",
			e.TxnID, len(e.Postings), ErrInvalidPosting)
	}
	var sum int64
	accounts := make(map[string]bool, len(e.Postings))
	for _, p := range e.Postings {
		if p.Account == "" || p.AmountCents == 0 {
			return fmt.Errorf("ledger: txn %s has empty account or zero amount: %w", e.TxnID, ErrInvalidPosting)
		}
		if accounts[p.Account] {
			return fmt.Errorf("ledger: txn %s repeats account %q: %w", e.TxnID, p.Account, ErrInvalidPosting)
		}
		accounts[p.Account] = true
		sum += p.AmountCents
	}
	if sum != 0 {
		return fmt.Errorf("ledger: txn %s sums to %d: %w", e.TxnID, sum, ErrUnbalanced)
	}
	return nil
}

// Post appends a balanced transaction.
func (l *Ledger) Post(e Entry) error {
	if err := validate(e); err != nil {
		return err
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	if _, exists := l.entries[e.TxnID]; exists {
		return fmt.Errorf("ledger: post %s: %w", e.TxnID, ErrDuplicateTxn)
	}
	postings := append([]Posting(nil), e.Postings...)
	l.entries[e.TxnID] = Entry{TxnID: e.TxnID, Postings: postings, ReversalOf: e.ReversalOf}
	l.order = append(l.order, e.TxnID)
	for _, p := range postings {
		l.balances[p.Account] += p.AmountCents
	}
	return nil
}

// Reverse posts the exact negation of originalID. Repeating the same reversal
// id is a no-op; a second reversal under a different id is rejected.
func (l *Ledger) Reverse(originalID, reversalID string) (Entry, bool, error) {
	if reversalID == "" || reversalID == originalID {
		return Entry{}, false, fmt.Errorf("ledger: reverse %s with id %q: %w",
			originalID, reversalID, ErrInvalidPosting)
	}
	l.mu.Lock()
	defer l.mu.Unlock()

	original, ok := l.entries[originalID]
	if !ok {
		return Entry{}, false, fmt.Errorf("ledger: reverse %s: %w", originalID, ErrUnknownTxn)
	}
	if existing, done := l.reversedBy[originalID]; done {
		if existing != reversalID {
			return Entry{}, false, fmt.Errorf("ledger: reverse %s by %s: %w",
				originalID, existing, ErrAlreadyReversed)
		}
		return l.entries[existing], false, nil
	}
	if _, exists := l.entries[reversalID]; exists {
		return Entry{}, false, fmt.Errorf("ledger: reverse %s: id %s: %w",
			originalID, reversalID, ErrDuplicateTxn)
	}

	rev := Entry{TxnID: reversalID, ReversalOf: originalID}
	for _, p := range original.Postings {
		rev.Postings = append(rev.Postings, Posting{Account: p.Account, AmountCents: -p.AmountCents})
	}
	l.entries[reversalID] = rev
	l.order = append(l.order, reversalID)
	for _, p := range rev.Postings {
		l.balances[p.Account] += p.AmountCents
	}
	l.reversedBy[originalID] = reversalID
	return rev, true, nil
}

// Balance returns the signed balance of one account.
func (l *Ledger) Balance(account string) int64 {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.balances[account]
}

// TotalBalance is the sum of every account balance; double entry keeps it 0.
func (l *Ledger) TotalBalance() int64 {
	l.mu.Lock()
	defer l.mu.Unlock()
	var total int64
	for _, v := range l.balances {
		total += v
	}
	return total
}

// Entries returns a copy in posting order.
func (l *Ledger) Entries() []Entry {
	l.mu.Lock()
	defer l.mu.Unlock()
	out := make([]Entry, 0, len(l.order))
	for _, id := range l.order {
		e := l.entries[id]
		e.Postings = append([]Posting(nil), e.Postings...)
		out = append(out, e)
	}
	return out
}

// Check recomputes invariants from the entries instead of trusting the running
// balances, and returns violations in stable order.
func (l *Ledger) Check() []string {
	l.mu.Lock()
	defer l.mu.Unlock()
	var violations []string
	recomputed := make(map[string]int64)
	for _, id := range l.order {
		e := l.entries[id]
		var sum int64
		for _, p := range e.Postings {
			sum += p.AmountCents
			recomputed[p.Account] += p.AmountCents
		}
		if sum != 0 {
			violations = append(violations, fmt.Sprintf("txn %s is unbalanced: %d", id, sum))
		}
	}
	for account, want := range recomputed {
		if got := l.balances[account]; got != want {
			violations = append(violations, fmt.Sprintf("account %s balance %d, recomputed %d", account, got, want))
		}
	}
	for _, id := range l.order {
		revID, ok := l.reversedBy[id]
		if !ok {
			continue
		}
		original, rev := l.entries[id], l.entries[revID]
		if rev.ReversalOf != id {
			violations = append(violations, fmt.Sprintf("reversal %s points to %q, want %s", revID, rev.ReversalOf, id))
			continue
		}
		orig := make(map[string]int64, len(original.Postings))
		for _, p := range original.Postings {
			orig[p.Account] = p.AmountCents
		}
		for _, p := range rev.Postings {
			if orig[p.Account] != -p.AmountCents {
				violations = append(violations, fmt.Sprintf("reversal %s does not negate %s on %s", revID, id, p.Account))
			}
		}
	}
	var total int64
	for _, v := range l.balances {
		total += v
	}
	if total != 0 {
		violations = append(violations, fmt.Sprintf("total balance is %d, want 0", total))
	}
	sort.Strings(violations)
	return violations
}

$ go test ./internal/platform/ledger/ -count=1 -v
=== RUN   TestPostBalancedPayment
    ledger_test.go:40: channel=1999 suspense=-1999 total=0 violations=0
=== RUN   TestUnbalancedTxnRejected
    ledger_test.go:60: single_leg=ErrInvalidPosting lopsided=ErrUnbalanced stored=0
=== RUN   TestReverseRestoresBalances
    ledger_test.go:83: suspense_before=-1999 suspense_after=0 entries=2 total=0 violations=0
=== RUN   TestSameReversalIsIdempotent
    ledger_test.go:102: second_reversal created=false entries=2
=== RUN   TestSecondReversalUnderNewIDRejected
    ledger_test.go:117: double_reversal_error=ledger: reverse pay-1001 by rev-1001: ledger: transaction already reversed entries=2
=== RUN   TestHangingSuspenseClearsAfterResolve
    ledger_test.go:142: hanging_cents=-1999 suspense_after=0 receivable=-1999 total=0
=== RUN   TestConcurrentPostsConserve
    ledger_test.go:193: posts=128 channel=12800 receivable=-12800 total=0 violations=0
PASS
ok  	archlab/internal/platform/ledger	0.385s