《后端架构设计·分布式与数据一致性》13.2 一次跨库下单的一致性验证

// file: internal/orderflow/orderflow.go
// tier: L2
// verify: go test ./internal/orderflow/ -count=1
package orderflow

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

// ErrTimeout models a dependency that committed but answered too late.
var ErrTimeout = errors.New("injected timeout after commit")

type Status string

const (
	Created   Status = "CREATED"
	Reserved  Status = "INVENTORY_RESERVED"
	Paid      Status = "PAID"
	Cancelled Status = "CANCELLED"
)

// Order is the aggregate owned by the order shard.
type Order struct {
	ID, UserID, SKU string
	Qty             int
	Status          Status
}

type Event struct {
	ID, Topic string
	Payload   any
}

type CreatedPayload struct {
	OrderID, SKU string
	Qty          int
}
type ReservedPayload struct{ OrderID string }
type RejectedPayload struct{ OrderID string }

// Shard holds one database shard's business rows, inbox and outbox. The
// order shard uses orders/payments; the inventory shard uses stock and
// reservations; the struct is shared to keep the lab small.
type Shard struct {
	mu       sync.Mutex
	orders   map[string]*Order
	held     map[string]int
	payments map[string]bool
	stock    int
	initial  int
	inbox    map[string]bool
	outbox   []Event
}

func newShard() *Shard {
	return &Shard{orders: make(map[string]*Order), held: make(map[string]int),
		payments: make(map[string]bool), inbox: make(map[string]bool)}
}

func (s *Shard) push(ev Event) { s.outbox = append(s.outbox, ev) }

// Broker is an at-least-once queue. duplicate makes every publish deliver
// twice; a timeout requeues the message after a handler returns an error.
type Broker struct {
	q         []Event
	duplicate bool
}

func (b *Broker) Publish(ev Event) {
	b.q = append(b.q, ev)
	if b.duplicate {
		b.q = append(b.q, ev)
	}
}

type System struct {
	orders, inventory *Shard
	broker            *Broker
	reserveTimeouts   map[string]bool
	payTimeouts       map[string]bool
	declines          map[string]bool
	relayCrash        bool
}

func NewSystem(stock int) *System {
	orders, inv := newShard(), newShard()
	inv.stock, inv.initial = stock, stock
	return &System{orders: orders, inventory: inv, broker: &Broker{},
		reserveTimeouts: map[string]bool{}, payTimeouts: map[string]bool{}, declines: map[string]bool{}}
}

// Place commits the order and its outbox row in one local transaction.
func (s *System) Place(id, user, sku string, qty int) {
	s.orders.mu.Lock()
	defer s.orders.mu.Unlock()
	s.orders.orders[id] = &Order{ID: id, UserID: user, SKU: sku, Qty: qty, Status: Created}
	s.orders.push(Event{ID: "created-" + id, Topic: "OrderCreated",
		Payload: CreatedPayload{OrderID: id, SKU: sku, Qty: qty}})
}

// Relay moves unsent outbox rows to the broker. A crash after publish leaves
// the rows in place, so the next relay publishes them again.
func (s *System) Relay() {
	for _, sh := range []*Shard{s.orders, s.inventory} {
		sh.mu.Lock()
		rows := append([]Event(nil), sh.outbox...)
		for _, ev := range rows {
			s.broker.Publish(ev)
		}
		if !s.relayCrash {
			sh.outbox = sh.outbox[len(rows):]
		}
		sh.mu.Unlock()
	}
	s.relayCrash = false
}

func (s *System) Drain(max int) {
	for i := 0; i < max; i++ {
		s.Relay()
		if len(s.broker.q) == 0 {
			return
		}
		ev := s.broker.q[0]
		s.broker.q = s.broker.q[1:]
		if err := s.Handle(ev); err != nil {
			s.broker.q = append(s.broker.q, ev)
		}
	}
}

func (s *System) Handle(ev Event) error {
	switch ev.Topic {
	case "OrderCreated":
		return s.reserve(ev)
	case "InventoryReserved":
		return s.confirm(ev)
	case "InventoryRejected":
		return s.reject(ev)
	case "ReleaseRequested":
		return s.release(ev)
	default:
		return fmt.Errorf("unknown topic %q", ev.Topic)
	}
}

// reserve is the inventory step. The inbox row is committed with the stock
// movement, so a duplicate or a timeout retry cannot reserve twice.
func (s *System) reserve(ev Event) error {
	s.inventory.mu.Lock()
	defer s.inventory.mu.Unlock()
	if s.inventory.inbox[ev.ID] {
		return nil
	}
	p := ev.Payload.(CreatedPayload)
	s.inventory.inbox[ev.ID] = true
	if s.inventory.stock < p.Qty {
		s.inventory.push(Event{ID: "rejected-" + p.OrderID, Topic: "InventoryRejected",
			Payload: RejectedPayload{OrderID: p.OrderID}})
		return nil
	}
	s.inventory.stock -= p.Qty
	s.inventory.held[p.OrderID] = p.Qty
	s.inventory.push(Event{ID: "reserved-" + p.OrderID, Topic: "InventoryReserved",
		Payload: ReservedPayload{OrderID: p.OrderID}})
	if s.reserveTimeouts[p.OrderID] {
		delete(s.reserveTimeouts, p.OrderID)
		return ErrTimeout
	}
	return nil
}

