《MMO 服务器开发》3.1 协议与超时预算

// file: internal/rpc/frame.go
// tier: L1
// verify: go test ./internal/rpc/
package rpc

import (
	"encoding/binary"
	"errors"
	"fmt"
)

// aegis 固定头 11 字节，与卷一 / 卷二定稿一致：
// length(4, BE) | cmd(2, BE) | seq(4, BE) | flags(1)
// length = 7 + N（7 = cmd+seq+flags），帧总长 = 11 + N。
const (
	HeaderSize = 11
	MaxPayload = 1 << 20
)

var (
	ErrShortFrame = errors.New("rpc: short frame")
	ErrTooLarge   = errors.New("rpc: payload too large")
)

// Frame 与卷一 / 卷二同名同义，跨服 RPC 直接复用同一份帧。
type Frame struct {
	Cmd     uint16
	Seq     uint32
	Flags   uint8
	Payload []byte
}

// Encode 序列化一帧。
func (f Frame) Encode() ([]byte, error) {
	n := len(f.Payload)
	if n > MaxPayload {
		return nil, fmt.Errorf("%w: %d > %d", ErrTooLarge, n, MaxPayload)
	}
	buf := make([]byte, HeaderSize+n)
	binary.BigEndian.PutUint32(buf[0:4], uint32(7+n))
	binary.BigEndian.PutUint16(buf[4:6], f.Cmd)
	binary.BigEndian.PutUint32(buf[6:10], f.Seq)
	buf[10] = f.Flags
	copy(buf[HeaderSize:], f.Payload)
	return buf, nil
}

// Decode 反序列化一帧，并校验 length 与真实长度一致。
func Decode(raw []byte) (Frame, error) {
	if len(raw) < HeaderSize {
		return Frame{}, ErrShortFrame
	}
	length := binary.BigEndian.Uint32(raw[0:4])
	if int(length) != len(raw)-4 {
		return Frame{}, fmt.Errorf("rpc: length %d mismatch frame %d", length, len(raw)-4)
	}
	f := Frame{
		Cmd:   binary.BigEndian.Uint16(raw[4:6]),
		Seq:   binary.BigEndian.Uint32(raw[6:10]),
		Flags: raw[10],
	}
	if length > 7 {
		f.Payload = make([]byte, length-7)
		copy(f.Payload, raw[HeaderSize:])
	}
	return f, nil
}

// file: internal/rpc/budget.go
// tier: L1
// verify: go test ./internal/rpc/
package rpc

import (
	"fmt"
	"time"
)

// Hop 是调用链上的一跳。
type Hop struct {
	Name  string
	Share float64 // 这一跳在可用预算里的占比，所有跳之和应为 1。
}

// Budget 把客户端的总 deadline 拆成每一跳的 timeout。
// safety 是预留的链路开销比例（序列化、排队、网络），不参与分配。
type Budget struct {
	Total  time.Duration
	Hops   []Hop
	safety float64
}

func NewBudget(total time.Duration, safety float64, hops ...Hop) *Budget {
	return &Budget{Total: total, Hops: hops, safety: safety}
}

// Allocate 返回每一跳的 timeout，所有跳之和 = Total*(1-safety)。
func (b *Budget) Allocate() map[string]time.Duration {
	usable := time.Duration(float64(b.Total) * (1 - b.safety))
	out := make(map[string]time.Duration, len(b.Hops))
	for _, h := range b.Hops {
		out[h.Name] = time.Duration(float64(usable) * h.Share)
	}
	return out
}

// Remaining 计算「从 start 起已经花了 now-start，还剩多少给当前跳」。
// 返回 0 与 false 表示预算已耗尽，应当立刻放弃而不是继续往下调。
func (b *Budget) Remaining(start, now time.Time) (time.Duration, bool) {
	left := b.Total - now.Sub(start)
	if left <= 0 {
		return 0, false
	}
	return left, true
}

func (b *Budget) String() string {
	return fmt.Sprintf("budget(total=%s safety=%.2f hops=%d)", b.Total, b.safety, len(b.Hops))
}

// file: internal/rpc/live_timeout_test.go
// tier: L2
// verify: go test ./internal/rpc/ -run 'TestTimeoutLive|TestIdempotencyKeyLive' -v
package rpc

