《MMO 服务器开发》4.3 优雅停服与有状态服务发布

// file: internal/deploy/shutdown.go
// tier: L2
// verify: go test ./internal/deploy/ -run TestGracefulShutdown -v
// 4.3 的优雅停服：SIGTERM 到退出，次序固定——停收新连接、排空在途请求、落
// 库、返回退 0。跳过排空会丢飞到一半的请求，跳过落库会丢最后一次提交。
package deploy

import (
	"bufio"
	"fmt"
	"io"
	"net"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// Server 是极小的 TCP 节点：每连接处理一个请求——读一行、干活、写 OK、记账。
type Server struct {
	ln           net.Listener
	inFlight     atomic.Int64
	drainTimeout time.Duration
	mu           sync.Mutex
	ledger       []string
}

// NewServer 返回一个排空最多等 drainTimeout 的 Server。
func NewServer(drainTimeout time.Duration) *Server {
	return &Server{drainTimeout: drainTimeout}
}

// Listen 在 addr 上开始接受连接，返回实际绑定的地址。
func (s *Server) Listen(addr string) (string, error) {
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return "", err
	}
	s.ln = ln
	go s.accept()
	return ln.Addr().String(), nil
}

func (s *Server) accept() {
	for {
		c, err := s.ln.Accept()
		if err != nil {
			return // 监听已被 Shutdown 关掉
		}
		go s.serve(c)
	}
}

func (s *Server) serve(c net.Conn) {
	defer c.Close()
	s.inFlight.Add(1)
	defer s.inFlight.Add(-1)
	line, err := bufio.NewReader(c).ReadString('\n')
	if err != nil {
		return
	}
	time.Sleep(50 * time.Millisecond) // 代替真实的请求处理
	s.mu.Lock()
	s.ledger = append(s.ledger, strings.TrimRight(line, "\r\n"))
	s.mu.Unlock()
	io.WriteString(c, "OK\n")
}

// Shutdown 执行四步排空，返回落库了多少条记录。
func (s *Server) Shutdown(flushPath string) (int, error) {
	if s.ln != nil {
		s.ln.Close() // 1. 停收新连接
	}
	deadline := time.Now().Add(s.drainTimeout) // 2. 排空在途请求
	for s.inFlight.Load() > 0 {
		if time.Now().After(deadline) {
			return 0, fmt.Errorf("drain timed out with %d in flight", s.inFlight.Load())
		}
		time.Sleep(2 * time.Millisecond)
	}
	s.mu.Lock() // 3. 落库
	records := append([]string(nil), s.ledger...)
	s.mu.Unlock()
	f, err := os.Create(flushPath)
	if err != nil {
		return 0, err
	}
	defer f.Close()
	w := bufio.NewWriter(f)
	for _, r := range records {
		fmt.Fprintln(w, r)
	}
	if err := w.Flush(); err != nil {
		return 0, err
	}
	return len(records), nil // 4. 返回，调用方退 0
}

// file: internal/deploy/shutdown_test.go
// tier: L2
// verify: go test ./internal/deploy/ -run TestGracefulShutdown -v
package deploy

import (
	"bufio"
	"fmt"
	"io"
	"net"
	"os"
	"os/exec"
	"os/signal"
	"strings"
	"syscall"
	"testing"
	"time"
)

// TestHelperProcess 不是真测试：AEGIS_HELPER=1 时在子进程里跑服务器，好让父
// 进程发一个真 SIGTERM。
func TestHelperProcess(t *testing.T) {
	if os.Getenv("AEGIS_HELPER") != "1" {
		return
	}
	srv := NewServer(2 * time.Second)
	addr, err := srv.Listen("127.0.0.1:0")
	if err != nil {
		fmt.Println("ERR listen:", err)
		os.Exit(2)
	}
	fmt.Println("READY", addr)
	os.Stdout.Sync()
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGTERM)
	<-sig
	flushed, err := srv.Shutdown(os.Getenv("AEGIS_FLUSH"))
	if err != nil {
		fmt.Println("ERR shutdown:", err)
		os.Exit(3)
	}
	fmt.Printf("SHUTDOWN flushed=%d\n", flushed)
	os.Exit(0)
}

