《后端架构设计·高性能与高可用》9.2 回滚、兼容窗口与数据迁移

// file: internal/migrate/orderstatus.go
// tier: L1
// verify: go test ./internal/migrate/ -count=1 -v
package migrate

import (
	"errors"
	"fmt"
)

type Status string

const (
	StatusCreated           Status = "CREATED"
	StatusInventoryReserved Status = "INVENTORY_RESERVED"
	StatusPaid              Status = "PAID"
	StatusFulfilled         Status = "FULFILLED"
	StatusCancelled         Status = "CANCELLED"
	StatusRefunded          Status = "REFUNDED"
)
type Code int16
var (
	statusCodes = map[Status]Code{
		StatusCreated:           Code(10),
		StatusInventoryReserved: Code(20),
		StatusPaid:              Code(30),
		StatusFulfilled:         Code(40),
		StatusCancelled:         Code(50),
		StatusRefunded:          Code(60),
	}
	codeStatuses = map[Code]Status{
		Code(10): StatusCreated,
		Code(20): StatusInventoryReserved,
		Code(30): StatusPaid,
		Code(40): StatusFulfilled,
		Code(50): StatusCancelled,
		Code(60): StatusRefunded,
	}
)
var (
	ErrUnknownStatus = errors.New("unknown order status")
	ErrLegacyMissing = errors.New("legacy status column is empty")
)
func Encode(s Status) (Code, error) {
	c, ok := statusCodes[s]
	if !ok {
		return 0, fmt.Errorf("%w: %q", ErrUnknownStatus, s)
	}
	return c, nil
}
func Decode(c Code) (Status, error) {
	s, ok := codeStatuses[c]
	if !ok {
		return "", fmt.Errorf("%w: code %d", ErrUnknownStatus, c)
	}
	return s, nil
}
// Row 是迁移期间的一行；V2 为 nil 表示历史行尚未回填。
type Row struct {
	OrderID int64
	Legacy  string
	V2      *Code
}
// ReadNew 是新代码的读取路径：优先新列，缺失时回退旧列。
func ReadNew(r Row) (Status, error) {
	if r.V2 != nil {
		return Decode(*r.V2)
	}
	return parseLegacy(r)
}
// ReadOld 是旧代码的读取路径：只认旧列，删列后必然失败。
func ReadOld(r Row) (Status, error) {
	if r.Legacy == "" {
		return "", ErrLegacyMissing
	}
	return parseLegacy(r)
}
func parseLegacy(r Row) (Status, error) {
	s := Status(r.Legacy)
	if _, err := Encode(s); err != nil {
		return "", err
	}
	return s, nil
}
// WriteDual 是 dual write 阶段的唯一合法写入方式：两列必须同时写。
func WriteDual(orderID int64, s Status) (Row, error) {
	c, err := Encode(s)
	if err != nil {
		return Row{}, err
	}
	return Row{OrderID: orderID, Legacy: string(s), V2: &c}, nil
}
// BackfillReport 记录回填结果；Mismatched 非零时禁止进入 contract。
type BackfillReport struct {
	Migrated   int
	Mismatched int
}
func Backfill(rows []Row) (BackfillReport, error) {
	var rep BackfillReport
	for i := range rows {
		r := &rows[i]
		fromLegacy, err := Encode(Status(r.Legacy))
		if err != nil {
			return rep, fmt.Errorf("order %d: %w", r.OrderID, err)
		}
		if r.V2 == nil {
			c := fromLegacy
			r.V2 = &c
			rep.Migrated++
			continue
		}
		if *r.V2 != fromLegacy {
			rep.Mismatched++
		}
	}
	return rep, nil
}

// file: internal/migrate/guard.go
// tier: L1
// verify: go test ./internal/migrate/ -count=1 -v
package migrate

import (
	"errors"
	"fmt"
	"time"
)

type Phase string

const (
	PhaseExpanded   Phase = "expanded"
	PhaseDualWrite  Phase = "dual_write"
	PhaseBackfilled Phase = "backfilled"
	PhaseContracted Phase = "contracted"
)

var (
	ErrPhaseTooEarly      = errors.New("phase must be backfilled before contract")
	ErrBackfillIncomplete = errors.New("backfill is not complete")
	ErrMismatch           = errors.New("legacy and v2 values disagree")
	ErrOldReadersRemain   = errors.New("old code can still read the legacy column")
	ErrWindowOpen         = errors.New("compatibility window has not elapsed")
	ErrNoBackup           = errors.New("no verified backup before contract")
	ErrContractIsOneWay   = errors.New("contract is one-way; restore from backup")
	ErrRollbackForward    = errors.New("rollback target must be an earlier phase")
)

var phaseRanks = map[Phase]int{
	PhaseExpanded: 0, PhaseDualWrite: 1, PhaseBackfilled: 2, PhaseContracted: 3,
}

type ContractInput struct {
	Phase          Phase
	BackfillDone   bool
	Mismatches     int
	OldReaders     int
	WindowElapsed  time.Duration
	RequiredWindow time.Duration
	BackupVerified bool
}

func CheckContract(in ContractInput) error {
	if in.Phase != PhaseBackfilled {
		return fmt.Errorf("%w: current=%s", ErrPhaseTooEarly, in.Phase)
	}
	if !in.BackfillDone {
		return ErrBackfillIncomplete
	}
	if in.Mismatches > 0 {
		return fmt.Errorf("%w: rows=%d", ErrMismatch, in.Mismatches)
	}
	if in.OldReaders > 0 {
		return fmt.Errorf("%w: readers=%d", ErrOldReadersRemain, in.OldReaders)
	}
	if in.WindowElapsed < in.RequiredWindow {
		return fmt.Errorf("%w: elapsed=%s required=%s", ErrWindowOpen, in.WindowElapsed, in.RequiredWindow)
	}
	if !in.BackupVerified {
		return ErrNoBackup
	}
	return nil
}

