feat: supervise gateway control plane sessions

README.md

@@ -42,8 +42,9 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
   审计与 Outbox；不保存 Proxy 明细或逐次提取记录。
 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、
   目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/
-  Runtime 会话组件已有实现与定向测试。Controller 可向 Worker 下发无凭据引用的
-  已归属 Proxy 快照；`proxy-gateway` 命令、凭据分发和 Routing Snapshot 发布仍待装配。
+  Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行
+  有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 快照；
+  `proxy-gateway` 命令、凭据分发和 Routing Snapshot 发布仍待装配。
 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持
   CIDR、可信代理、严格请求解析和敏感信息最小化。
 

docs/api/control-plane.md

@@ -93,9 +93,10 @@ Gateway 接收完整快照时必须拒绝缺失、格式错误或已到期的 `v
 发起新的上游连接，也不查询 Redis 或 PostgreSQL 补偿。
 
 Controller 只会下发尚未到期的完整快照，并在最近一次成功下发快照的
-`valid_until` 到达时结束 `WatchSnapshots` 流。Gateway 的会话调用方必须在流结束后
-按带 jitter 的有界退避重建 Register/Watch 会话；在新快照通过校验并原子替换前，
-旧视图仍按其整体有效期 fail-closed。
+`valid_until` 到达时结束 `WatchSnapshots` 流。Gateway 的 `SessionSupervisor` 会在流
+结束或可恢复控制面错误后按带 jitter 的有界退避重建 Register/Watch 会话；参数、认证
+和协议不兼容错误直接返回。新快照通过校验并原子替换前，旧视图仍按其整体有效期
+fail-closed。
 
 Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 checksum
 验证成功时才能应用；否则丢弃 Delta 并请求完整 Snapshot。构建在后台完成，

docs/design/architecture.md

@@ -57,6 +57,7 @@ Provider、Pool、Routing、Distribution 在首版需要共享事务和一致性
 - 首条命中 Routing、Upstream 选择和 Proxy least-connections 选择。
 - 本地原子容量预留、建连、Active 计数和结果上报。
 - 本地不可变 Snapshot；热路径无数据库/Redis/Provider API。
+- 独立的控制面 SessionSupervisor；快照流到期或中断后以有界退避重建会话。
 - SSRF 与 DNS Rebinding 防护。
 
 ### 3.2 proxy-controller

docs/development/implementation-plan.md

@@ -215,8 +215,9 @@ Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话
 按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本，并可构建无凭据引用的
 已归属 Proxy payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成，
 重注册会清除旧引用，避免迟到 Stream 覆盖新 session。Worker 服务端会在最近完整
-Snapshot 的 `valid_until` 到达时结束流，令 Gateway 调用方可按退避策略重建会话；
-Routing payload、凭据分发、Gateway 命令与 Outcome 上报仍未实现。
+Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已为 Gateway 调用方
+提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Routing payload、
+凭据分发、Gateway 命令与 Outcome 上报仍未实现。
 
 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider
 Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖

docs/requirements/completion-audit.md

@@ -54,8 +54,9 @@
   索引构建无凭据引用的已归属 Proxy payload，并以租约收紧可用期；Routing payload、
   凭据分发、Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
   `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller
