《后端架构设计·分布式与数据一致性》8.1 事务性 Outbox：同库写入与投递保证

if err := tx.Commit(); err != nil {
    return err
}
if err := mq.Publish(ctx, "order.paid", event); err != nil {
    return err
}

position = 0
loop:
  rows = SELECT * FROM outbox WHERE seq > position ORDER BY seq LIMIT batch
  for row in rows:
    broker.send(row)          -- 失败则退出本轮，不推进
    UPDATE relay_position SET seq = row.seq

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

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

var (
	// ErrConflict is returned when an order ID is written twice.
	ErrConflict = errors.New("order conflict")
	// ErrInvalidState is returned when an order fails validation.
	ErrInvalidState = errors.New("invalid order state")
	// ErrSimulatedCrash marks an injected crash between send and offset save.
	ErrSimulatedCrash = errors.New("simulated crash")
)

// Order is the aggregate persisted by the order service.
type Order struct {
	ID          string
	AmountCents int64
	Status      string
}

// Message is one outbox row. Seq is assigned inside the same transaction as
// the business write and is the relay's delivery order.
type Message struct {
	Seq     int64
	ID      string
	Topic   string
	Key     string
	Payload string
}

// Store models one database: business rows, outbox rows and the relay offset
// all live in it. Each method is a single local transaction.
type Store struct {
	mu       sync.Mutex
	orders   map[string]Order
	outbox   []Message
	position int64
	seq      int64
}

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

// PlaceOrder writes the order and its outbox row atomically. A returned
// error means neither write happened.
func (s *Store) PlaceOrder(o Order) (Message, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, ok := s.orders[o.ID]; ok {
		return Message{}, fmt.Errorf("outbox: order %s exists: %w", o.ID, ErrConflict)
	}
	if o.AmountCents <= 0 {
		return Message{}, fmt.Errorf("outbox: amount %d: %w", o.AmountCents, ErrInvalidState)
	}
	if o.Status == "" {
		o.Status = "CREATED"
	}
	s.orders[o.ID] = o
	s.seq++
	msg := Message{
		Seq:     s.seq,
		ID:      fmt.Sprintf("evt-%s-%d", o.ID, s.seq),
		Topic:   "order.created",
		Key:     o.ID,
		Payload: fmt.Sprintf("%s:%d", o.ID, o.AmountCents),
	}
	s.outbox = append(s.outbox, msg)
	return msg, nil
}

// Position returns the last contiguous offset the relay marked delivered.
func (s *Store) Position() int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.position
}

// AdvancePosition moves the relay offset forward. It never moves backward.
func (s *Store) AdvancePosition(seq int64) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if seq <= s.position {
		return fmt.Errorf("outbox: position %d already at %d: %w", seq, s.position, ErrInvalidState)
	}
	s.position = seq
	return nil
}

// Pending returns unsent rows after the offset, in Seq order.
func (s *Store) Pending(after int64, limit int) []Message {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Message, 0, limit)
	for _, msg := range s.outbox {
		if msg.Seq <= after {
			continue
		}
		out = append(out, msg)
		if len(out) == limit {
			break
		}
	}
	return out
}

// OutboxLen reports how many outbox rows exist.
func (s *Store) OutboxLen() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.outbox)
}

// Sender delivers one message to the broker.
type Sender interface {
	Send(Message) error
}

// Relay polls the outbox and advances the offset only after a successful
// send. CrashAfterSend injects the crash window for tests.
type Relay struct {
	store          *Store
	sender         Sender
	batch          int
	CrashAfterSend bool
}

// NewRelay returns a relay with a fixed batch size.
func NewRelay(store *Store, sender Sender, batch int) *Relay {
	if batch <= 0 {
		batch = 100
	}
	return &Relay{store: store, sender: sender, batch: batch}
}

// RunOnce polls one batch. It is safe to call again after an error: rows the
// offset did not cover are redelivered.
func (r *Relay) RunOnce() (delivered int, err error) {
	msgs := r.store.Pending(r.store.Position(), r.batch)
	for _, msg := range msgs {
		if sendErr := r.sender.Send(msg); sendErr != nil {
			return delivered, fmt.Errorf("relay: send seq %d: %w", msg.Seq, sendErr)
		}
		delivered++
		if r.CrashAfterSend {
			r.CrashAfterSend = false
			return delivered, ErrSimulatedCrash
		}
		if advErr := r.store.AdvancePosition(msg.Seq); advErr != nil {
			return delivered, fmt.Errorf("relay: advance to %d: %w", msg.Seq, advErr)
		}
	}
	return delivered, nil
}

$ go test ./internal/platform/outbox/ -v -count=1
=== RUN   TestCommitThenCrash
    outbox_test.go:77: commit-then-crash: rows=3 delivered=3 position=3 duplicates=0
--- PASS: TestCommitThenCrash (0.00s)
=== RUN   TestCrashAfterDelivery
    outbox_test.go:110: crash-after-delivery: seq=1 deliveries=2 position=1 outcome=duplicate
--- PASS: TestCrashAfterDelivery (0.00s)
=== RUN   TestFailedWriteLeavesNoOutbox
    outbox_test.go:134: failed-writes: outbox_rows=1 orders=1
--- PASS: TestFailedWriteLeavesNoOutbox (0.00s)
=== RUN   TestBatchDeliveryOrder
    outbox_test.go:162: batch-order: rows=5 batch=2 runs=3 sequence=1..5
--- PASS: TestBatchDeliveryOrder (0.00s)
=== RUN   TestSendErrorKeepsPosition
    outbox_test.go:190: send-error: position_after_failure=1 position_after_retry=3 attempts=4
--- PASS: TestSendErrorKeepsPosition (0.00s)
PASS
ok  	archlab/internal/platform/outbox	0.445s