// confirm charges the payment and advances the order. The order ID is the
// payment idempotency key, so a retry after a timeout is free.
func (s *System) confirm(ev Event) error {
	s.orders.mu.Lock()
	defer s.orders.mu.Unlock()
	if s.orders.inbox[ev.ID] {
		return nil
	}
	p := ev.Payload.(ReservedPayload)
	o, ok := s.orders.orders[p.OrderID]
	if !ok {
		return fmt.Errorf("order %s not found", p.OrderID)
	}
	if o.Status == Created {
		o.Status = Reserved
	}
	if o.Status != Reserved {
		s.orders.inbox[ev.ID] = true
		return nil
	}
	if s.declines[p.OrderID] {
		delete(s.declines, p.OrderID)
		o.Status = Cancelled
		s.orders.inbox[ev.ID] = true
		s.orders.push(Event{ID: "release-" + o.ID, Topic: "ReleaseRequested",
			Payload: ReservedPayload{OrderID: o.ID}})
		return nil
	}
	if !s.orders.payments[o.ID] {
		s.orders.payments[o.ID] = true
	}
	if s.payTimeouts[o.ID] {
		delete(s.payTimeouts, o.ID)
		return ErrTimeout
	}
	o.Status = Paid
	s.orders.inbox[ev.ID] = true
	return nil
}

func (s *System) reject(ev Event) error {
	s.orders.mu.Lock()
	defer s.orders.mu.Unlock()
	if s.orders.inbox[ev.ID] {
		return nil
	}
	p := ev.Payload.(RejectedPayload)
	if o, ok := s.orders.orders[p.OrderID]; ok && o.Status == Created {
		o.Status = Cancelled
	}
	s.orders.inbox[ev.ID] = true
	return nil
}

// release is the compensation step: it returns the held quantity exactly
// once, keyed by the reservation row.
func (s *System) release(ev Event) error {
	s.inventory.mu.Lock()
	defer s.inventory.mu.Unlock()
	if s.inventory.inbox[ev.ID] {
		return nil
	}
	p := ev.Payload.(ReservedPayload)
	if qty, ok := s.inventory.held[p.OrderID]; ok {
		delete(s.inventory.held, p.OrderID)
		s.inventory.stock += qty
	}
	s.inventory.inbox[ev.ID] = true
	return nil
}

// Reconcile repairs what at-least-once delivery alone cannot: lost tail
// events, a payment recorded before the status update, and a cancelled order
// whose reservation was never released. Every repair is idempotent.
func (s *System) Reconcile() {
	s.orders.mu.Lock()
	snap := make([]Order, 0, len(s.orders.orders))
	for _, o := range s.orders.orders {
		snap = append(snap, *o)
	}
	paid := make(map[string]bool, len(s.orders.payments))
	for id := range s.orders.payments {
		paid[id] = true
	}
	s.orders.mu.Unlock()
	var repairs []Event
	for _, o := range snap {
		switch {
		case o.Status == Created && s.isHeld(o.ID):
			repairs = append(repairs, Event{ID: "reserved-" + o.ID, Topic: "InventoryReserved", Payload: ReservedPayload{o.ID}})
		case o.Status == Created:
			repairs = append(repairs, Event{ID: "created-" + o.ID, Topic: "OrderCreated", Payload: CreatedPayload{o.ID, o.SKU, o.Qty}})
		case o.Status == Reserved && paid[o.ID]:
			s.orders.mu.Lock()
			s.orders.orders[o.ID].Status = Paid
			s.orders.mu.Unlock()
		case o.Status == Reserved && s.isHeld(o.ID):
			repairs = append(repairs, Event{ID: "reserved-" + o.ID, Topic: "InventoryReserved", Payload: ReservedPayload{o.ID}})
		case o.Status == Cancelled && s.isHeld(o.ID):
			repairs = append(repairs, Event{ID: "release-" + o.ID, Topic: "ReleaseRequested", Payload: ReservedPayload{o.ID}})
		}
	}
	s.orders.mu.Lock()
	for _, ev := range repairs {
		s.orders.push(ev)
	}
	s.orders.mu.Unlock()
}

func (s *System) isHeld(orderID string) bool {
	s.inventory.mu.Lock()
	defer s.inventory.mu.Unlock()
	_, ok := s.inventory.held[orderID]
	return ok
}

// Diff returns every broken invariant; an empty slice means convergence.
func (s *System) Diff() []string {
	var out []string
	s.inventory.mu.Lock()
	held := 0
	for _, qty := range s.inventory.held {
		held += qty
	}
	if s.inventory.stock < 0 || s.inventory.stock+held != s.inventory.initial {
		out = append(out, fmt.Sprintf("stock=%d held=%d want=%d", s.inventory.stock, held, s.inventory.initial))
	}
	s.inventory.mu.Unlock()
	s.orders.mu.Lock()
	for _, o := range s.orders.orders {
		if o.Status == Paid && (!s.orders.payments[o.ID] || !s.isHeld(o.ID)) {
			out = append(out, "paid invariant broken: "+o.ID)
		}
		if o.Status == Cancelled && s.isHeld(o.ID) {
			out = append(out, "cancelled invariant broken: "+o.ID)
		}
	}
	s.orders.mu.Unlock()
	return out
}

=== RUN   TestCrossShardOrderFlow/baseline
    orderflow_test.go:68: paid=40 cancelled=0 stock=960 payments=40 diff=0
=== RUN   TestCrossShardOrderFlow/oversell
    orderflow_test.go:68: paid=5 cancelled=5 stock=0 payments=5 diff=0
=== RUN   TestCrossShardOrderFlow/decline_and_crash
    orderflow_test.go:68: paid=37 cancelled=3 stock=963 payments=37 diff=0
ok  	archlab/internal/orderflow	0.247s