-  在最近成功下发的 Snapshot `valid_until` 到达时关闭流，Gateway 会校验并执行
-  Snapshot 整体 `valid_until`，过期视图不再分配新 Proxy。
+  在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用
+  `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot
+  整体 `valid_until`，过期视图不再分配新 Proxy。
 - `PostgreSQL 管理面`：已定义 `adminstate` 事务 seam、并发安全 MemoryStore、
   公用契约、100 并发 Routing CAS、租约 Outbox 和只含六张管理表的 Schema；pgx
   Adapter 已在真实 PostgreSQL 18 上通过同一契约、迁移幂等、审计/Outbox

internal/gateway/controlplane/session_supervisor.go

@@ -0,0 +1,154 @@
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var ErrInvalidReconnectOptions = errors.New("invalid gateway reconnect options")
+
+// SessionLifecycle is the long-running worker control-plane session boundary.
+// SessionRunner satisfies this interface.
+type SessionLifecycle interface {
+	Run(context.Context) error
+}
+
+// ReconnectOptions controls bounded exponential reconnect delays. Jitter is a
+// percentage applied symmetrically to each base delay.
+type ReconnectOptions struct {
+	InitialDelay time.Duration
+	MaxDelay     time.Duration
+	Jitter       int
+}
+
+// SessionSupervisorRuntime makes reconnect timing deterministic in tests.
+// Nil fields use the production timer and random source.
+type SessionSupervisorRuntime struct {
+	Sleeper SessionSleeper
+	Random  SessionRandom
+}
+
+type SessionSleeper interface {
+	Sleep(context.Context, time.Duration) error
+}
+
+type SessionRandom interface {
+	Float64() float64
+}
+
+// SessionSupervisor owns retry policy around repeated worker sessions without
+// coupling it to SnapshotWatcher or RuntimeReporter.
+type SessionSupervisor struct {
+	lifecycle SessionLifecycle
+	options   ReconnectOptions
+	runtime   SessionSupervisorRuntime
+}
+
+func NewSessionSupervisor(lifecycle SessionLifecycle, options ReconnectOptions, runtimes ...SessionSupervisorRuntime) (*SessionSupervisor, error) {
+	if lifecycle == nil || options.InitialDelay <= 0 || options.MaxDelay < options.InitialDelay || options.Jitter < 0 || options.Jitter > 100 || len(runtimes) > 1 {
+		return nil, ErrInvalidReconnectOptions
+	}
+	runtime := SessionSupervisorRuntime{Sleeper: timerSessionSleeper{}, Random: globalSessionRandom{}}
+	if len(runtimes) == 1 {
+		if runtimes[0].Sleeper != nil {
+			runtime.Sleeper = runtimes[0].Sleeper
+		}
+		if runtimes[0].Random != nil {
+			runtime.Random = runtimes[0].Random
+		}
+	}
+	return &SessionSupervisor{lifecycle: lifecycle, options: options, runtime: runtime}, nil
+}
+
+// Run restarts recoverable sessions until the context is canceled. A session
+// that ends without an error is treated as a closed snapshot stream.
+func (supervisor *SessionSupervisor) Run(ctx context.Context) error {
+	if supervisor == nil || ctx == nil {
+		return ErrInvalidReconnectOptions
+	}
+	for failedAttempts := uint(0); ; failedAttempts++ {
+		err := supervisor.lifecycle.Run(ctx)
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		if err == nil {
+			err = ErrSnapshotStreamClosed
+		}
+		if !retryableSessionError(err) {
+			return err
+		}
+		if err := supervisor.runtime.Sleeper.Sleep(ctx, supervisor.retryDelay(failedAttempts)); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return err
+		}
+	}
+}
+
+func (supervisor *SessionSupervisor) retryDelay(failedAttempts uint) time.Duration {
+	delay := supervisor.options.InitialDelay
+	for attempt := uint(0); attempt < failedAttempts && delay < supervisor.options.MaxDelay; attempt++ {
+		if delay >= supervisor.options.MaxDelay/2 {
+			delay = supervisor.options.MaxDelay
+			break
+		}
+		delay *= 2
+	}
+	if supervisor.options.Jitter == 0 {
+		return delay
+	}
+	random := supervisor.runtime.Random.Float64()
+	if random < 0 {
+		random = 0
+	} else if random > 1 {
+		random = 1
+	}
+	spread := float64(supervisor.options.Jitter) / 100
+	delay = time.Duration(float64(delay) * (1 + (2*random-1)*spread))
+	if delay < time.Millisecond {
+		delay = time.Millisecond
+	}
+	if delay > supervisor.options.MaxDelay {
+		return supervisor.options.MaxDelay
+	}
+	return delay
+}
+
+func retryableSessionError(err error) bool {
+	if errors.Is(err, ErrInvalidOptions) {
+		return false
+	}
+	switch status.Code(err) {
+	case codes.InvalidArgument, codes.PermissionDenied, codes.Unauthenticated, codes.Unimplemented:
+		return false
+	default:
+		return true
+	}
+}
+
+type timerSessionSleeper struct{}
+
+func (timerSessionSleeper) Sleep(ctx context.Context, delay time.Duration) error {
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+type globalSessionRandom struct{}
+
+func (globalSessionRandom) Float64() float64 {
+	return rand.Float64()
+}
+
+var _ SessionLifecycle = (*SessionRunner)(nil)

internal/gateway/controlplane/session_supervisor_test.go

@@ -0,0 +1,141 @@
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func TestSessionSupervisorRetriesRecoverableFailuresWithCappedBackoff(t *testing.T) {
+	runner := &sequenceSessionLifecycle{results: []error{
+		ErrSnapshotStreamClosed,
+		errors.New("temporary control plane failure"),
+		status.Error(codes.PermissionDenied, "worker identity is not authorized"),
+	}}
+	sleeper := &recordingSessionSleeper{}
+	supervisor, err := NewSessionSupervisor(runner, ReconnectOptions{
+		InitialDelay: 100 * time.Millisecond,
+		MaxDelay:     150 * time.Millisecond,
+	}, SessionSupervisorRuntime{Sleeper: sleeper, Random: fixedSessionRandom(0.5)})
+	if err != nil {
+		t.Fatalf("NewSessionSupervisor(): %v", err)
+	}
+
+	err = supervisor.Run(context.Background())
+	if status.Code(err) != codes.PermissionDenied {
+		t.Fatalf("Run() code = %s, want PermissionDenied; error=%v", status.Code(err), err)
+	}
+	if runner.calls != 3 {
+		t.Fatalf("Run() calls = %d, want 3", runner.calls)
+	}
+	if got, want := sleeper.delays, []time.Duration{100 * time.Millisecond, 150 * time.Millisecond}; !equalSessionDelays(got, want) {
+		t.Fatalf("backoff delays = %v, want %v", got, want)
+	}
+}
+
+func TestSessionSupervisorStopsWhenContextCancelsDuringBackoff(t *testing.T) {
+	runner := &sequenceSessionLifecycle{results: []error{ErrSnapshotStreamClosed}}
+	sleeper := blockingSessionSleeper{started: make(chan struct{})}
+	supervisor, err := NewSessionSupervisor(runner, ReconnectOptions{
+		InitialDelay: time.Second,
+		MaxDelay:     time.Second,
+	}, SessionSupervisorRuntime{Sleeper: sleeper, Random: fixedSessionRandom(0.5)})
+	if err != nil {
+		t.Fatalf("NewSessionSupervisor(): %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	result := make(chan error, 1)
+	go func() { result <- supervisor.Run(ctx) }()
+	select {
+	case <-sleeper.started:
+	case <-time.After(time.Second):
+		t.Fatal("reconnect backoff did not start")
+	}
+	cancel()
+	if err := <-result; !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want context canceled", err)
+	}
+	if runner.calls != 1 {
+		t.Fatalf("Run() calls = %d, want 1", runner.calls)
+	}
+}
+
+func TestSessionSupervisorRejectsInvalidOptionsAndDoesNotRetryPermanentErrors(t *testing.T) {
+	if _, err := NewSessionSupervisor(&sequenceSessionLifecycle{}, ReconnectOptions{}); !errors.Is(err, ErrInvalidReconnectOptions) {
+		t.Fatalf("NewSessionSupervisor(invalid options) error = %v, want ErrInvalidReconnectOptions", err)
+	}
+
+	runner := &sequenceSessionLifecycle{results: []error{status.Error(codes.InvalidArgument, "invalid worker")}}
+	sleeper := &recordingSessionSleeper{}
+	supervisor, err := NewSessionSupervisor(runner, ReconnectOptions{
+		InitialDelay: time.Millisecond,
+		MaxDelay:     time.Millisecond,
+	}, SessionSupervisorRuntime{Sleeper: sleeper, Random: fixedSessionRandom(0.5)})
+	if err != nil {
+		t.Fatalf("NewSessionSupervisor(): %v", err)
+	}
+	if err := supervisor.Run(context.Background()); status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("Run() code = %s, want InvalidArgument; error=%v", status.Code(err), err)
+	}
+	if runner.calls != 1 || len(sleeper.delays) != 0 {
+		t.Fatalf("permanent error calls=%d delays=%v, want one call and no delay", runner.calls, sleeper.delays)
+	}
+}
+
+type sequenceSessionLifecycle struct {
+	results []error
+	calls   int
+}
+
+func (runner *sequenceSessionLifecycle) Run(context.Context) error {
+	runner.calls++
+	if len(runner.results) == 0 {
+		return ErrSnapshotStreamClosed
+	}
+	result := runner.results[0]
+	runner.results = runner.results[1:]
+	return result
+}
+
+type recordingSessionSleeper struct {
+	delays []time.Duration
+}
+
+func (sleeper *recordingSessionSleeper) Sleep(_ context.Context, delay time.Duration) error {
+	sleeper.delays = append(sleeper.delays, delay)
+	return nil
+}
+
+type blockingSessionSleeper struct {
+	started chan struct{}
+}
+
+func (sleeper blockingSessionSleeper) Sleep(ctx context.Context, _ time.Duration) error {
+	close(sleeper.started)
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+type fixedSessionRandom float64
+
+func (random fixedSessionRandom) Float64() float64 {
+	return float64(random)
+}
+
+func equalSessionDelays(left, right []time.Duration) bool {
+	if len(left) != len(right) {
+		return false
+	}
+	for index := range left {
+		if left[index] != right[index] {
+			return false
+		}
+	}
+	return true
+}