《后端架构设计·分布式与数据一致性》6.3 从一致性模型到业务不变量

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

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

// Sentinel errors shared by the invariant checks.
var (
	ErrInsufficientStock = errors.New("invariant: insufficient stock")
	ErrConflict          = errors.New("invariant: version conflict")
	ErrInvalidState      = errors.New("invariant: invalid state transition")
	ErrDuplicateCharge   = errors.New("invariant: duplicate charge")
)

// Inventory tracks one SKU in integer units. Reserved and Sold must never
// exceed Total; available is a derived value, not stored state.
type Inventory struct {
	mu       sync.Mutex
	total    int64
	reserved int64
	sold     int64
	version  uint64
}

// NewInventory creates a stock record with the given total.
func NewInventory(total int64) *Inventory { return &Inventory{total: total} }

// Available returns total - reserved - sold under the current lock.
func (i *Inventory) Available() int64 {
	i.mu.Lock()
	defer i.mu.Unlock()
	return i.total - i.reserved - i.sold
}

// ReserveIfAvailable is the atomic conditional update: the check and the
// increment happen under one lock, so the invariant cannot be interleaved.
func (i *Inventory) ReserveIfAvailable(qty int64) (uint64, error) {
	if qty <= 0 {
		return 0, fmt.Errorf("invariant: reserve qty must be positive, got %d", qty)
	}
	i.mu.Lock()
	defer i.mu.Unlock()
	if i.total-i.reserved-i.sold < qty {
		return 0, fmt.Errorf("%w: available %d, want %d", ErrInsufficientStock, i.total-i.reserved-i.sold, qty)
	}
	i.reserved += qty
	i.version++
	return i.version, nil
}

// ReserveFromStaleSnapshot encodes the weaker read-then-act model: the caller
// supplies an available value read earlier and the write does not revalidate.
// It exists to show which consistency level the invariant actually needs.
func (i *Inventory) ReserveFromStaleSnapshot(qty, available int64) error {
	if available < qty {
		return fmt.Errorf("%w: snapshot available %d, want %d", ErrInsufficientStock, available, qty)
	}
	i.mu.Lock()
	defer i.mu.Unlock()
	i.reserved += qty
	i.version++
	return nil
}

// Check validates total >= reserved + sold and non-negative fields.
func (i *Inventory) Check() error {
	i.mu.Lock()
	defer i.mu.Unlock()
	if i.total < 0 || i.reserved < 0 || i.sold < 0 {
		return fmt.Errorf("invariant: negative inventory field: total=%d reserved=%d sold=%d", i.total, i.reserved, i.sold)
	}
	if i.reserved+i.sold > i.total {
		return fmt.Errorf("invariant: oversold: total=%d reserved=%d sold=%d", i.total, i.reserved, i.sold)
	}
	return nil
}

// Status is the order lifecycle state.
type Status string

// Order states; terminal states have no outgoing edges.
const (
	Created           Status = "CREATED"
	InventoryReserved Status = "INVENTORY_RESERVED"
	Paid              Status = "PAID"
	Fulfilled         Status = "FULFILLED"
	Cancelled         Status = "CANCELLED"
	Refunded          Status = "REFUNDED"
)

var allowed = map[Status]map[Status]bool{
	Created:           {InventoryReserved: true, Cancelled: true},
	InventoryReserved: {Paid: true, Cancelled: true},
	Paid:              {Fulfilled: true, Refunded: true},
	Fulfilled:         {Refunded: true},
	Cancelled:         {},
	Refunded:          {},
}

// ValidateTransition accepts exactly the edges of the order state machine.
func ValidateTransition(from, to Status) error {
	if allowed[from][to] {
		return nil
	}
	return fmt.Errorf("%w: %s -> %s", ErrInvalidState, from, to)
}

// CheckStatusSequence validates a recorded state history, including the first
// state, so a sequence that starts in the middle is rejected only if the first
// state is not a valid starting point.
func CheckStatusSequence(states []Status) error {
	if len(states) == 0 {
		return fmt.Errorf("%w: empty status sequence", ErrInvalidState)
	}
	if states[0] != Created {
		return fmt.Errorf("%w: sequence starts at %s", ErrInvalidState, states[0])
	}
	for i := 1; i < len(states); i++ {
		if err := ValidateTransition(states[i-1], states[i]); err != nil {
			return fmt.Errorf("invariant: step %d: %w", i, err)
		}
	}
	return nil
}

