《后端架构设计·分布式与数据一致性》7.3 TCC、本地消息表与事务状态的可见性

// file: internal/platform/txn/tcc/tcc.go
// tier: L1
// verify: go test ./internal/platform/txn/tcc/ -count=1
package tcc

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

// State is the durable transaction state.
type State string

const (
	StateTrying    State = "TRYING"
	StateConfirmed State = "CONFIRMED"
	StateCancelled State = "CANCELLED"
)

var (
	// ErrCancelled is returned when Try arrives after a Cancel tombstone.
	ErrCancelled = errors.New("tcc: already cancelled")
	// ErrInvalidState reports a transition that violates the state machine.
	ErrInvalidState = errors.New("tcc: invalid state")
)

// Message is one row of the local message table.
type Message struct {
	TxID    string
	Payload string
	Sent    bool
}

type record struct {
	state    State
	reserved int64
}

// Store holds TCC state and the local message table behind one mutex. In a
// real service the business row and the message row are written in one
// database transaction.
type Store struct {
	mu       sync.Mutex
	balance  int64
	txs      map[string]*record
	messages []*Message
}

// NewStore creates a store with the given available balance in cents.
func NewStore(balance int64) *Store {
	return &Store{balance: balance, txs: make(map[string]*record)}
}

// Try reserves amount and appends a TRYING message. A Cancel that arrived
// first leaves a tombstone, so the late Try is rejected (no hanging).
func (s *Store) Try(txID string, amount int64) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if r, ok := s.txs[txID]; ok {
		switch r.state {
		case StateTrying:
			return nil
		case StateCancelled:
			return fmt.Errorf("tcc: try %s: %w", txID, ErrCancelled)
		default:
			return fmt.Errorf("tcc: try %s: %w", txID, ErrInvalidState)
		}
	}
	if amount <= 0 {
		return fmt.Errorf("tcc: try %s: amount must be positive", txID)
	}
	if s.balance < amount {
		return fmt.Errorf("tcc: try %s: balance %d < %d", txID, s.balance, amount)
	}
	s.balance -= amount
	s.txs[txID] = &record{state: StateTrying, reserved: amount}
	s.messages = append(s.messages, &Message{TxID: txID, Payload: "reserved"})
	return nil
}

// Confirm commits the reservation. Repeat calls are no-ops.
func (s *Store) Confirm(txID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.txs[txID]
	if !ok {
		return fmt.Errorf("tcc: confirm %s: %w", txID, ErrInvalidState)
	}
	if r.state == StateConfirmed {
		return nil
	}
	if r.state != StateTrying {
		return fmt.Errorf("tcc: confirm %s: %w", txID, ErrInvalidState)
	}
	r.state = StateConfirmed
	r.reserved = 0
	s.messages = append(s.messages, &Message{TxID: txID, Payload: "confirmed"})
	return nil
}

// Cancel releases a reservation, or writes a tombstone when Cancel arrives
// before Try. Repeat Cancel calls are no-ops.
func (s *Store) Cancel(txID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.txs[txID]
	if !ok {
		s.txs[txID] = &record{state: StateCancelled}
		s.messages = append(s.messages, &Message{TxID: txID, Payload: "cancelled-before-try"})
		return nil
	}
	switch r.state {
	case StateCancelled:
		return nil
	case StateConfirmed:
		return fmt.Errorf("tcc: cancel %s: %w", txID, ErrInvalidState)
	}
	s.balance += r.reserved
	r.reserved = 0
	r.state = StateCancelled
	s.messages = append(s.messages, &Message{TxID: txID, Payload: "cancelled"})
	return nil
}

// Relay publishes unsent messages, as a relay worker would, and reports how
// many rows became visible to downstream consumers.
func (s *Store) Relay() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	sent := 0
	for _, m := range s.messages {
		if !m.Sent {
			m.Sent = true
			sent++
		}
	}
	return sent
}

// Query returns the transaction state, whether it exists, and whether its
// latest message has already been delivered downstream.
func (s *Store) Query(txID string) (state State, found, visible bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.txs[txID]
	if !ok {
		return "", false, false
	}
	for i := len(s.messages) - 1; i >= 0; i-- {
		if s.messages[i].TxID == txID {
			return r.state, true, s.messages[i].Sent
		}
	}
	return r.state, true, false
}

// Balance returns the available balance.
func (s *Store) Balance() int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.balance
}

$ go test ./internal/platform/txn/tcc/ -count=1 -v
=== RUN   TestTryConfirmHappyPath
--- PASS: TestTryConfirmHappyPath (0.00s)
=== RUN   TestEmptyRollbackTombstoneRejectsLateTry
--- PASS: TestEmptyRollbackTombstoneRejectsLateTry (0.00s)
=== RUN   TestIdempotentTryAndConfirm
--- PASS: TestIdempotentTryAndConfirm (0.00s)
=== RUN   TestCancelAfterConfirmIsRejected
--- PASS: TestCancelAfterConfirmIsRejected (0.00s)
=== RUN   TestIntermediateStateBecomesVisibleAfterRelay
--- PASS: TestIntermediateStateBecomesVisibleAfterRelay (0.00s)
=== RUN   TestInsufficientBalanceLeavesNoState
--- PASS: TestInsufficientBalanceLeavesNoState (0.00s)
=== RUN   TestConcurrentTryReservesOnce
--- PASS: TestConcurrentTryReservesOnce (0.00s)
PASS
ok  	archlab/internal/platform/txn/tcc	2.666s