// CheckRollback 允许回到更早阶段，但 contract 之后只能从备份恢复。
func CheckRollback(current, target Phase) error {
	if current == PhaseContracted {
		return ErrContractIsOneWay
	}
	if phaseRanks[target] >= phaseRanks[current] {
		return ErrRollbackForward
	}
	return nil
}

// file: internal/migrate/migrate_test.go
// tier: L1
// verify: go test ./internal/migrate/ -count=1 -v
package migrate

import (
	"errors"
	"testing"
	"time"
)

func TestReadCompatibility(t *testing.T) {
	for _, s := range []Status{StatusCreated, StatusInventoryReserved, StatusPaid, StatusFulfilled, StatusCancelled, StatusRefunded} {
		if got, err := ReadNew(Row{OrderID: 1, Legacy: string(s)}); err != nil || got != s {
			t.Fatalf("ReadNew(%s)=(%s,%v)", s, got, err)
		}
		row, err := WriteDual(10, s)
		if err != nil {
			t.Fatal(err)
		}
		if row.V2 == nil {
			t.Fatalf("%s: dual write left V2 nil", s)
		}
		if got, err := ReadOld(row); err != nil || got != s {
			t.Fatalf("rollback read %s=(%s,%v)", s, got, err)
		}
	}
	row, _ := WriteDual(100, StatusPaid)
	old, err := ReadOld(row)
	if err != nil {
		t.Fatal(err)
	}
	t.Logf("dual_write legacy=%q v2=%v old_reader=%s", row.Legacy, *row.V2, old)
}
func TestBackfillAndMismatch(t *testing.T) {
	rows := []Row{{OrderID: 1, Legacy: string(StatusCreated)}, {OrderID: 2, Legacy: string(StatusPaid)}}
	rep, err := Backfill(rows)
	if err != nil || rep.Migrated != 2 || rep.Mismatched != 0 {
		t.Fatalf("backfill rep=%+v err=%v", rep, err)
	}
	migrated := rep.Migrated
	bad, _ := Encode(StatusCancelled)
	rows[0].V2 = &bad
	if rep, err = Backfill(rows); err != nil || rep.Mismatched != 1 {
		t.Fatalf("mismatch rep=%+v err=%v", rep, err)
	}
	t.Logf("backfill migrated=%d mismatched=%d", migrated, rep.Mismatched)
}

func TestContractBlockedUntilAllGuardsPass(t *testing.T) {
	good := ContractInput{Phase: PhaseBackfilled, BackfillDone: true, OldReaders: 0,
		WindowElapsed: 7 * 24 * time.Hour, RequiredWindow: 7 * 24 * time.Hour, BackupVerified: true}
	if err := CheckContract(good); err != nil {
		t.Fatalf("good contract rejected: %v", err)
	}
	cases := []struct {
		name string
		mut  func(*ContractInput)
		want error
	}{
		{"phase", func(in *ContractInput) { in.Phase = PhaseDualWrite }, ErrPhaseTooEarly},
		{"backfill", func(in *ContractInput) { in.BackfillDone = false }, ErrBackfillIncomplete},
		{"mismatch", func(in *ContractInput) { in.Mismatches = 3 }, ErrMismatch},
		{"readers", func(in *ContractInput) { in.OldReaders = 1 }, ErrOldReadersRemain},
		{"window", func(in *ContractInput) { in.WindowElapsed = 6 * 24 * time.Hour }, ErrWindowOpen},
		{"backup", func(in *ContractInput) { in.BackupVerified = false }, ErrNoBackup},
	}
	for _, tc := range cases {
		in := good
		tc.mut(&in)
		if err := CheckContract(in); !errors.Is(err, tc.want) {
			t.Fatalf("%s: err=%v want %v", tc.name, err, tc.want)
		}
	}
}

func TestRollbackAfterContractIsOneWay(t *testing.T) {
	paid, _ := Encode(StatusPaid)
	row := Row{OrderID: 9, V2: &paid}
	if _, err := ReadOld(row); !errors.Is(err, ErrLegacyMissing) {
		t.Fatalf("old code after contract err=%v", err)
	}
	if err := CheckRollback(PhaseContracted, PhaseDualWrite); !errors.Is(err, ErrContractIsOneWay) {
		t.Fatalf("rollback after contract err=%v", err)
	}
	if err := CheckRollback(PhaseBackfilled, PhaseDualWrite); err != nil {
		t.Fatalf("rollback before contract err=%v", err)
	}
	if err := CheckRollback(PhaseDualWrite, PhaseBackfilled); !errors.Is(err, ErrRollbackForward) {
		t.Fatalf("forward rollback err=%v", err)
	}
	t.Log("rollback_point=redeploy_old_code_legacy_column_authoritative")
}

$ go test -v ./... -count=1
=== RUN   TestReadCompatibility
    migrate_test.go:30: dual_write legacy="PAID" v2=30 old_reader=PAID
--- PASS: TestReadCompatibility (0.00s)
=== RUN   TestBackfillAndMismatch
    migrate_test.go:44: backfill migrated=2 mismatched=1
--- PASS: TestBackfillAndMismatch (0.00s)
=== RUN   TestContractBlockedUntilAllGuardsPass
--- PASS: TestContractBlockedUntilAllGuardsPass (0.00s)
=== RUN   TestRollbackAfterContractIsOneWay
    migrate_test.go:89: rollback_point=redeploy_old_code_legacy_column_authoritative
--- PASS: TestRollbackAfterContractIsOneWay (0.00s)
PASS
ok  	archlab/internal/migrate	0.236s