diff --git a/README.md b/README.md index 336f221..37d988d 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,8 @@ Proxy Pool 用 Controller 协调这些变化,并让 Gateway 数据面只消费 View 匹配请求,并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待, 以及仍经过目标地址策略的 direct;`proxy-gateway` 通过独立控制面拨号地址维护 - Session,并仅在持有未过期 Snapshot 时 Ready;凭据材料只保留在当前节点内存 View。 + Session,并在每份 Snapshot 有效期的一半前接收版本递增的完整刷新;仅在持有未过期 + Snapshot 时 Ready,凭据材料只保留在当前节点内存 View。 - **安全边界**:Gateway、Distribution 与 Admin 使用各自的认证语义,并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index b06d2f4..0fb1b51 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -25,7 +25,9 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和 监听允许明文 fixture 模式。单消息大小、并发流数和 gRPC keepalive 策略由 `controlPlane` 配置限定。 -`WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照, +`WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照,并在 +每份快照有效期的一半前重新构建、下发版本递增的完整快照;这样 ownership、Routing 和 +凭据变化会在同一长连接内收敛,而无需等待有效期到达后重新注册。 Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership 索引构建已归属 Proxy 内容,并将租约到期收紧到 Proxy 的 `usable_until`。Proxy 引用的 凭据材料按 `secret_ref + credential_version` 去重,随完整 Snapshot 经 mTLS 下发,仅保留在 @@ -108,11 +110,10 @@ Gateway 接收完整快照时必须拒绝缺失、格式错误或已到期的 `v 保存在本地不可变视图。该整体期限到达后,调度直接按无候选处理,不再使用旧视图 发起新的上游连接,也不查询 Redis 或 PostgreSQL 补偿。 -Controller 只会下发尚未到期的完整快照,并在最近一次成功下发快照的 -`valid_until` 到达时结束 `WatchSnapshots` 流。Gateway 的 `SessionSupervisor` 会在流 -结束或可恢复控制面错误后按带 jitter 的有界退避重建 Register/Watch 会话;参数、认证 -和协议不兼容错误直接返回。新快照通过校验并原子替换前,旧视图仍按其整体有效期 -fail-closed。 +Controller 只会下发尚未到期的完整快照,并在每份快照有效期的一半前发送下一版完整 +快照。刷新构建失败、流结束或可恢复控制面错误时,Gateway 的 `SessionSupervisor` 会按 +带 jitter 的有界退避重建 Register/Watch 会话;参数、认证和协议不兼容错误直接返回。 +新快照通过校验并原子替换前,旧视图仍按其整体有效期 fail-closed。 Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 checksum 验证成功时才能应用;否则丢弃 Delta 并请求完整 Snapshot。构建在后台完成, diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 168cca5..a2629c3 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -22,7 +22,7 @@ go run ./cmd/proxy-controller -config CONFIG_FILE PostgreSQL 管理面迁移、Redis 活动池、Distribution/Admin 独立监听与优雅停机; Controller Metrics 独立监听、`/livez`、`/readyz` 和基础 Prometheus 运行时指标; Provider 自动补池、分布式配额、动态重载和 Admin 低基数统计已装配。Gateway 进程、 -Worker 控制面会话、Snapshot 凭据分发和 Snapshot 就绪探针已装配;Checker 的任务流与 +Worker 控制面会话、持续完整 Snapshot 刷新、凭据分发和 Snapshot 就绪探针已装配;Checker 的任务流与 租约契约已接入控制面,但 Redis 共享调度、独立执行器与代表性负载验证仍在后续实施范围。 所有时间值使用 Go duration,例如 `500ms`、`30s`、`5m`。示例中的 diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 62f4e3d..0e6e377 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -57,7 +57,8 @@ Provider、Pool、Routing、Distribution 在首版需要共享事务和一致性 - 首条命中 Routing、Upstream 选择和 Proxy least-connections 选择。 - 本地原子容量预留、建连、Active 计数和结果上报。 - 本地不可变 Snapshot;热路径无数据库/Redis/Provider API。 -- 独立的控制面 SessionSupervisor;快照流到期或中断后以有界退避重建会话。 +- 独立的控制面 SessionSupervisor;Controller 在有效期一半前连续下发版本递增的完整 + 快照,流中断后才以有界退避重建会话。 - SSRF 与 DNS Rebinding 防护。 ### 3.2 proxy-controller diff --git a/findings.md b/findings.md index d7f5865..f5c45b3 100644 --- a/findings.md +++ b/findings.md @@ -114,6 +114,16 @@ Routing 自上而下匹配,首条命中停止;支持 Gateway 与 Extract 两 维护该索引。Memory 单测、Redis 集成契约和配置/Reaper 单测覆盖恢复、阈值、延后与 Drain/ACK 后清理。 +## Worker Snapshot 刷新(2026-08-02) + +- `OwnedSnapshotSource` 每次调用只构建一份权威完整快照;新增 + `RefreshingSnapshotSource` 以有效期一半为上限持续调用该窄接口,并传递最新版本和 + checksum。短于全局 `maxStaleAge` 的 Proxy/ownership 租约会进一步缩短下一次刷新间隔。 +- gRPC Handler 每次成功发送完整 Snapshot 都重置服务端到期计时器;流中刷新失败或中断时 + Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot + 消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报,但 Controller 尚未具备 + 自动 `BeginDrain`/`AcknowledgeDrain` 编排或 Worker 分配器,不能视为自动 Drain 闭环。 + ## Git 同步事实(2026-07-29) - PostgreSQL 管理面基础文档已提交为 `7951c29`。 diff --git a/internal/controller/worker/server.go b/internal/controller/worker/server.go index 19b7793..0eb6375 100644 --- a/internal/controller/worker/server.go +++ b/internal/controller/worker/server.go @@ -60,9 +60,13 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server } } if snapshots == nil { - snapshots, err = NewInitialSnapshotSource(service, controlPlane.MaxStaleAge.Value(), time.Now) + initial, initialErr := NewInitialSnapshotSource(service, controlPlane.MaxStaleAge.Value(), time.Now) + if initialErr != nil { + return nil, fmt.Errorf("%w: build initial snapshot source: %v", ErrInvalidServer, initialErr) + } + snapshots, err = NewRefreshingSnapshotSource(initial, snapshotRefreshEvery(controlPlane.MaxStaleAge.Value())) if err != nil { - return nil, fmt.Errorf("%w: build initial snapshot source: %v", ErrInvalidServer, err) + return nil, fmt.Errorf("%w: build refreshing snapshot source: %v", ErrInvalidServer, err) } } serverOptions = append(serverOptions, diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index 586ada6..aed3818 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -156,7 +156,11 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err if err != nil { return nil, errors.Join(ErrInvalidCommand, err) } - result.snapshots = source + refreshed, refreshErr := NewRefreshingSnapshotSource(source, snapshotRefreshEvery(options.MaxStaleAge)) + if refreshErr != nil { + return nil, errors.Join(ErrInvalidCommand, refreshErr) + } + result.snapshots = refreshed } return result, nil } diff --git a/internal/controller/worker/snapshot_source.go b/internal/controller/worker/snapshot_source.go index 20712b9..f820afe 100644 --- a/internal/controller/worker/snapshot_source.go +++ b/internal/controller/worker/snapshot_source.go @@ -25,6 +25,113 @@ type SnapshotSource interface { Watch(context.Context, SnapshotWatchRequest) (<-chan *controlplanev1.WorkerSnapshot, error) } +// RefreshingSnapshotSource keeps a Worker snapshot stream current without +// requiring the Gateway to reconnect after every validity window. It only +// forwards complete snapshots, so Gateway delta handling remains fail-closed. +type RefreshingSnapshotSource struct { + source SnapshotSource + refreshEvery time.Duration +} + +func NewRefreshingSnapshotSource(source SnapshotSource, refreshEvery time.Duration) (*RefreshingSnapshotSource, error) { + if source == nil || refreshEvery <= 0 { + return nil, ErrSnapshotsUnavailable + } + return &RefreshingSnapshotSource{source: source, refreshEvery: refreshEvery}, nil +} + +func snapshotRefreshEvery(validFor time.Duration) time.Duration { + if validFor <= 0 { + return 0 + } + if half := validFor / 2; half > 0 { + return half + } + return validFor +} + +func (source *RefreshingSnapshotSource) Watch( + ctx context.Context, + request SnapshotWatchRequest, +) (<-chan *controlplanev1.WorkerSnapshot, error) { + if source == nil || source.source == nil || ctx == nil { + return nil, ErrSnapshotsUnavailable + } + first, err := source.next(ctx, request) + if err != nil { + return nil, err + } + updates := make(chan *controlplanev1.WorkerSnapshot, 1) + updates <- first + go source.refresh(ctx, updates, request, first) + return updates, nil +} + +func (source *RefreshingSnapshotSource) refresh( + ctx context.Context, + updates chan<- *controlplanev1.WorkerSnapshot, + request SnapshotWatchRequest, + current *controlplanev1.WorkerSnapshot, +) { + defer close(updates) + version := current.GetVersion() + checksum := append([]byte(nil), current.GetChecksum()...) + timer := time.NewTimer(source.delayFor(current)) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + request.LastAppliedVersion = version + request.LastChecksum = append(request.LastChecksum[:0], checksum...) + next, err := source.next(ctx, request) + if err != nil { + return + } + select { + case <-ctx.Done(): + return + case updates <- next: + version = next.GetVersion() + checksum = append(checksum[:0], next.GetChecksum()...) + timer.Reset(source.delayFor(next)) + } + } + } +} + +func (source *RefreshingSnapshotSource) next( + ctx context.Context, + request SnapshotWatchRequest, +) (*controlplanev1.WorkerSnapshot, error) { + stream, err := source.source.Watch(ctx, request) + if err != nil { + return nil, err + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case snapshot, ok := <-stream: + if !ok || snapshot == nil || snapshot.GetVersion() <= request.LastAppliedVersion { + return nil, ErrSnapshotsUnavailable + } + return snapshot, nil + } +} + +func (source *RefreshingSnapshotSource) delayFor(snapshot *controlplanev1.WorkerSnapshot) time.Duration { + delay := source.refreshEvery + if snapshot == nil || snapshot.GetGeneratedAt() == nil || snapshot.GetValidUntil() == nil { + return delay + } + lifetime := snapshot.GetValidUntil().AsTime().Sub(snapshot.GetGeneratedAt().AsTime()) + if half := lifetime / 2; half > 0 && half < delay { + return half + } + return delay +} + type OwnershipEpochReader interface { CurrentOwnershipEpoch(context.Context) (uint64, error) } @@ -73,3 +180,4 @@ func (source *InitialSnapshotSource) Watch(ctx context.Context, request Snapshot } var _ SnapshotSource = (*InitialSnapshotSource)(nil) +var _ SnapshotSource = (*RefreshingSnapshotSource)(nil) diff --git a/internal/controller/worker/snapshot_source_test.go b/internal/controller/worker/snapshot_source_test.go index 1b92840..fccd79c 100644 --- a/internal/controller/worker/snapshot_source_test.go +++ b/internal/controller/worker/snapshot_source_test.go @@ -3,9 +3,11 @@ package worker import ( "context" "errors" + "sync" "testing" "time" + "google.golang.org/protobuf/types/known/timestamppb" controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/controlplane/snapshotwire" ownershipDomain "proxy-pool/internal/domain/ownership" @@ -35,6 +37,47 @@ func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) { } } +func TestRefreshingSnapshotSourceIssuesSequentialFullSnapshots(t *testing.T) { + base := &refreshingSnapshotSourceStub{} + source, err := NewRefreshingSnapshotSource(base, 10*time.Millisecond) + if err != nil { + t.Fatalf("NewRefreshingSnapshotSource() = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + updates, err := source.Watch(ctx, SnapshotWatchRequest{ + WorkerID: "worker-a", SessionID: "session-a", LastAppliedVersion: 4, LastChecksum: []byte{4}, + }) + if err != nil { + t.Fatalf("Watch() = %v", err) + } + first := receiveSnapshot(t, updates) + second := receiveSnapshot(t, updates) + if first.GetVersion() != 5 || second.GetVersion() != 6 || second.GetVersion() <= first.GetVersion() { + t.Fatalf("refreshed versions = (%d, %d)", first.GetVersion(), second.GetVersion()) + } + if calls := base.Calls(); len(calls) < 2 || calls[0] != 4 || calls[1] != 5 { + t.Fatalf("base requested versions = %v", calls) + } + if checksums := base.Checksums(); len(checksums) < 2 || string(checksums[0]) != string([]byte{4}) || + string(checksums[1]) != string([]byte{5}) { + t.Fatalf("base requested checksums = %v", checksums) + } +} + +func TestRefreshingSnapshotSourceRefreshesBeforeShortSnapshotExpiry(t *testing.T) { + source, err := NewRefreshingSnapshotSource(&refreshingSnapshotSourceStub{}, 2*time.Second) + if err != nil { + t.Fatalf("NewRefreshingSnapshotSource() = %v", err) + } + now := time.Now().UTC() + if delay := source.delayFor(&controlplanev1.WorkerSnapshot{ + GeneratedAt: timestamppb.New(now), ValidUntil: timestamppb.New(now.Add(100 * time.Millisecond)), + }); delay != 50*time.Millisecond { + t.Fatalf("delayFor(short-lived snapshot) = %s, want 50ms", delay) + } +} + func TestOwnedSnapshotSourceBuildsBoundedProxySnapshot(t *testing.T) { now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) expiresAt := now.Add(10 * time.Minute) @@ -167,6 +210,57 @@ type epochReaderStub struct { err error } +type refreshingSnapshotSourceStub struct { + mu sync.Mutex + calls []uint64 + checksums [][]byte +} + +func (source *refreshingSnapshotSourceStub) Watch( + _ context.Context, + request SnapshotWatchRequest, +) (<-chan *controlplanev1.WorkerSnapshot, error) { + source.mu.Lock() + source.calls = append(source.calls, request.LastAppliedVersion) + source.checksums = append(source.checksums, append([]byte(nil), request.LastChecksum...)) + source.mu.Unlock() + updates := make(chan *controlplanev1.WorkerSnapshot, 1) + updates <- &controlplanev1.WorkerSnapshot{ + Version: request.LastAppliedVersion + 1, Checksum: []byte{byte(request.LastAppliedVersion + 1)}, + } + return updates, nil +} + +func (source *refreshingSnapshotSourceStub) Calls() []uint64 { + source.mu.Lock() + defer source.mu.Unlock() + return append([]uint64(nil), source.calls...) +} + +func (source *refreshingSnapshotSourceStub) Checksums() [][]byte { + source.mu.Lock() + defer source.mu.Unlock() + checksums := make([][]byte, len(source.checksums)) + for index := range source.checksums { + checksums[index] = append([]byte(nil), source.checksums[index]...) + } + return checksums +} + +func receiveSnapshot(t *testing.T, updates <-chan *controlplanev1.WorkerSnapshot) *controlplanev1.WorkerSnapshot { + t.Helper() + select { + case snapshot, ok := <-updates: + if !ok || snapshot == nil { + t.Fatal("snapshot updates closed before next full snapshot") + } + return snapshot + case <-time.After(time.Second): + t.Fatal("timed out waiting for snapshot") + return nil + } +} + func (reader epochReaderStub) CurrentOwnershipEpoch(context.Context) (uint64, error) { return reader.epoch, reader.err } diff --git a/progress.md b/progress.md index 6f320dc..4b9a920 100644 --- a/progress.md +++ b/progress.md @@ -15,6 +15,10 @@ - Redis 集成标签测试在未设置 `PROXY_POOL_TEST_REDIS_URL` 时现能正确跳过所有契约 子测试;`go test ./...`、`go test -tags=integration ./internal/adapters/redisactivity` 与 `scripts/verify.ps1` 已通过,race 仍由 Linux CI 执行。 +- Worker 控制面改为持续完整 Snapshot 发布:Controller 在每份 Snapshot 有效期的一半 + 前重建并下发递增版本,Gateway 在同一长连接内原子替换视图;构建失败或流中断时才由 + SessionSupervisor 按原有退避重连。该机制复用 `RefreshingSnapshotSource`,不降低 + Delta 的 fail-closed 校验,也不把自动 Drain/ACK 记为已完成。 ## 2026-07-30 diff --git a/task_plan.md b/task_plan.md index 09304c9..8121e36 100644 --- a/task_plan.md +++ b/task_plan.md @@ -40,8 +40,8 @@ 14. [已完成] 收敛健康故障生命周期:将 `UNHEALTHY` 的持续时长纳入权威活动池状态, 通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy;拥有 Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。 -15. [进行中] 收敛 Worker 发布生命周期:补齐权威 Proxy/Routing 的增量发布和 - 自动 Drain 编排,使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。 +15. [进行中] 收敛 Worker 发布生命周期:已完成权威 Proxy/Routing 的持续完整快照 + 刷新;后续补齐自动 Drain 编排,使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。 ## 串并行关系