《互联网产品的服务端架构·产品、流量与基础能力》10.3 数据兼容与双写窗口

// file: internal/release/schema.go
// tier: L1
// verify: go test ./internal/release/

package release

import "fmt"

// Phase 是列变更所处的阶段。顺序必须是 expand、migrate、contract。
type Phase string

const (
	PhaseExpand   Phase = "expand"
	PhaseMigrate  Phase = "migrate"
	PhaseContract Phase = "contract"
)

// ColumnChange 是一次列变更。Drop 表示删除旧列。
type ColumnChange struct {
	Table      string
	Column     string
	Nullable   bool
	HasDefault bool
	Drop       bool
}

// Check 检查某个阶段是否允许执行这次列变更。
func Check(phase Phase, c ColumnChange) error {
	if c.Table == "" || c.Column == "" {
		return fmt.Errorf("release: column change needs table and column")
	}
	switch phase {
	case PhaseExpand:
		if c.Drop {
			return fmt.Errorf("release: expand phase cannot drop %s.%s", c.Table, c.Column)
		}
		if !c.Nullable && !c.HasDefault {
			return fmt.Errorf("release: expand phase needs nullable or default for %s.%s",
				c.Table, c.Column)
		}
	case PhaseMigrate:
		if c.Drop {
			return fmt.Errorf("release: migrate phase must keep old column %s.%s", c.Table, c.Column)
		}
	case PhaseContract:
		if !c.Drop {
			return fmt.Errorf("release: contract phase must drop old column %s.%s", c.Table, c.Column)
		}
	default:
		return fmt.Errorf("release: unknown migration phase %q", phase)
	}
	return nil
}

// ValidatePhases 检查一次列变更的阶段计划。
// 可以停在 migrate，但一旦进入 contract，必须已经过 migrate。
func ValidatePhases(phases []Phase) error {
	if len(phases) == 0 {
		return fmt.Errorf("release: empty migration phase plan")
	}
	seen := make(map[Phase]bool, len(phases))
	for i, p := range phases {
		switch p {
		case PhaseExpand, PhaseMigrate, PhaseContract:
		default:
			return fmt.Errorf("release: unknown migration phase %q", p)
		}
		if seen[p] {
			return fmt.Errorf("release: phase %q appears twice", p)
		}
		seen[p] = true
		if i == 0 && p != PhaseExpand {
			return fmt.Errorf("release: phase plan must start with expand, got %q", p)
		}
	}
	if !seen[PhaseExpand] {
		return fmt.Errorf("release: phase plan is missing expand")
	}
	if seen[PhaseContract] {
		if !seen[PhaseMigrate] {
			return fmt.Errorf("release: contract phase skips migrate")
		}
		if indexOf(phases, PhaseMigrate) > indexOf(phases, PhaseContract) {
			return fmt.Errorf("release: contract phase must follow migrate")
		}
	}
	return nil
}

func indexOf(phases []Phase, want Phase) int {
	for i, p := range phases {
		if p == want {
			return i
		}
	}
	return -1
}

// DualWrite 描述新旧两张表的双写窗口。Since 是窗口开始的 Unix 秒。
type DualWrite struct {
	OldTable string
	NewTable string
	Since    int64
}

// Idempotent 返回 applies 次重复执行后真正生效的逻辑写入次数。
// 双写依赖业务主键去重，同一变更重放多少次，两张表都只保留一份。
func Idempotent(w DualWrite, applies int) (int, error) {
	if applies < 0 {
		return 0, fmt.Errorf("release: applies must be non-negative, got %d", applies)
	}
	if w.OldTable == "" || w.NewTable == "" {
		return 0, fmt.Errorf("release: dual write needs both table names")
	}
	if w.OldTable == w.NewTable {
		return 0, fmt.Errorf("release: dual write tables must differ")
	}
	if w.Since < 0 {
		return 0, fmt.Errorf("release: dual write start time must be non-negative")
	}
	if applies == 0 {
		return 0, nil
	}
	return 1, nil
}

// file: internal/release/schema_test.go
// tier: L1
// verify: go test ./internal/release/

package release

import "testing"

func TestCheckExpand(t *testing.T) {
	base := ColumnChange{Table: "orders", Column: "channel"}
	cases := []struct {
		name string
		c    ColumnChange
		want bool
	}{
		{name: "nullable", c: base, want: true},
		{name: "default", c: ColumnChange{Table: "orders", Column: "channel", HasDefault: true}, want: true},
		{name: "not nullable without default", c: ColumnChange{Table: "orders", Column: "channel"}, want: false},
		{name: "drop in expand", c: ColumnChange{Table: "orders", Column: "channel", Drop: true}, want: false},
		{name: "missing column", c: ColumnChange{Table: "orders"}, want: false},
	}
	// nullable case needs Nullable=true
	cases[0].c.Nullable = true
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := Check(PhaseExpand, tc.c)
			if tc.want && err != nil {
				t.Fatalf("Check() = %v, want nil", err)
			}
			if !tc.want && err == nil {
				t.Fatal("Check() = nil, want error")
			}
		})
	}
}

