《后端架构设计·分布式与数据一致性》12.3 容灾切换：RTO/RPO、演练与回切

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

import (
	"fmt"
	"sort"
	"time"
)

// Phase is one stage of a controlled switchover.
type Phase string

const (
	// Normal serves reads and writes from the original primary.
	Normal Phase = "NORMAL"
	// ReadOnly freezes writes at the original primary.
	ReadOnly Phase = "READ_ONLY"
	// Promoted has promoted the replica but not opened client traffic.
	Promoted Phase = "PROMOTED"
	// Serving sends clients to the new primary.
	Serving Phase = "SERVING"
)

var allowed = map[Phase][]Phase{
	Normal:   {ReadOnly},
	ReadOnly: {Normal, Promoted},
	Promoted: {Serving},
	Serving:  {},
}

// Switch tracks the phase and the facts that decide whether a rollback is
// safe. It does not perform I/O; the runbook drives it.
type Switch struct {
	phase              Phase
	writesOnNewPrimary int64
	reverseSynced      bool
}

// NewSwitch returns a switch in Normal.
func NewSwitch() *Switch {
	return &Switch{phase: Normal}
}

// Phase returns the current phase.
func (s *Switch) Phase() Phase {
	return s.phase
}

// Apply moves to next if the transition is allowed.
func (s *Switch) Apply(next Phase) error {
	for _, candidate := range allowed[s.phase] {
		if candidate == next {
			s.phase = next
			return nil
		}
	}
	return fmt.Errorf("dr: transition %s -> %s not allowed", s.phase, next)
}

// RecordWrite counts a write accepted by the new primary. It is only valid
// after promotion, when the new primary can accept client writes.
func (s *Switch) RecordWrite() error {
	if s.phase != Promoted && s.phase != Serving {
		return fmt.Errorf("dr: write recorded in phase %s", s.phase)
	}
	s.writesOnNewPrimary++
	return nil
}

// MarkReverseSynced records that writes accepted by the new primary were
// copied back to the original primary.
func (s *Switch) MarkReverseSynced() {
	s.reverseSynced = true
}

// WritesOnNewPrimary returns the number of writes accepted after promotion.
func (s *Switch) WritesOnNewPrimary() int64 {
	return s.writesOnNewPrimary
}

// RollbackAllowed reports whether traffic may return to the original primary.
// Before promotion a rollback is just leaving read-only mode. After promotion
// every write accepted by the new primary must exist on the old one, either
// because there were none or because reverse replication completed.
func (s *Switch) RollbackAllowed() error {
	switch s.phase {
	case ReadOnly:
		return nil
	case Promoted, Serving:
		if s.writesOnNewPrimary == 0 || s.reverseSynced {
			return nil
		}
		return fmt.Errorf("dr: rollback blocked by %d unsynced writes on new primary", s.writesOnNewPrimary)
	default:
		return fmt.Errorf("dr: no rollback path from phase %s", s.phase)
	}
}

// AckPoint is the last write acknowledged to a client together with the
// replica position observed at that moment.
type AckPoint struct {
	AckedSeq   uint64
	AckedAt    time.Time
	ReplicaSeq uint64
	ReplicaAt  time.Time
}

// RPO returns how many acknowledged transactions the replica may be missing
// and the width of the wall-clock window they were acknowledged in.
func (a AckPoint) RPO() (uint64, time.Duration, error) {
	if a.ReplicaSeq > a.AckedSeq {
		return 0, 0, fmt.Errorf("dr: replica seq %d ahead of acked seq %d", a.ReplicaSeq, a.AckedSeq)
	}
	window := a.AckedAt.Sub(a.ReplicaAt)
	if window < 0 {
		return 0, 0, fmt.Errorf("dr: replica timestamp ahead by %s, check clock skew", -window)
	}
	return a.AckedSeq - a.ReplicaSeq, window, nil
}

// Step is one runbook action with an estimated duration.
type Step struct {
	Name     string
	Duration time.Duration
	After    []string
}

// Plan is a dependency graph of switchover steps.
type Plan struct {
	Steps []Step
}

