From 9641edd5825c4d85233f62be83f0e524b194308e Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 15:07:51 +0800 Subject: [PATCH] feat: supervise gateway control plane sessions --- README.md | 5 +- docs/api/control-plane.md | 7 +- docs/design/architecture.md | 1 + docs/development/implementation-plan.md | 5 +- docs/requirements/completion-audit.md | 5 +- .../controlplane/session_supervisor.go | 154 ++++++++++++++++++ .../controlplane/session_supervisor_test.go | 141 ++++++++++++++++ 7 files changed, 309 insertions(+), 9 deletions(-) create mode 100644 internal/gateway/controlplane/session_supervisor.go create mode 100644 internal/gateway/controlplane/session_supervisor_test.go diff --git a/README.md b/README.md index 3464847..31c8359 100644 --- a/README.md +++ b/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、可信代理、严格请求解析和敏感信息最小化。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index a643d34..58f94c6 100644 --- a/docs/api/control-plane.md +++ b/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。构建在后台完成， diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 825dd1d..6df8135 100644 --- a/docs/design/architecture.md +++ b/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 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 089361e..c2349f5 100644 --- a/docs/development/implementation-plan.md +++ b/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 和分配安全余量的批次；已覆盖 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 41c11fa..f9a6424 100644 --- a/docs/requirements/completion-audit.md +++ b/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 diff --git a/internal/gateway/controlplane/session_supervisor.go b/internal/gateway/controlplane/session_supervisor.go new file mode 100644 index 0000000..1815547 --- /dev/null +++ b/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) diff --git a/internal/gateway/controlplane/session_supervisor_test.go b/internal/gateway/controlplane/session_supervisor_test.go new file mode 100644 index 0000000..9b67f0d --- /dev/null +++ b/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 +}