func TestCheckMigrateAndContract(t *testing.T) {
	old := ColumnChange{Table: "orders", Column: "channel_legacy"}
	if err := Check(PhaseMigrate, old); err != nil {
		t.Fatalf("migrate with old column = %v, want nil", err)
	}
	if err := Check(PhaseMigrate, ColumnChange{Table: "orders", Column: "channel_legacy", Drop: true}); err == nil {
		t.Fatal("migrate drop = nil, want error")
	}
	if err := Check(PhaseContract, ColumnChange{Table: "orders", Column: "channel_legacy", Drop: true}); err != nil {
		t.Fatalf("contract drop = %v, want nil", err)
	}
	if err := Check(PhaseContract, old); err == nil {
		t.Fatal("contract without drop = nil, want error")
	}
	if err := Check("revert", old); err == nil {
		t.Fatal("unknown phase = nil, want error")
	}
}

func TestValidatePhases(t *testing.T) {
	ok := [][]Phase{
		{PhaseExpand},
		{PhaseExpand, PhaseMigrate},
		{PhaseExpand, PhaseMigrate, PhaseContract},
	}
	for _, phases := range ok {
		if err := ValidatePhases(phases); err != nil {
			t.Fatalf("ValidatePhases(%v) = %v, want nil", phases, err)
		}
	}
	bad := [][]Phase{
		{},
		{PhaseMigrate},
		{PhaseExpand, PhaseContract},
		{PhaseExpand, PhaseMigrate, PhaseMigrate},
		{PhaseExpand, PhaseContract, PhaseMigrate},
	}
	for _, phases := range bad {
		if err := ValidatePhases(phases); err == nil {
			t.Fatalf("ValidatePhases(%v) = nil, want error", phases)
		}
	}
}

func TestIdempotentDeduplicatesReplays(t *testing.T) {
	w := DualWrite{OldTable: "orders", NewTable: "orders_v2", Since: 1760000000}
	if got, err := Idempotent(w, 0); err != nil || got != 0 {
		t.Fatalf("Idempotent(0) = %d, %v, want 0, nil", got, err)
	}
	for _, applies := range []int{1, 2, 99} {
		got, err := Idempotent(w, applies)
		if err != nil {
			t.Fatal(err)
		}
		if got != 1 {
			t.Fatalf("Idempotent(%d) = %d, want 1 logical write", applies, got)
		}
	}
	if _, err := Idempotent(w, -1); err == nil {
		t.Fatal("Idempotent(-1) = nil error, want error")
	}
	if _, err := Idempotent(DualWrite{OldTable: "orders", NewTable: "orders", Since: 1}, 1); err == nil {
		t.Fatal("same table = nil error, want error")
	}
	if _, err := Idempotent(DualWrite{OldTable: "", NewTable: "orders_v2", Since: 1}, 1); err == nil {
		t.Fatal("empty table = nil error, want error")
	}
}

cd .local/checks/webapp-lab && go test ./internal/release/ -count=1 -run 'TestCheckExpand|TestCheckMigrateAndContract|TestValidatePhases|TestIdempotentDeduplicatesReplays' -v

=== RUN   TestCheckExpand
=== RUN   TestCheckExpand/nullable
=== RUN   TestCheckExpand/default
=== RUN   TestCheckExpand/not_nullable_without_default
=== RUN   TestCheckExpand/drop_in_expand
=== RUN   TestCheckExpand/missing_column
--- PASS: TestCheckExpand (0.00s)
    --- PASS: TestCheckExpand/nullable (0.00s)
    --- PASS: TestCheckExpand/default (0.00s)
    --- PASS: TestCheckExpand/not_nullable_without_default (0.00s)
    --- PASS: TestCheckExpand/drop_in_expand (0.00s)
    --- PASS: TestCheckExpand/missing_column (0.00s)
=== RUN   TestCheckMigrateAndContract
--- PASS: TestCheckMigrateAndContract (0.00s)
=== RUN   TestValidatePhases
--- PASS: TestValidatePhases (0.00s)
=== RUN   TestIdempotentDeduplicatesReplays
--- PASS: TestIdempotentDeduplicatesReplays (0.00s)
PASS
ok  	webapp/internal/release	0.223s