《后端架构设计·分布式与数据一致性》8.3 消息顺序、重放与死信的对账闭环

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

import (
	"errors"
	"fmt"
	"hash/fnv"
	"sync"
)

var (
	// ErrConflict means the event arrived before the version it depends on.
	ErrConflict = errors.New("event version conflict")
	// ErrPoison means the event can never be applied as written.
	ErrPoison = errors.New("poison event")
)

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

// Order projection states, matching the arch-lab state machine.
const (
	StatusCreated   Status = "CREATED"
	StatusReserved  Status = "INVENTORY_RESERVED"
	StatusPaid      Status = "PAID"
	StatusFulfilled Status = "FULFILLED"
)

// Event is one order event. Version is the projection version it produces.
type Event struct {
	ID      string
	OrderID string
	Version int64
	Type    string
}

// OrderState is the projection row for one order.
type OrderState struct {
	Version int64
	Status  Status
}

func eventTarget(e Event) (Status, int64, error) {
	switch e.Type {
	case "reserve":
		return StatusReserved, 1, nil
	case "pay":
		return StatusPaid, 2, nil
	case "fulfill":
		return StatusFulfilled, 3, nil
	default:
		return "", 0, fmt.Errorf("ordering: event %s has type %q: %w", e.ID, e.Type, ErrPoison)
	}
}

// Projection applies order events in version order.
type Projection struct {
	mu      sync.Mutex
	orders  map[string]OrderState
	applied int
}

// NewProjection returns an empty projection.
func NewProjection() *Projection {
	return &Projection{orders: make(map[string]OrderState)}
}

// Apply advances the order to e.Version. It reports false for duplicates and
// returns ErrConflict when an earlier version is missing.
func (p *Projection) Apply(e Event) (bool, error) {
	status, expected, err := eventTarget(e)
	if err != nil {
		return false, err
	}
	if e.Version != expected {
		return false, fmt.Errorf("ordering: %s expects v%d, got v%d: %w", e.Type, expected, e.Version, ErrPoison)
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	cur := p.orders[e.OrderID]
	if e.Version <= cur.Version {
		return false, nil
	}
	if e.Version != cur.Version+1 {
		return false, fmt.Errorf("ordering: order %s at v%d, got v%d: %w", e.OrderID, cur.Version, e.Version, ErrConflict)
	}
	p.orders[e.OrderID] = OrderState{Version: e.Version, Status: status}
	p.applied++
	return true, nil
}

// Get returns the projection row for one order.
func (p *Projection) Get(orderID string) (OrderState, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	st, ok := p.orders[orderID]
	return st, ok
}

// AppliedCount reports how many version advances were applied.
func (p *Projection) AppliedCount() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.applied
}

// Partition maps an order key to a stable partition.
func Partition(orderID string, partitions int) int {
	if partitions <= 0 {
		return 0
	}
	h := fnv.New32a()
	_, _ = h.Write([]byte(orderID))
	return int(h.Sum32() % uint32(partitions))
}

// Outcome is what the dispatcher should do with a delivery.
type Outcome string

// Dispatch outcomes.
const (
	OutcomeApplied   Outcome = "applied"
	OutcomeDuplicate Outcome = "duplicate"
	OutcomeRetry     Outcome = "retry"
	OutcomeDead      Outcome = "dead"
)

// DLQEntry is one dead-lettered event.
type DLQEntry struct {
	Event    Event
	Attempts int
	Reason   string
}

// Consumer applies events for one partition and classifies failures.
type Consumer struct {
	proj        *Projection
	MaxAttempts int
	dlq         []DLQEntry
}

// NewConsumer returns a consumer with a retry budget per event.
func NewConsumer(p *Projection, maxAttempts int) *Consumer {
	if maxAttempts <= 0 {
		maxAttempts = 3
	}
	return &Consumer{proj: p, MaxAttempts: maxAttempts}
}

