《后端架构设计·分布式与数据一致性》8.2 消费端幂等、去重表与恰好一次的真实边界

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

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

var (
	// ErrAckLost marks a crash between the local commit and the broker ack.
	ErrAckLost = errors.New("ack lost")
	// ErrSimulatedCrash marks an injected crash in the unsafe ordering test.
	ErrSimulatedCrash = errors.New("simulated crash")
)

// Message is one broker delivery. Seq is the producer's outbox sequence and
// bounds the dedup retention window.
type Message struct {
	Producer    string
	ID          string
	Seq         int64
	OrderID     string
	AmountCents int64
}

// DedupKey scopes the event ID by producer, so two producers may reuse an ID.
func DedupKey(m Message) string {
	return m.Producer + ":" + m.ID
}

// Ledger applies payment amounts to orders. Effects and dedup keys live in
// the same local store, which is what makes the two writes atomic.
type Ledger struct {
	mu       sync.Mutex
	balances map[string]int64
	applied  map[string]int64
	seenSeq  map[string]int64
	applies  int
}

// NewLedger returns an empty ledger.
func NewLedger() *Ledger {
	return &Ledger{
		balances: make(map[string]int64),
		applied:  make(map[string]int64),
		seenSeq:  make(map[string]int64),
	}
}

// ApplyOnce applies the effect and records the dedup key in one critical
// section, modelling one local transaction. It returns false for a duplicate.
func (l *Ledger) ApplyOnce(m Message) (bool, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	key := DedupKey(m)
	if _, ok := l.applied[key]; ok {
		return false, nil
	}
	l.balances[m.OrderID] += m.AmountCents
	l.applied[key] = m.AmountCents
	l.seenSeq[key] = m.Seq
	l.applies++
	return true, nil
}

// NaiveApply applies every delivery. It exists to measure the cost of not
// deduplicating; it is not a safe handler.
func (l *Ledger) NaiveApply(m Message) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.balances[m.OrderID] += m.AmountCents
	l.applies++
}

// MarkBeforeEffect records the dedup key before applying the effect. The
// order is intentionally wrong: a crash in between loses the message, because
// redelivery is mistaken for a duplicate.
func (l *Ledger) MarkBeforeEffect(m Message, crash bool) error {
	l.mu.Lock()
	defer l.mu.Unlock()
	key := DedupKey(m)
	if _, ok := l.applied[key]; ok {
		return nil
	}
	l.applied[key] = m.AmountCents
	l.seenSeq[key] = m.Seq
	if crash {
		return ErrSimulatedCrash
	}
	l.balances[m.OrderID] += m.AmountCents
	l.applies++
	return nil
}

// Prune forgets dedup keys with Seq <= maxSeq, simulating a retention window.
func (l *Ledger) Prune(maxSeq int64) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	removed := 0
	for key, seq := range l.seenSeq {
		if seq <= maxSeq {
			delete(l.applied, key)
			delete(l.seenSeq, key)
			removed++
		}
	}
	return removed
}

// Balance returns the applied amount for an order.
func (l *Ledger) Balance(orderID string) int64 {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.balances[orderID]
}

// Applies reports how many effects were applied.
func (l *Ledger) Applies() int {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.applies
}

// AckStore acknowledges a delivery to the broker.
type AckStore interface {
	Ack(Message) error
}

// MemoryAck records acks and can lose exactly one ack on demand.
type MemoryAck struct {
	mu    sync.Mutex
	acked map[string]bool
	Lost  bool
	acks  int
}

// NewMemoryAck returns an empty ack store.
func NewMemoryAck() *MemoryAck {
	return &MemoryAck{acked: make(map[string]bool)}
}

// Ack records the message or simulates a lost ack.
func (a *MemoryAck) Ack(m Message) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.acks++
	if a.Lost {
		a.Lost = false
		return ErrAckLost
	}
	a.acked[DedupKey(m)] = true
	return nil
}

// Acked reports whether the dedup key was acknowledged.
func (a *MemoryAck) Acked(m Message) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.acked[DedupKey(m)]
}

// Acks returns the number of ack attempts.
func (a *MemoryAck) Acks() int {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.acks
}

// Consume applies one delivery and then acks it. The order is the whole
// effect-once argument: effect first, ack second.
func Consume(l *Ledger, m Message, ack AckStore) (bool, error) {
	applied, err := l.ApplyOnce(m)
	if err != nil {
		return false, fmt.Errorf("inbox: apply %s: %w", DedupKey(m), err)
	}
	if err := ack.Ack(m); err != nil {
		return applied, fmt.Errorf("inbox: ack %s: %w", DedupKey(m), err)
	}
	return applied, nil
}

$ go test ./internal/platform/inbox/ -v -count=1
=== RUN   TestSideEffectThenAckLost
    inbox_test.go:41: ack-lost: deliveries=2 effects=1 balance=2599 acks=2
--- PASS: TestSideEffectThenAckLost (0.00s)
=== RUN   TestNaiveRedeliveryDoubleApplies
    inbox_test.go:56: naive-redelivery: deliveries=2 effects=2 balance=5198
--- PASS: TestNaiveRedeliveryDoubleApplies (0.00s)
=== RUN   TestDedupKeyIncludesProducer
    inbox_test.go:73: dedup-scope: producers=2 same_event_id=evt-1 effects=2 balance=600
--- PASS: TestDedupKeyIncludesProducer (0.00s)
=== RUN   TestPruneOpensReplayWindow
    inbox_test.go:98: prune-window: pruned=1 replay_applied=true balance=2400
--- PASS: TestPruneOpensReplayWindow (0.00s)
=== RUN   TestMarkBeforeEffectLosesMessage
    inbox_test.go:120: mark-before-effect: balance=0 effects=0 redelivery_applied=false
--- PASS: TestMarkBeforeEffectLosesMessage (0.00s)
PASS
ok  	archlab/internal/platform/inbox	0.304s