import (
	"context"
	"fmt"
	"net"
	"sync"
	"testing"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"
)

// startPing 在 127.0.0.1 的随机端口起一个真 gRPC 服务端。
func startPing(t *testing.T, h PingHandler) (addr string, stop func()) {
	t.Helper()
	lis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	s := grpc.NewServer(grpc.ForceServerCodec(rawCodec{}))
	s.RegisterService(&PingServiceDesc, h)
	go func() { _ = s.Serve(lis) }()
	return lis.Addr().String(), s.Stop
}

func dialPing(t *testing.T, addr string) *grpc.ClientConn {
	t.Helper()
	cc, err := grpc.NewClient(addr,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(grpc.ForceCodec(rawCodec{})),
	)
	if err != nil {
		t.Fatal(err)
	}
	return cc
}

func TestTimeoutLive(t *testing.T) {
	addr, stop := startPing(t, func(ctx context.Context, in []byte) ([]byte, error) {
		select {
		case <-time.After(300 * time.Millisecond): // 模拟慢下游
			f, _ := Decode(in)
			return Frame{Cmd: f.Cmd + 1, Seq: f.Seq, Payload: f.Payload}.Encode()
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	})
	defer stop()
	cc := dialPing(t, addr)
	defer cc.Close()

	req, _ := Frame{Cmd: 0x0E01, Seq: 7, Payload: []byte("dawn-realm|hello")}.Encode()

	// 预算 100ms < 下游 300ms：必须超时
	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	var out []byte
	err := cc.Invoke(ctx, "/aegis.Ping/Call", req, &out)
	cancel()
	if status.Code(err) != codes.DeadlineExceeded {
		t.Fatalf("want DeadlineExceeded, got %v", err)
	}
	t.Logf("预算 100ms：%s", status.Code(err))

	// 预算 800ms > 下游 300ms：必须成功
	ctx2, cancel2 := context.WithTimeout(context.Background(), 800*time.Millisecond)
	defer cancel2()
	out = nil
	if err := cc.Invoke(ctx2, "/aegis.Ping/Call", req, &out); err != nil {
		t.Fatalf("want success, got %v", err)
	}
	resp, _ := Decode(out)
	if resp.Cmd != 0x0E02 || resp.Seq != 7 {
		t.Fatalf("bad resp %+v", resp)
	}
}

func TestIdempotencyKeyLive(t *testing.T) {
	var mu sync.Mutex
	applied := 0
	store := map[string]int{} // 幂等键 -> 效果编号
	addr, stop := startPing(t, func(ctx context.Context, in []byte) ([]byte, error) {
		f, err := Decode(in)
		if err != nil {
			return nil, err
		}
		key := string(f.Payload)
		time.Sleep(200 * time.Millisecond) // 慢到超过客户端预算，但效果照常提交
		mu.Lock()
		if _, ok := store[key]; !ok {
			applied++
			store[key] = applied
		}
		v := store[key]
		mu.Unlock()
		return Frame{Cmd: f.Cmd + 1, Seq: f.Seq, Payload: []byte(fmt.Sprintf("applied=%d", v))}.Encode()
	})
	defer stop()
	cc := dialPing(t, addr)
	defer cc.Close()

	req, _ := Frame{Cmd: 0x0E01, Seq: 9, Payload: []byte("idem-cross-chat-9")}.Encode()

	// 第一次：预算 50ms，必然超时；但服务端其实已把效果提交了
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	_ = cc.Invoke(ctx, "/aegis.Ping/Call", req, new([]byte))
	cancel()

	// 客户端不知道成没成功，用同一幂等键重试
	ctx2, cancel2 := context.WithTimeout(context.Background(), 800*time.Millisecond)
	var out []byte
	if err := cc.Invoke(ctx2, "/aegis.Ping/Call", req, &out); err != nil {
		t.Fatalf("retry failed: %v", err)
	}
	cancel2()

	mu.Lock()
	n := applied
	mu.Unlock()
	if n != 1 {
		t.Fatalf("effect applied %d times, want 1", n)
	}
	t.Logf("幂等键生效：同一 key 重试后效果只应用了 %d 次", n)
}