《后端架构设计·分布式与数据一致性》10.3 锁失效后的正确性：从互斥到幂等

UPDATE inventory
SET stock = stock - 1, version = version + 1
WHERE sku = 'sku-1' AND stock >= 1 AND version = 7;

// file: internal/platform/correctness/inventory.go
// tier: L2
// verify: go test ./correctness/ -count=1
package correctness

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

var (
	// ErrConflict means the row changed between read and conditional write.
	ErrConflict = errors.New("correctness: version conflict")
	// ErrInsufficientStock means the reservation would make stock negative.
	ErrInsufficientStock = errors.New("correctness: insufficient stock")
	// ErrNotFound means the order does not exist.
	ErrNotFound = errors.New("correctness: order not found")
	// ErrInvalidState means the order is not in the expected source state.
	ErrInvalidState = errors.New("correctness: invalid order state")
)

const maxAttempts = 64

// Item is a stock row. Version is the optimistic concurrency token; no
// distributed lock participates in the update.
type Item struct {
	Stock   int64
	Version int64
}

// Reservation is the result of one successful stock reservation.
type Reservation struct {
	OrderID string
	SKU     string
	Qty     int64
}

// Store is a single-row atomic store. CAS is the only write primitive, which
// is what a database offers as UPDATE ... WHERE version = ?.
type Store struct {
	mu        sync.Mutex
	items     map[string]Item
	conflicts int
}

// NewStore returns an empty store.
func NewStore() *Store {
	return &Store{items: make(map[string]Item)}
}

// Seed inserts or replaces a stock row at version 0.
func (s *Store) Seed(sku string, stock int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.items[sku] = Item{Stock: stock, Version: 0}
}

// Load returns the current row; a missing row reads as zero.
func (s *Store) Load(sku string) Item {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.items[sku]
}

// CAS commits next only if the row is still at expected version.
func (s *Store) CAS(sku string, expected int64, next Item) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	cur, ok := s.items[sku]
	if !ok || cur.Version != expected {
		s.conflicts++
		return false
	}
	next.Version = expected + 1
	s.items[sku] = next
	return true
}

// Conflicts reports failed CAS attempts.
func (s *Store) Conflicts() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.conflicts
}

// Service reserves stock using read-check-CAS retries.
type Service struct {
	store *Store
}

// NewService wires a service to a store.
func NewService(store *Store) *Service { return &Service{store: store} }

// Reserve decrements stock for one order. It never holds a lock across the
// read and write; correctness comes from the version check and Stock >= 0.
func (s *Service) Reserve(orderID, sku string, qty int64) (Reservation, error) {
	for attempt := 0; attempt < maxAttempts; attempt++ {
		cur := s.store.Load(sku)
		if cur.Stock < qty {
			return Reservation{}, fmt.Errorf("reserve %s/%s qty=%d stock=%d: %w",
				orderID, sku, qty, cur.Stock, ErrInsufficientStock)
		}
		next := Item{Stock: cur.Stock - qty}
		if s.store.CAS(sku, cur.Version, next) {
			return Reservation{OrderID: orderID, SKU: sku, Qty: qty}, nil
		}
	}
	return Reservation{}, fmt.Errorf("reserve %s after %d attempts: %w", orderID, maxAttempts, ErrConflict)
}

// Idempotent applies fn at most once per key. The in-process mutex stands in
// for an INSERT ... ON CONFLICT DO NOTHING unique index; the effect must still
// be safe if the same key arrives after a crash.
type Idempotent struct {
	mu      sync.Mutex
	results map[string]Reservation
}

// NewIdempotent returns an empty idempotency table.
func NewIdempotent() *Idempotent {
	return &Idempotent{results: make(map[string]Reservation)}
}

// Do returns the cached result or executes fn once.
func (i *Idempotent) Do(key string, fn func() (Reservation, error)) (Reservation, error) {
	i.mu.Lock()
	defer i.mu.Unlock()
	if r, ok := i.results[key]; ok {
		return r, nil
	}
	r, err := fn()
	if err != nil {
		return Reservation{}, err
	}
	i.results[key] = r
	return r, nil
}

// OrderState is the order state machine value.
type OrderState string

// Order states used by arch-lab.
const (
	StateCreated   OrderState = "CREATED"
	StateReserved  OrderState = "INVENTORY_RESERVED"
	StatePaid      OrderState = "PAID"
	StateFulfilled OrderState = "FULFILLED"
	StateCancelled OrderState = "CANCELLED"
)

// Order is a versioned order row.
type Order struct {
	ID      string
	State   OrderState
	Version int64
}

// Orders stores orders and enforces conditional state transitions.
type Orders struct {
	mu     sync.Mutex
	orders map[string]Order
}

// NewOrders returns an empty order table.
func NewOrders() *Orders { return &Orders{orders: make(map[string]Order)} }

// Create inserts a new order in CREATED at version 0.
func (o *Orders) Create(id string) Order {
	o.mu.Lock()
	defer o.mu.Unlock()
	ord := Order{ID: id, State: StateCreated}
	o.orders[id] = ord
	return ord
}

// Get returns an order.
func (o *Orders) Get(id string) (Order, error) {
	o.mu.Lock()
	defer o.mu.Unlock()
	ord, ok := o.orders[id]
	if !ok {
		return Order{}, fmt.Errorf("order %s: %w", id, ErrNotFound)
	}
	return ord, nil
}

// Transition moves id from one state to another only if it is currently in
// from. A duplicate callback sees the new state and is rejected.
func (o *Orders) Transition(id string, from, to OrderState) error {
	o.mu.Lock()
	defer o.mu.Unlock()
	ord, ok := o.orders[id]
	if !ok {
		return fmt.Errorf("transition %s: %w", id, ErrNotFound)
	}
	if ord.State != from {
		return fmt.Errorf("transition %s from %s, current %s: %w", id, from, ord.State, ErrInvalidState)
	}
	ord.State = to
	ord.Version++
	o.orders[id] = ord
	return nil
}

$ go test ./correctness/ -v -count=1
=== RUN   TestNoOversellWithoutLock
    inventory_test.go:48: no lock: ok=50 short=150 final_stock=0 conflicts=1
--- PASS: TestNoOversellWithoutLock (0.00s)
=== RUN   TestCASRejectsStaleVersion
    inventory_test.go:66: stale CAS rejected: stock=4 version=1 conflicts=1
--- PASS: TestCASRejectsStaleVersion (0.00s)
=== RUN   TestReserveRetriesOnVersionConflict
--- PASS: TestReserveRetriesOnVersionConflict (0.00s)
=== RUN   TestDuplicateReservationAppliesOnce
    inventory_test.go:130: 20 duplicate callbacks: stock=9 version=1
--- PASS: TestDuplicateReservationAppliesOnce (0.00s)
=== RUN   TestOrderStateMachineRejectsDuplicateCallback
    inventory_test.go:152: duplicate callback: state=PAID version=2
--- PASS: TestOrderStateMachineRejectsDuplicateCallback (0.00s)
PASS
ok  	archlab/correctness	0.450s
$ go test ./correctness/ -race -count=1
ok  	archlab/correctness	1.260s