// RTO returns the earliest duration in which every step can finish and the
// critical path that determines it.
func (p Plan) RTO() (time.Duration, []string, error) {
	if len(p.Steps) == 0 {
		return 0, nil, fmt.Errorf("dr: empty plan")
	}
	steps := make(map[string]Step, len(p.Steps))
	for _, step := range p.Steps {
		if step.Name == "" {
			return 0, nil, fmt.Errorf("dr: step with empty name")
		}
		if _, dup := steps[step.Name]; dup {
			return 0, nil, fmt.Errorf("dr: duplicate step %q", step.Name)
		}
		if step.Duration < 0 {
			return 0, nil, fmt.Errorf("dr: step %q has negative duration", step.Name)
		}
		steps[step.Name] = step
	}

	indegree := make(map[string]int, len(steps))
	successors := make(map[string][]string, len(steps))
	for name := range steps {
		indegree[name] = 0
	}
	for _, step := range p.Steps {
		for _, dep := range step.After {
			if _, ok := steps[dep]; !ok {
				return 0, nil, fmt.Errorf("dr: step %q depends on unknown %q", step.Name, dep)
			}
			successors[dep] = append(successors[dep], step.Name)
			indegree[step.Name]++
		}
	}
	for name := range successors {
		sort.Strings(successors[name])
	}

	finish := make(map[string]time.Duration, len(steps))
	start := make(map[string]time.Duration, len(steps))
	previous := make(map[string]string, len(steps))
	ready := make([]string, 0, len(steps))
	for name := range steps {
		if indegree[name] == 0 {
			ready = append(ready, name)
		}
	}
	sort.Strings(ready)

	done := 0
	for len(ready) > 0 {
		name := ready[0]
		ready = ready[1:]
		finish[name] = start[name] + steps[name].Duration
		done++
		for _, next := range successors[name] {
			if finish[name] > start[next] {
				start[next] = finish[name]
				previous[next] = name
			}
			indegree[next]--
			if indegree[next] == 0 {
				ready = append(ready, next)
				sort.Strings(ready)
			}
		}
	}
	if done != len(steps) {
		return 0, nil, fmt.Errorf("dr: dependency cycle across %d steps", len(steps)-done)
	}

	names := make([]string, 0, len(steps))
	for name := range steps {
		names = append(names, name)
	}
	sort.Strings(names)
	last := ""
	var total time.Duration
	for _, name := range names {
		if last == "" || finish[name] > total {
			last, total = name, finish[name]
		}
	}
	path := []string{last}
	for {
		parent, ok := previous[last]
		if !ok {
			break
		}
		path = append(path, parent)
		last = parent
	}
	for i, j := 0, len(path)-1; i < j; i, j = i+1, j-1 {
		path[i], path[j] = path[j], path[i]
	}
	return total, path, nil
}

$ go test ./internal/platform/dr/ -v -count=1
=== RUN   TestSwitchoverPhases
    switchover_test.go:17: rejected NORMAL -> SERVING: dr: transition NORMAL -> SERVING not allowed
    switchover_test.go:24: phase=READ_ONLY
    switchover_test.go:24: phase=PROMOTED
    switchover_test.go:24: phase=SERVING
=== RUN   TestRollbackRequiresReverseSync
    switchover_test.go:55: rollback_blocked writes=2 err=dr: rollback blocked by 2 unsynced writes on new primary
    switchover_test.go:60: rollback_allowed writes=2 reverse_synced=true
=== RUN   TestRPO
    switchover_test.go:78: rpo_transactions=4 window=700ms
    switchover_test.go:85: replica_ahead rejected: dr: replica seq 1050 ahead of acked seq 1042
    switchover_test.go:93: clock_skew rejected: dr: replica timestamp ahead by 300ms, check clock skew
=== RUN   TestPlanRTO
    switchover_test.go:122: rto=1m30s critical_path=[detect notify freeze dns warm open]
=== RUN   TestPlanRejectsInvalidGraphs
    switchover_test.go:141: unknown_dep rejected: dr: step "a" depends on unknown "missing"
=== RUN   TestRecordWriteBeforePromotion
    switchover_test.go:150: write_in_normal rejected: dr: write recorded in phase NORMAL
PASS
ok  	archlab/internal/platform/dr	0.258s

#!/bin/sh
## Switchover drill template. Replace *_ctl with the real control-plane CLI.
set -eu
PRIMARY=${PRIMARY:?}; REPLICA=${REPLICA:?}; TARGET_RTO=${TARGET_RTO:-180}
start=$(date +%s); echo "start=$(date -u +%FT%TZ)" >> timeline
frozen=0
trap 'echo "abort at $(date -u +%FT%TZ) frozen=$frozen" >> timeline' EXIT
seq_ctl capture "$PRIMARY" > pre.seq          ## stage 0: RPO baseline
write_ctl freeze "$PRIMARY" && frozen=1       ## stage 1: stop writes
write_ctl assert_rejected "$PRIMARY"          ## gate: no write may pass
replica_ctl promote "$REPLICA"                ## stage 2: promote
replica_ctl assert_writable "$REPLICA"
dns_ctl switch "$REPLICA"                     ## stage 3: traffic switch
dns_ctl assert_all_resolvers "$REPLICA"
app_ctl verify orders,payment,inventory       ## stage 4: business checks
elapsed=$(( $(date +%s) - start ))
[ "$elapsed" -le "$TARGET_RTO" ] || echo "RTO overrun: ${elapsed}s" >> timeline
echo "elapsed=${elapsed}s rpo=$(seq_ctl gap pre.seq "$REPLICA")" >> timeline