func TestGracefulShutdown(t *testing.T) {
	flushPath := t.TempDir() + "/ledger.txt"
	cmd := exec.Command(os.Args[0], "-test.run=TestHelperProcess")
	cmd.Env = append(os.Environ(), "AEGIS_HELPER=1", "AEGIS_FLUSH="+flushPath)
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		t.Fatal(err)
	}
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	sc := bufio.NewScanner(stdout)
	if !sc.Scan() || !strings.HasPrefix(sc.Text(), "READY ") {
		t.Fatalf("no READY line, got %q", sc.Text())
	}
	addr := strings.TrimSpace(strings.TrimPrefix(sc.Text(), "READY "))

	// 8 条连接各发一个请求；handler 睡 50ms，发信号时它们都在途。
	const n = 8
	conns := make([]net.Conn, 0, n)
	for i := 0; i < n; i++ {
		c, err := net.Dial("tcp", addr)
		if err != nil {
			t.Fatalf("dial %d: %v", i, err)
		}
		fmt.Fprintf(c, "req-%d\n", i)
		conns = append(conns, c)
	}
	time.Sleep(10 * time.Millisecond)
	if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
		t.Fatal(err)
	}

	// 在途请求必须全部拿到响应：是排空，不是丢弃。
	for i, c := range conns {
		c.SetReadDeadline(time.Now().Add(3 * time.Second))
		buf := make([]byte, 3)
		if _, err := io.ReadFull(c, buf); err != nil || string(buf) != "OK\n" {
			t.Errorf("conn %d not drained: resp=%q err=%v", i, buf, err)
		}
		c.Close()
	}
	// 监听已关：新连接必须被拒。
	if !eventuallyRefused(addr) {
		t.Error("listener still accepted a connection after SIGTERM")
	}
	// 子进程干净退出，并报出落库条数。
	var line string
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "SHUTDOWN ") {
			line = sc.Text()
		}
	}
	if err := cmd.Wait(); err != nil {
		t.Fatalf("child exited with error: %v", err)
	}
	if want := fmt.Sprintf("SHUTDOWN flushed=%d", n); line != want {
		t.Fatalf("child said %q, want %q", line, want)
	}
	// 落库真的写到磁盘，一条不少。
	data, err := os.ReadFile(flushPath)
	if err != nil {
		t.Fatal(err)
	}
	if got := strings.Count(string(data), "\n"); got != n {
		t.Errorf("ledger has %d records, want %d", got, n)
	}
}

func eventuallyRefused(addr string) bool {
	for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
		c, err := net.DialTimeout("tcp", addr, 100*time.Millisecond)
		if err != nil {
			return true
		}
		c.Close()
		time.Sleep(20 * time.Millisecond)
	}
	return false
}

=== RUN   TestGracefulShutdown
--- PASS: TestGracefulShutdown (0.12s)
ok  	aegis/internal/deploy	0.377s

// file: internal/deploy/epoch.go
// tier: L1
// verify: go test ./internal/deploy/ -run TestEpoch
// 4.3 的有状态半边：发布一个持单写者状态的节点（一个场景实例、一个世界分
// 片），不只是「起个新 Pod」。新主必须证明自己比它接管的状态更新，否则两个
// 主会同时服务同一个实例——脑裂。
//
// 每个有状态实例带一个代次 epoch：单调递增的世代号。只有当来者的 epoch 严格
// 大于状态上戳的 epoch，交接才被允许。
package deploy

// Epoch 是有状态实例的世代号。
type Epoch uint64

// CanTakeOver 报告 epoch 为 e 的主能否接管戳着 current 的状态。相等的 epoch
// 也被拒——两个主处在同一代，正是要防的脑裂。
func (e Epoch) CanTakeOver(current Epoch) bool { return e > current }

// Next 返回新主该采用的 epoch，永远严格大于它接管的状态。
func (e Epoch) Next() Epoch { return e + 1 }

// Handover 是一次有状态发布决策的结果。
type Handover struct {
	From    Epoch
	To      Epoch
	Allowed bool
	Reason  string
}

// PlanHandover 判定 epoch 为 to 的主能否接管戳着 from 的状态。
func PlanHandover(from, to Epoch) Handover {
	if to.CanTakeOver(from) {
		return Handover{From: from, To: to, Allowed: true, Reason: "newer generation"}
	}
	return Handover{From: from, To: to, Allowed: false, Reason: "epoch not newer: refusing to split brain"}
}