// Handle classifies one delivery. attempt is the 1-based broker delivery
// count; the caller redelivers when the outcome is OutcomeRetry.
func (c *Consumer) Handle(e Event, attempt int) Outcome {
	applied, err := c.proj.Apply(e)
	switch {
	case err == nil && applied:
		return OutcomeApplied
	case err == nil:
		return OutcomeDuplicate
	case errors.Is(err, ErrPoison):
		c.dead(e, attempt, err)
		return OutcomeDead
	case errors.Is(err, ErrConflict):
		if attempt >= c.MaxAttempts {
			c.dead(e, attempt, err)
			return OutcomeDead
		}
		return OutcomeRetry
	default:
		c.dead(e, attempt, err)
		return OutcomeDead
	}
}

func (c *Consumer) dead(e Event, attempt int, err error) {
	c.dlq = append(c.dlq, DLQEntry{Event: e, Attempts: attempt, Reason: err.Error()})
}

// DLQ returns a copy of the dead-letter queue.
func (c *Consumer) DLQ() []DLQEntry {
	return append([]DLQEntry(nil), c.dlq...)
}

// Source is the authoritative order store used by reconciliation.
type Source interface {
	Status(orderID string) (OrderState, bool)
}

// MapSource is an in-memory Source for tests and drills.
type MapSource map[string]OrderState

// Status implements Source.
func (m MapSource) Status(orderID string) (OrderState, bool) {
	st, ok := m[orderID]
	return st, ok
}

// Fix records one reconciliation repair.
type Fix struct {
	OrderID string
	From    OrderState
	To      OrderState
}

// Reconcile aligns the projection with the source of truth. It repairs
// differences that replay cannot: events that were dead-lettered or never
// produced.
func Reconcile(p *Projection, src Source, orderIDs []string) []Fix {
	p.mu.Lock()
	defer p.mu.Unlock()
	var fixes []Fix
	for _, id := range orderIDs {
		want, ok := src.Status(id)
		if !ok {
			continue
		}
		got := p.orders[id]
		if got == want {
			continue
		}
		p.orders[id] = want
		fixes = append(fixes, Fix{OrderID: id, From: got, To: want})
	}
	return fixes
}

$ go test ./internal/platform/ordering/ -v -count=1
=== RUN   TestPartitionAffinity
    ordering_test.go:32: partition-affinity: orders=8 partitions=4 used=4 stable=yes
--- PASS: TestPartitionAffinity (0.00s)
=== RUN   TestOutOfOrderRejectedThenConverges
    ordering_test.go:60: out-of-order: attempts=3 applied=2 final=v2/PAID dlq=0
--- PASS: TestOutOfOrderRejectedThenConverges (0.00s)
=== RUN   TestPoisonGoesStraightToDLQ
    ordering_test.go:85: poison: events=2 dead=2 attempts=1/1 applied=0
--- PASS: TestPoisonGoesStraightToDLQ (0.00s)
=== RUN   TestConflictExhaustsRetryBudget
    ordering_test.go:109: retry-budget: max=3 attempts=3 outcome=dead applied=0
--- PASS: TestConflictExhaustsRetryBudget (0.00s)
=== RUN   TestReconcileRepairsDeadLettered
    ordering_test.go:135: reconcile: before=v1/INVENTORY_RESERVED after=v2/PAID fixes=1 dlq=1
--- PASS: TestReconcileRepairsDeadLettered (0.00s)
=== RUN   TestReplayRebuildsProjection
    ordering_test.go:207: replay: events=5 rebuilt_applied=5 duplicates=2 offset=5
--- PASS: TestReplayRebuildsProjection (0.00s)
=== RUN   TestNextBatchBounds
    ordering_test.go:223: seek-bounds: out_of_range=1 zero_size=1 drained=1
--- PASS: TestNextBatchBounds (0.00s)
PASS
ok  	archlab/internal/platform/ordering	0.231s