《游戏服务端编程实践》21.3 Serverless 与 WASM 新范式

// file: ch21/wasmhost/wasmhost.go
// tier: L1
// verify: go test ./ch21/wasmhost/

// Package wasmhost 演示服务端插件宿主的最小契约：注册、调用、超时和
// panic 隔离。真实环境会把 Execute 的实现换成 WASM 模块实例化，但宿主侧
// 的错误语义、耗时统计和资源预算保持一致。
package wasmhost

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"
)

const (
	DefaultTimeout       = 50 * time.Millisecond
	DefaultMaxInputSize  = 64 << 10
	DefaultMaxOutputSize = 256 << 10
)

var (
	ErrEmptyName       = errors.New("wasmhost: empty plugin name")
	ErrDuplicatePlugin = errors.New("wasmhost: duplicate plugin")
	ErrPluginNotFound  = errors.New("wasmhost: plugin not found")
	ErrInputTooLarge   = errors.New("wasmhost: input too large")
	ErrOutputTooLarge  = errors.New("wasmhost: output too large")
	ErrTimeout         = errors.New("wasmhost: plugin timeout")
	ErrCanceled        = errors.New("wasmhost: call canceled")
	ErrPluginPanic     = errors.New("wasmhost: plugin panic")
	ErrPluginFailed    = errors.New("wasmhost: plugin failed")
)

// Plugin 是宿主唯一依赖的插件接口。Execute 必须尊重 ctx：宿主可以停止
// 等待，但无法强杀一个不检查 ctx 的 goroutine，WASM 运行时才有中断指令的能力。
type Plugin interface {
	Name() string
	Execute(ctx context.Context, input []byte) ([]byte, error)
}

// Limits 是宿主侧的资源预算。真实 WASM 运行时还应限制线性内存和指令数；
// 这里用输入、输出字节数和墙钟超时表达同一类边界。
type Limits struct {
	Timeout       time.Duration
	MaxInputSize  int
	MaxOutputSize int
}

func (l Limits) withDefaults() Limits {
	if l.Timeout <= 0 {
		l.Timeout = DefaultTimeout
	}
	if l.MaxInputSize <= 0 {
		l.MaxInputSize = DefaultMaxInputSize
	}
	if l.MaxOutputSize <= 0 {
		l.MaxOutputSize = DefaultMaxOutputSize
	}
	return l
}

// Result 记录一次调用的可观测信息。即使调用失败也返回，方便网关按插件
// 维度统计耗时、超时和 panic，而不是只看到一条错误日志。
type Result struct {
	Plugin   string
	Output   []byte
	Elapsed  time.Duration
	TimedOut bool
	Panicked bool
}

// Host 保存插件注册表。注册发生在启动阶段，调用发生在请求路径上，
// 读多写少，因此使用 RWMutex。
type Host struct {
	mu      sync.RWMutex
	plugins map[string]Plugin
	limits  Limits
}

func NewHost(limits Limits) *Host {
	return &Host{
		plugins: make(map[string]Plugin),
		limits:  limits.withDefaults(),
	}
}

// Register 拒绝空名称和重名插件。名称会去除首尾空白，避免 “echo” 和
// “ echo ” 被当成两个插件，导致路由行为依赖调用方拼写。
func (h *Host) Register(p Plugin) error {
	if p == nil {
		return fmt.Errorf("%w: <nil>", ErrEmptyName)
	}
	name := strings.TrimSpace(p.Name())
	if name == "" {
		return ErrEmptyName
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if _, ok := h.plugins[name]; ok {
		return fmt.Errorf("%w: %s", ErrDuplicatePlugin, name)
	}
	h.plugins[name] = p
	return nil
}

// PluginNames 返回排序后的插件清单，供启动日志和健康检查展示。
func (h *Host) PluginNames() []string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	names := make([]string, 0, len(h.plugins))
	for name := range h.plugins {
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

type outcome struct {
	output     []byte
	err        error
	panicValue any
}

// Call 在独立 goroutine 中执行插件，并同时等待结果、超时和父上下文取消。
// 通道带一个缓冲，插件在超时之后才返回也不会永久阻塞发送方。
func (h *Host) Call(ctx context.Context, name string, input []byte) (Result, error) {
	h.mu.RLock()
	plugin, ok := h.plugins[name]
	h.mu.RUnlock()
	base := Result{Plugin: name}
	if !ok {
		return base, fmt.Errorf("%w: %s", ErrPluginNotFound, name)
	}
	if len(input) > h.limits.MaxInputSize {
		return base, fmt.Errorf("%w: %s: %d > %d", ErrInputTooLarge, name,
			len(input), h.limits.MaxInputSize)
	}

	callCtx, cancel := context.WithTimeout(ctx, h.limits.Timeout)
	defer cancel()

	started := time.Now()
	done := make(chan outcome, 1)
	go func() {
		var o outcome
		defer func() {
			if r := recover(); r != nil {
				o = outcome{panicValue: r}
				done <- o
			}
		}()
		o.output, o.err = plugin.Execute(callCtx, input)
		done <- o
	}()

	select {
	case o := <-done:
		return h.finish(base, started, o)
	case <-callCtx.Done():
		// 插件可能恰好在截止时刻返回。先做一次非阻塞读取，避免把
		// 已经完成的结果误判成超时。
		select {
		case o := <-done:
			return h.finish(base, started, o)
		default:
		}
		base.Elapsed = time.Since(started)
		if errors.Is(callCtx.Err(), context.Canceled) {
			return base, fmt.Errorf("%w: %s", ErrCanceled, name)
		}
		base.TimedOut = true
		return base, fmt.Errorf("%w: %s after %s", ErrTimeout, name, h.limits.Timeout)
	}
}

func (h *Host) finish(base Result, started time.Time, o outcome) (Result, error) {
	base.Elapsed = time.Since(started)
	if o.panicValue != nil {
		base.Panicked = true
		return base, fmt.Errorf("%w: %s: %v", ErrPluginPanic, base.Plugin, o.panicValue)
	}
	if o.err != nil {
		return base, fmt.Errorf("%w: %s: %v", ErrPluginFailed, base.Plugin, o.err)
	}
	if len(o.output) > h.limits.MaxOutputSize {
		return base, fmt.Errorf("%w: %s: %d > %d", ErrOutputTooLarge, base.Plugin,
			len(o.output), h.limits.MaxOutputSize)
	}
	base.Output = o.output
	return base, nil
}

cd .local/checks/gs-practice-lab && gofmt -l ch21/wasmhost && go vet ./ch21/wasmhost/ && go test ./ch21/wasmhost/ -count=1

ok  	gspractice/ch21/wasmhost	0.287s