《后端架构设计·分布式与数据一致性》7.1 2PC 与 3PC：准备、提交与协调者故障

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

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

// Decision is the coordinator's durable outcome for a transaction.
type Decision string

const (
	DecisionPending Decision = "pending"
	DecisionCommit  Decision = "commit"
	DecisionAbort   Decision = "abort"
)

// ErrInvalidState reports an operation that contradicts the journal.
var ErrInvalidState = errors.New("twopc: invalid state")

// Journal is an append-only participant log. It keeps the last phase per
// transaction; a real implementation fsyncs before replying.
type Journal struct {
	mu   sync.Mutex
	last map[string]string
}

func (j *Journal) append(txID, phase string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.last == nil {
		j.last = make(map[string]string)
	}
	j.last[txID] = phase
}

// Last returns the last phase recorded for txID.
func (j *Journal) Last(txID string) (string, bool) {
	j.mu.Lock()
	defer j.mu.Unlock()
	phase, ok := j.last[txID]
	return phase, ok
}

type txState struct {
	reserved int64
	applied  bool
}

// Resource is one participant's lockable balance in cents.
type Resource struct {
	mu      sync.Mutex
	balance int64
	state   map[string]*txState
	journal Journal
}

// NewResource creates a participant resource with the given balance.
func NewResource(balance int64) *Resource {
	return &Resource{balance: balance, state: make(map[string]*txState)}
}

// Prepare reserves amount and journals "prepared". Repeat calls are no-ops.
func (r *Resource) Prepare(txID string, amount int64) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	if st, ok := r.state[txID]; ok {
		if st.applied || (st.reserved > 0 && st.reserved == amount) {
			return nil
		}
		if st.reserved > 0 {
			return fmt.Errorf("twopc: prepare %s: %w", txID, ErrInvalidState)
		}
	}
	if amount <= 0 {
		return fmt.Errorf("twopc: prepare %s: amount must be positive", txID)
	}
	available := r.balance
	for _, st := range r.state {
		available -= st.reserved
	}
	if available < amount {
		return fmt.Errorf("twopc: prepare %s: available %d < %d", txID, available, amount)
	}
	r.state[txID] = &txState{reserved: amount}
	r.journal.append(txID, "prepared")
	return nil
}

// Commit applies a reservation and journals "committed". Repeat calls are no-ops.
func (r *Resource) Commit(txID string) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	st, ok := r.state[txID]
	if !ok || st.applied || st.reserved == 0 {
		if ok && st.applied {
			return nil
		}
		return fmt.Errorf("twopc: commit %s: %w", txID, ErrInvalidState)
	}
	r.balance -= st.reserved
	st.reserved = 0
	st.applied = true
	r.journal.append(txID, "committed")
	return nil
}

// Rollback releases a reservation and journals "rolledback". Repeat calls are no-ops.
func (r *Resource) Rollback(txID string) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	st, ok := r.state[txID]
	if ok && st.applied {
		return fmt.Errorf("twopc: rollback %s: already committed", txID)
	}
	delete(r.state, txID)
	if _, seen := r.journal.Last(txID); !seen {
		r.journal.append(txID, "rolledback")
	}
	return nil
}

// DecisionLog is the coordinator's durable log. It must survive the crash
// that happens after every participant has voted.
type DecisionLog struct {
	mu        sync.Mutex
	decisions map[string]Decision
}

// NewDecisionLog creates an empty durable decision log.
func NewDecisionLog() *DecisionLog {
	return &DecisionLog{decisions: make(map[string]Decision)}
}

// Record writes a terminal decision once; conflicting writes are rejected.
func (l *DecisionLog) Record(txID string, d Decision) error {
	l.mu.Lock()
	defer l.mu.Unlock()
	if old, ok := l.decisions[txID]; ok && old != d {
		return fmt.Errorf("twopc: decision %s: %w", txID, ErrInvalidState)
	}
	l.decisions[txID] = d
	return nil
}

// Lookup returns the recorded decision, if any.
func (l *DecisionLog) Lookup(txID string) (Decision, bool) {
	l.mu.Lock()
	defer l.mu.Unlock()
	d, ok := l.decisions[txID]
	return d, ok
}

// Leg is one participant's prepare input.
type Leg struct {
	Participant string
	Amount      int64
}

// Coordinator drives two-phase commit across named participants.
type Coordinator struct {
	Participants map[string]*Resource
	Log          *DecisionLog
}

// Execute runs one 2PC transaction. With crashAfterPrepare it returns after
// all votes but before recording a decision, simulating a coordinator crash.
func (c *Coordinator) Execute(txID string, legs []Leg, crashAfterPrepare bool) (Decision, error) {
	if c.Log == nil {
		return DecisionPending, errors.New("twopc: nil decision log")
	}
	prepared := make([]*Resource, 0, len(legs))
	for _, leg := range legs {
		r, ok := c.Participants[leg.Participant]
		if !ok {
			return DecisionPending, fmt.Errorf("twopc: unknown participant %q", leg.Participant)
		}
		if err := r.Prepare(txID, leg.Amount); err != nil {
			if recErr := c.Log.Record(txID, DecisionAbort); recErr != nil {
				return DecisionPending, recErr
			}
			for _, done := range prepared {
				if rbErr := done.Rollback(txID); rbErr != nil {
					return DecisionAbort, rbErr
				}
			}
			return DecisionAbort, err
		}
		prepared = append(prepared, r)
	}
	if crashAfterPrepare {
		return DecisionPending, nil
	}
	if err := c.Log.Record(txID, DecisionCommit); err != nil {
		return DecisionPending, err
	}
	for _, r := range prepared {
		if err := r.Commit(txID); err != nil {
			return DecisionCommit, err
		}
	}
	return DecisionCommit, nil
}

// Recover resolves prepared participants from the durable decision log.
// Without a decision a prepared participant stays in doubt and keeps its lock.
func Recover(log *DecisionLog, participants map[string]*Resource, txID string) map[string]string {
	out := make(map[string]string, len(participants))
	decision, decided := log.Lookup(txID)
	for name, r := range participants {
		phase, _ := r.journal.Last(txID)
		switch {
		case phase == "prepared" && decided && decision == DecisionCommit:
			_ = r.Commit(txID)
			out[name] = "committed"
		case phase == "prepared" && decided && decision == DecisionAbort:
			_ = r.Rollback(txID)
			out[name] = "rolledback"
		case phase == "prepared":
			out[name] = "in-doubt"
		default:
			out[name] = phase
		}
	}
	return out
}

$ go test ./internal/platform/txn/twopc/ -count=1 -v
=== RUN   TestHappyPathCommitsBothParticipants
--- PASS: TestHappyPathCommitsBothParticipants (0.00s)
=== RUN   TestAbortReleasesPreparedLocks
--- PASS: TestAbortReleasesPreparedLocks (0.00s)
=== RUN   TestCrashLeavesInDoubtUntilDecisionRecovered
--- PASS: TestCrashLeavesInDoubtUntilDecisionRecovered (0.00s)
=== RUN   TestRecoveryHonoursAbortDecision
--- PASS: TestRecoveryHonoursAbortDecision (0.00s)
=== RUN   TestParticipantOperationsAreIdempotent
--- PASS: TestParticipantOperationsAreIdempotent (0.00s)
PASS
ok  	archlab/internal/platform/txn/twopc	1.609s