《后端架构设计·演进与治理》9.3 数据库变更与发布协同：expand-contract

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

import (
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// Phase is one step of an expand-contract migration.
type Phase string

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

// Op is a schema operation whose code compatibility is known up front.
type Op string

const (
	OpAddNullableColumn Op = "add_nullable_column"
	OpAddIndex          Op = "add_index"
	OpBackfill          Op = "backfill"
	OpSetNotNull        Op = "set_not_null"
	OpDropColumn        Op = "drop_column"
	OpRenameColumn      Op = "rename_column"
)

// Change is one schema operation tied to a phase.
type Change struct {
	ID     string
	Column string
	Phase  Phase
	Op     Op
}

// Plan is the ordered migration plan plus the release constraints.
type Plan struct {
	Changes       []Change
	MinAppVersion string
	QuietPeriod   time.Duration
}

// State is what the release controller can observe right now.
type State struct {
	Now            time.Time
	AppVersions    []string
	Backfilled     map[string]time.Time // backfill change ID -> completion
	StoppedWriting map[string]time.Time // column -> last old-code write
	DualWrite      bool
}

func phaseRank(p Phase) int {
	switch p {
	case PhaseExpand:
		return 0
	case PhaseMigrate:
		return 1
	case PhaseContract:
		return 2
	}
	return -1
}

func oldCodeSafe(op Op) bool {
	return op == OpAddNullableColumn || op == OpAddIndex || op == OpBackfill
}

// Check applies structural rules first, then the runtime gates: dual write,
// old readers, completed backfill and the quiet period after last old write.
func Check(p Plan, st State) []error {
	var errs []error
	if len(p.Changes) == 0 {
		return []error{errors.New("release: migration plan is empty")}
	}
	if p.MinAppVersion == "" {
		errs = append(errs, errors.New("release: plan has no minimum app version"))
	}
	if p.QuietPeriod <= 0 {
		errs = append(errs, errors.New("release: plan has no quiet period"))
	}
	if st.Now.IsZero() {
		errs = append(errs, errors.New("release: state has no clock"))
	}
	seen := map[string]bool{}
	rank := -1
	for i, ch := range p.Changes {
		where := fmt.Sprintf("change[%d] %s", i, ch.ID)
		if ch.ID == "" {
			errs = append(errs, fmt.Errorf("change[%d]: id is empty", i))
		} else if seen[ch.ID] {
			errs = append(errs, fmt.Errorf("%s: duplicate id", where))
		}
		seen[ch.ID] = true
		if ch.Column == "" {
			errs = append(errs, fmt.Errorf("%s: column is empty", where))
		}
		r := phaseRank(ch.Phase)
		if r < 0 {
			errs = append(errs, fmt.Errorf("%s: unknown phase %q", where, ch.Phase))
			continue
		}
		if r < rank {
			errs = append(errs, fmt.Errorf("%s: phase %s after a later phase", where, ch.Phase))
		}
		rank = r
		if ch.Op == OpRenameColumn {
			errs = append(errs, fmt.Errorf("%s: rename must be add+backfill+drop", where))
		}
		if ch.Phase == PhaseExpand && !oldCodeSafe(ch.Op) {
			errs = append(errs, fmt.Errorf("%s: %s is not safe for old code", where, ch.Op))
		}
	}
	for _, ch := range p.Changes {
		if ch.Phase == PhaseMigrate && !st.DualWrite {
			errs = append(errs, fmt.Errorf("%s: backfill needs dual write", ch.ID))
		}
		if ch.Phase != PhaseContract {
			continue
		}
		if !versionsAtLeast(st.AppVersions, p.MinAppVersion) {
			errs = append(errs, fmt.Errorf("%s: old code still serves: %v < %s", ch.ID, st.AppVersions, p.MinAppVersion))
		}
		if missing := missingBackfills(p, st); len(missing) > 0 {
			errs = append(errs, fmt.Errorf("%s: contract waits for backfill %v", ch.ID, missing))
		}
		stopped, ok := st.StoppedWriting[ch.Column]
		if !ok {
			errs = append(errs, fmt.Errorf("%s: no stop-writing record for %s", ch.ID, ch.Column))
		} else if quiet := st.Now.Sub(stopped); quiet < p.QuietPeriod {
			errs = append(errs, fmt.Errorf("%s: quiet %s < %s", ch.ID, quiet, p.QuietPeriod))
		}
	}
	return errs
}

func missingBackfills(p Plan, st State) []string {
	var missing []string
	for _, ch := range p.Changes {
		if ch.Phase != PhaseMigrate {
			continue
		}
		if _, ok := st.Backfilled[ch.ID]; !ok {
			missing = append(missing, ch.ID)
		}
	}
	return missing
}

func versionsAtLeast(have []string, want string) bool {
	for _, v := range have {
		if compareVersion(v, want) < 0 {
			return false
		}
	}
	return true
}

// compareVersion compares dotted versions numerically: v1.10.0 > v1.9.0.
func compareVersion(a, b string) int {
	as := strings.Split(strings.TrimPrefix(a, "v"), ".")
	bs := strings.Split(strings.TrimPrefix(b, "v"), ".")
	n := len(as)
	if len(bs) > n {
		n = len(bs)
	}
	for i := 0; i < n; i++ {
		x, y := 0, 0
		if i < len(as) {
			x, _ = strconv.Atoi(as[i])
		}
		if i < len(bs) {
			y, _ = strconv.Atoi(bs[i])
		}
		if x != y {
			if x < y {
				return -1
			}
			return 1
		}
	}
	return 0
}

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

import (
	"strings"
	"testing"
	"time"
)

func validPlan() Plan {
	return Plan{
		Changes: []Change{
			{ID: "E1", Column: "courier_id", Phase: PhaseExpand, Op: OpAddNullableColumn},
			{ID: "M1", Column: "courier_id", Phase: PhaseMigrate, Op: OpBackfill},
			{ID: "C1", Column: "legacy_status", Phase: PhaseContract, Op: OpDropColumn},
		},
		MinAppVersion: "v1.10.0",
		QuietPeriod:   72 * time.Hour,
	}
}

func readyState(now time.Time) State {
	return State{
		Now:            now,
		AppVersions:    []string{"v1.10.0", "v1.10.1"},
		Backfilled:     map[string]time.Time{"M1": now.Add(-80 * time.Hour)},
		StoppedWriting: map[string]time.Time{"legacy_status": now.Add(-73 * time.Hour)},
		DualWrite:      true,
	}
}

func TestValidPlanPasses(t *testing.T) {
	now := time.Date(2026, 10, 31, 8, 20, 0, 0, time.UTC)
	if errs := Check(validPlan(), readyState(now)); len(errs) != 0 {
		t.Fatalf("ready plan rejected: %v", errs)
	}
	t.Logf("changes=%d min_app=%s quiet=%s", len(validPlan().Changes), validPlan().MinAppVersion, validPlan().QuietPeriod)
	for _, tc := range []struct {
		a, b string
		want int
	}{{"v1.10.0", "v1.9.0", 1}, {"1.9.0", "v1.10.0", -1}} {
		if got := compareVersion(tc.a, tc.b); got != tc.want {
			t.Fatalf("compare(%s,%s)=%d", tc.a, tc.b, got)
		}
	}
}

func TestPlanValidationRejectsUnsafeOrderAndOps(t *testing.T) {
	cases := map[string]Plan{
		"contract_first": {Changes: []Change{
			{ID: "C1", Column: "a", Phase: PhaseContract, Op: OpDropColumn},
			{ID: "E1", Column: "b", Phase: PhaseExpand, Op: OpAddNullableColumn},
		}, MinAppVersion: "v1.0.0", QuietPeriod: time.Hour},
		"not_null_in_expand": {Changes: []Change{
			{ID: "E1", Column: "a", Phase: PhaseExpand, Op: OpSetNotNull},
		}, MinAppVersion: "v1.0.0", QuietPeriod: time.Hour},
		"rename": {Changes: []Change{
			{ID: "E1", Column: "a", Phase: PhaseExpand, Op: OpRenameColumn},
		}, MinAppVersion: "v1.0.0", QuietPeriod: time.Hour},
		"empty": {},
	}
	now := time.Date(2026, 10, 31, 8, 0, 0, 0, time.UTC)
	for name, p := range cases {
		if errs := Check(p, State{Now: now}); len(errs) == 0 {
			t.Fatalf("%s accepted", name)
		} else {
			t.Logf("%s: %v", name, errs[0])
		}
	}
}

func TestContractRuntimeGates(t *testing.T) {
	now := time.Date(2026, 10, 31, 8, 20, 0, 0, time.UTC)
	cases := []struct {
		name   string
		mutate func(*State)
		expect string
	}{
		{"old_app", func(s *State) { s.AppVersions = []string{"v1.9.0"} }, "old code still serves"},
		{"no_dual_write", func(s *State) { s.DualWrite = false }, "backfill needs dual write"},
		{"not_backfilled", func(s *State) { delete(s.Backfilled, "M1") }, "contract waits for backfill"},
		{"quiet_too_short", func(s *State) { s.StoppedWriting["legacy_status"] = now.Add(-71 * time.Hour) }, "quiet"},
		{"no_stop_record", func(s *State) { delete(s.StoppedWriting, "legacy_status") }, "no stop-writing record"},
	}
	for _, tc := range cases {
		st := readyState(now)
		tc.mutate(&st)
		errs := Check(validPlan(), st)
		found := false
		for _, err := range errs {
			found = found || strings.Contains(err.Error(), tc.expect)
		}
		if !found {
			t.Fatalf("%s: errors %v missing %q", tc.name, errs, tc.expect)
		}
		t.Logf("%s: %v", tc.name, errs[0])
	}
}

$ go test ./internal/platform/release/ -run 'TestValidPlanPasses|TestPlanValidationRejectsUnsafeOrderAndOps|TestContractRuntimeGates' -v -count=1
=== RUN   TestValidPlanPasses
    migrate_test.go:39: changes=3 min_app=v1.10.0 quiet=72h0m0s
--- PASS: TestValidPlanPasses (0.00s)
=== RUN   TestPlanValidationRejectsUnsafeOrderAndOps
    migrate_test.go:69: empty: release: migration plan is empty
    migrate_test.go:69: contract_first: change[1] E1: phase expand after a later phase
    migrate_test.go:69: not_null_in_expand: change[0] E1: set_not_null is not safe for old code
    migrate_test.go:69: rename: change[0] E1: rename must be add+backfill+drop
--- PASS: TestPlanValidationRejectsUnsafeOrderAndOps (0.00s)
=== RUN   TestContractRuntimeGates
    migrate_test.go:98: old_app: C1: old code still serves: [v1.9.0] < v1.10.0
    migrate_test.go:98: no_dual_write: M1: backfill needs dual write
    migrate_test.go:98: not_backfilled: C1: contract waits for backfill [M1]
    migrate_test.go:98: quiet_too_short: C1: quiet 71h0m0s < 72h0m0s
    migrate_test.go:98: no_stop_record: C1: no stop-writing record for legacy_status
--- PASS: TestContractRuntimeGates (0.00s)
PASS
ok  	archlab/internal/platform/release	0.274s