// Order updates its status through an expected-version compare-and-set.
type Order struct {
	mu      sync.Mutex
	id      string
	status  Status
	version uint64
}

// NewOrder creates an order in CREATED at version zero.
func NewOrder(id string) *Order { return &Order{id: id, status: Created} }

// Snapshot returns the current status and version atomically.
func (o *Order) Snapshot() (Status, uint64) {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.status, o.version
}

// Transition applies one edge only if the caller's version is current.
func (o *Order) Transition(to Status, expectedVersion uint64) (uint64, error) {
	o.mu.Lock()
	defer o.mu.Unlock()
	if o.version != expectedVersion {
		return 0, fmt.Errorf("%w: order %s version %d, caller expected %d", ErrConflict, o.id, o.version, expectedVersion)
	}
	if err := ValidateTransition(o.status, to); err != nil {
		return 0, err
	}
	o.status = to
	o.version++
	return o.version, nil
}

// PaymentBook enforces at most one successful charge per order.
type PaymentBook struct {
	mu      sync.Mutex
	charged map[string]int64
	seen    map[string]bool
}

// NewPaymentBook creates an empty charge ledger.
func NewPaymentBook() *PaymentBook {
	return &PaymentBook{charged: make(map[string]int64), seen: make(map[string]bool)}
}

// Charge records a charge exactly once per idempotency key. It returns true
// when this call created the charge and false for a replay.
func (p *PaymentBook) Charge(orderID, idemKey string, amount int64) (bool, error) {
	if orderID == "" || idemKey == "" || amount <= 0 {
		return false, fmt.Errorf("invariant: invalid charge order=%q key=%q amount=%d", orderID, idemKey, amount)
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	key := orderID + "|" + idemKey
	if p.seen[key] {
		return false, nil
	}
	if _, ok := p.charged[orderID]; ok {
		return false, fmt.Errorf("%w: order %s already charged", ErrDuplicateCharge, orderID)
	}
	p.seen[key] = true
	p.charged[orderID] = amount
	return true, nil
}

// Charged returns the recorded amount for an order.
func (p *PaymentBook) Charged(orderID string) (int64, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	amount, ok := p.charged[orderID]
	return amount, ok
}

package invariant

import (
	"errors"
	"sync"
	"testing"
)

func TestStaleSnapshotOversells(t *testing.T) {
	stock := NewInventory(1)
	snapshot := stock.Available()

	var wg sync.WaitGroup
	start := make(chan struct{})
	errs := make(chan error, 2)
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			errs <- stock.ReserveFromStaleSnapshot(1, snapshot)
		}()
	}
	close(start)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("stale reservation rejected: %v", err)
		}
	}
	if err := stock.Check(); err == nil {
		t.Fatalf("Check = nil, want an oversell violation")
	}
}

func TestConditionalUpdateKeepsInvariant(t *testing.T) {
	stock := NewInventory(1)

	var wg sync.WaitGroup
	start := make(chan struct{})
	errs := make(chan error, 2)
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			_, err := stock.ReserveIfAvailable(1)
			errs <- err
		}()
	}
	close(start)
	wg.Wait()
	close(errs)

	var ok, insufficient int
	for err := range errs {
		switch {
		case err == nil:
			ok++
		case errors.Is(err, ErrInsufficientStock):
			insufficient++
		default:
			t.Fatalf("unexpected error: %v", err)
		}
	}
	if ok != 1 || insufficient != 1 {
		t.Fatalf("ok=%d insufficient=%d, want 1 and 1", ok, insufficient)
	}
	if err := stock.Check(); err != nil {
		t.Fatalf("Check: %v", err)
	}
}

$ go test ./article/invariant/ -count=1 -v
=== RUN   TestStaleSnapshotOversells
--- PASS: TestStaleSnapshotOversells (0.00s)
=== RUN   TestConditionalUpdateKeepsInvariant
--- PASS: TestConditionalUpdateKeepsInvariant (0.00s)
=== RUN   TestOrderStateSequence
--- PASS: TestOrderStateSequence (0.00s)
=== RUN   TestConcurrentTransitionOneWins
--- PASS: TestConcurrentTransitionOneWins (0.00s)
=== RUN   TestDuplicateChargeIsRejected
--- PASS: TestDuplicateChargeIsRejected (0.00s)
PASS
ok  	archlab/article/invariant	0.489s