diff --git a/README.md b/README.md index a135dd3..b0a041c 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,9 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待， 以及仍经过目标地址策略的 direct；`proxy-gateway` 通过独立控制面拨号地址维护 Session，并在每份 Snapshot 有效期的一半前接收版本递增的完整刷新；仅在持有未过期 - Snapshot 时 Ready，凭据材料只保留在当前节点内存 View。 + Snapshot 时 Ready，凭据材料只保留在当前节点内存 View。Controller 内成功提交的 + Upstream 启停、Routing 切换和配置发布会向本进程全部在线 Worker 快照流广播刷新； + 定时刷新仍作为跨进程收敛与失效保护。 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 @@ -267,7 +269,9 @@ go run ./cmd/proxy-loadgen ` TARGET 事实按路由目标 Profile 独立归并。 - **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与 `onUnavailable` 已接入；上游停用会从后续完整 Snapshot 排除，并对现有 Worker - ownership 发起带策略 revision 栅栏的 Drain。动态容量调整仍待完成。 + ownership 发起带策略 revision 栅栏的 Drain。Routing 切换立即刷新在线 Worker 的 + 完整快照；Sequential 的新分配切到新上游，旧 Proxy 与既有连接自然排空。动态容量 + 调整仍待完成。 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑， 以及故障转移和恢复演练。 - **P2 - 容量证明**：`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测，以及 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 48480a4..2f7ffcf 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -27,7 +27,9 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和 `WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照，并在 每份快照有效期的一半前重新构建、下发版本递增的完整快照；这样 ownership、Routing 和 -凭据变化会在同一长连接内收敛，而无需等待有效期到达后重新注册。 +凭据变化会在同一长连接内收敛，而无需等待有效期到达后重新注册。Controller 内已提交的 +Upstream 启停、Routing 切换和配置发布还会向全部本地 Worker 流广播一次合并后的立即刷新； +多 Controller 副本仍由该定时机制完成跨进程收敛。 Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership 索引构建已归属 Proxy 内容，并将租约到期收紧到 Proxy 的 `usable_until`。Proxy 引用的 凭据材料按 `secret_ref + credential_version` 去重，随完整 Snapshot 经 mTLS 下发，仅保留在 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 18b9505..d460089 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -10,7 +10,7 @@ | ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP/CONNECT/Extract 校验场景均已实现 | | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 | | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 | -| ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；静态配置与 Admin 管理态共同决定 Provider 启停，停用策略以 Upstream revision 栅栏驱动既有 ownership Drain；多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | +| ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；静态配置与 Admin 管理态共同决定 Provider 启停，停用策略以 Upstream revision 栅栏驱动既有 ownership Drain；Routing 切换会向本进程 Worker 快照流广播完整刷新，多副本按定时刷新与权威 HMAC 指纹/revision 栅栏收敛并拒绝旧配置换主 | | ARCH-005 | 100k QPS 峰值使用多 Worker 集群 | 当前会话 | 未验证设计目标；待代表性集群负载报告 | ## Routing 与 Upstream @@ -22,7 +22,7 @@ | ROUTE-003 | 支持 sequential、random、roundRobin、weighted、leastConnections | 5825-6467 | 五种领域策略、同版本 Gateway Snapshot 派发和定向测试已完成；Distribution 接线待完成 | | ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；持久化恢复和跨实例 CAS 待完成 | | ROUTE-005 | 空计数属于 Upstream，当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 | -| ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽，不因切换直接丢弃 | 6618-6641 | 通用 ownership Drain/ACK 原语已测试；Routing 切换到 Drain 的编排待完成 | +| ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽，不因切换直接丢弃 | 6618-6641 | Routing 成功切换后立即发布完整快照；Sequential 仅将新分配切到新的 CurrentUpstream，旧 Proxy 仍保留在快照，既有 Active/Reserved 由本地运行态自然归零。共享 Upstream 不按单 Routing 强制 Drain，避免影响其他 Routing | | ROUTE-007 | 无可用 Upstream 时显式 reject、wait 或 direct，默认 reject | 5075-5294, 6743-6760 | Gateway 已实现 reject、带 `wait_timeout` 的本地容量等待与经 TargetPolicy 的 direct；Distribution 接线和默认化策略待完成 | ## Provider 与补池 diff --git a/findings.md b/findings.md index a2425b5..de432bb 100644 --- a/findings.md +++ b/findings.md @@ -130,6 +130,11 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 - Snapshot 的 `version` 是同一 Worker 流的连续序列，`ownership_epoch` 是独立且只能前进 的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1， 与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。 +- `SnapshotRefreshBroker` 是 Controller 进程内的公共扇出接口：已提交的 Upstream 启停、 + Routing 切换和配置发布各向每条本地 Worker 流发送一个可合并刷新信号。Routing 不记录 + 单代理归属，不能按单条 Routing 对共享 Upstream 强制 Drain；Sequential 切换只改变后续 + 新分配，旧 Proxy/连接由原有 Snapshot 与运行态自然排空。其他 Controller 副本继续在 + 固定刷新周期内读取 PostgreSQL 权威管理态并收敛。 ## Drain Ticket（2026-08-02） diff --git a/internal/controller/admin/service.go b/internal/controller/admin/service.go index e5eb126..cf72a47 100644 --- a/internal/controller/admin/service.go +++ b/internal/controller/admin/service.go @@ -37,14 +37,21 @@ type RuntimeController interface { ValidateUpstream(context.Context, string) error } +// SnapshotRefreshNotifier requests immediate full Worker snapshots after a +// committed management change. Implementations must coalesce notifications. +type SnapshotRefreshNotifier interface { + NotifySnapshotRefresh() +} + var _ ConfigurationPublisher = (*config.Store)(nil) type ApplicationDependencies struct { - State StateRepository - Operations OperationalStatusReader - Configuration ConfigurationLoader - Publisher ConfigurationPublisher - Runtime RuntimeController + State StateRepository + Operations OperationalStatusReader + Configuration ConfigurationLoader + Publisher ConfigurationPublisher + Runtime RuntimeController + SnapshotRefresh SnapshotRefreshNotifier } type ApplicationOptions struct { @@ -75,13 +82,14 @@ type LoadedConfiguration struct { } type ApplicationService struct { - state StateRepository - operations OperationalStatusReader - configuration ConfigurationLoader - publisher ConfigurationPublisher - runtime RuntimeController - now func() time.Time - fingerprintKey []byte + state StateRepository + operations OperationalStatusReader + configuration ConfigurationLoader + publisher ConfigurationPublisher + runtime RuntimeController + snapshotRefresh SnapshotRefreshNotifier + now func() time.Time + fingerprintKey []byte } var _ Service = (*ApplicationService)(nil) @@ -93,13 +101,14 @@ func NewApplicationService(dependencies ApplicationDependencies, options Applica return nil, ErrInvalidApplicationService } return &ApplicationService{ - state: dependencies.State, - operations: dependencies.Operations, - configuration: dependencies.Configuration, - publisher: dependencies.Publisher, - runtime: dependencies.Runtime, - now: options.Now, - fingerprintKey: append([]byte(nil), options.FingerprintKey...), + state: dependencies.State, + operations: dependencies.Operations, + configuration: dependencies.Configuration, + publisher: dependencies.Publisher, + runtime: dependencies.Runtime, + snapshotRefresh: dependencies.SnapshotRefresh, + now: options.Now, + fingerprintKey: append([]byte(nil), options.FingerprintKey...), }, nil } @@ -119,6 +128,9 @@ func (service *ApplicationService) SetUpstreamEnabled(ctx context.Context, comma if err == nil && service.runtime != nil { service.runtime.Notify() } + if err == nil && result.Changed && service.snapshotRefresh != nil { + service.snapshotRefresh.NotifySnapshotRefresh() + } return mutationResult(result), mapAdminStateError(err) } @@ -132,6 +144,9 @@ func (service *ApplicationService) SwitchRouting(ctx context.Context, command Sw Target: command.Target, Reason: command.Reason, }) + if err == nil && result.Changed && service.snapshotRefresh != nil { + service.snapshotRefresh.NotifySnapshotRefresh() + } return mutationResult(result), mapAdminStateError(err) } @@ -256,6 +271,9 @@ func (service *ApplicationService) ApplyConfiguration( if published && service.runtime != nil { service.runtime.Notify() } + if published && service.snapshotRefresh != nil { + service.snapshotRefresh.NotifySnapshotRefresh() + } return mutationResult(result), nil } diff --git a/internal/controller/admin/service_test.go b/internal/controller/admin/service_test.go index 7dcaddd..0ba0c3a 100644 --- a/internal/controller/admin/service_test.go +++ b/internal/controller/admin/service_test.go @@ -29,12 +29,14 @@ func TestApplicationServiceMapsUpstreamMutationToAdminState(t *testing.T) { }, } runtime := &recordingRuntimeNotifier{} + refresh := &recordingSnapshotRefreshNotifier{} service, err := NewApplicationService(ApplicationDependencies{ - State: state, - Operations: staticOperationalStatusReader{}, - Configuration: staticConfigurationLoader{}, - Publisher: &recordingConfigurationPublisher{}, - Runtime: runtime, + State: state, + Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, + Runtime: runtime, + SnapshotRefresh: refresh, }, applicationTestOptions(func() time.Time { return now })) if err != nil { t.Fatalf("NewApplicationService() error = %v", err) @@ -65,6 +67,9 @@ func TestApplicationServiceMapsUpstreamMutationToAdminState(t *testing.T) { if runtime.notifications != 1 { t.Fatalf("runtime notifications = %d, want 1", runtime.notifications) } + if refresh.notifications != 1 { + t.Fatalf("snapshot refresh notifications = %d, want 1", refresh.notifications) + } } func TestApplicationServicePreflightsProviderRuntimeBeforeMutation(t *testing.T) { @@ -147,7 +152,14 @@ func TestApplicationServiceMapsRoutingSwitchAndDomainErrors(t *testing.T) { t.Parallel() now := time.Date(2026, 7, 29, 11, 0, 0, 0, time.FixedZone("test", 8*60*60)) state := &recordingAdminState{mutation: adminstate.MutationResult{RequestID: "req-switch", Changed: true, Revision: 21}} - service := mustApplicationService(t, state, applicationTestOptions(func() time.Time { return now })) + refresh := &recordingSnapshotRefreshNotifier{} + service, serviceErr := NewApplicationService(ApplicationDependencies{ + State: state, Operations: staticOperationalStatusReader{}, Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, SnapshotRefresh: refresh, + }, applicationTestOptions(func() time.Time { return now })) + if serviceErr != nil { + t.Fatalf("NewApplicationService() = %v", serviceErr) + } result, err := service.SwitchRouting(context.Background(), SwitchCommand{ RequestID: "req-switch", ActorID: "admin:bob", SourceIP: "198.51.100.7", @@ -166,6 +178,9 @@ func TestApplicationServiceMapsRoutingSwitchAndDomainErrors(t *testing.T) { if state.lastSwitch != wantCommand { t.Fatalf("admin state command = %+v, want %+v", state.lastSwitch, wantCommand) } + if refresh.notifications != 1 { + t.Fatalf("snapshot refresh notifications = %d, want 1", refresh.notifications) + } tests := []struct { domain error @@ -186,6 +201,9 @@ func TestApplicationServiceMapsRoutingSwitchAndDomainErrors(t *testing.T) { t.Fatalf("SwitchRouting(%v) error = %v, want mapped %v preserving cause", test.domain, err, test.want) } } + if refresh.notifications != 1 { + t.Fatalf("snapshot refresh notifications after failed switches = %d, want 1", refresh.notifications) + } } func TestApplicationServiceBuildsStatusFromAuthoritativeAndOperationalSnapshots(t *testing.T) { @@ -281,6 +299,7 @@ func TestApplicationServiceReloadPersistsManagementViewBeforePublishing(t *testi configuration := validReloadConfiguration() publisher := &recordingConfigurationPublisher{} runtime := &recordingRuntimeNotifier{} + refresh := &recordingSnapshotRefreshNotifier{} state := &recordingAdminState{ mutation: adminstate.MutationResult{RequestID: "req-reload", Changed: true, Revision: 42}, snapshot: adminstate.Snapshot{Routings: []adminstate.RoutingState{ @@ -297,7 +316,7 @@ func TestApplicationServiceReloadPersistsManagementViewBeforePublishing(t *testi Configuration: staticConfigurationLoader{loaded: LoadedConfiguration{ Value: configuration, Source: "configs/proxy-pool.yaml", }}, - Publisher: publisher, Runtime: runtime, + Publisher: publisher, Runtime: runtime, SnapshotRefresh: refresh, }, applicationTestOptions(func() time.Time { return now })) if err != nil { t.Fatalf("NewApplicationService() error = %v", err) @@ -318,6 +337,9 @@ func TestApplicationServiceReloadPersistsManagementViewBeforePublishing(t *testi if runtime.notifications != 1 { t.Fatalf("runtime notifications = %d, want 1", runtime.notifications) } + if refresh.notifications != 1 { + t.Fatalf("snapshot refresh notifications = %d, want 1", refresh.notifications) + } command := state.lastConfig if command.RequestID != "req-reload" || command.Actor != (adminstate.Actor{ID: "admin:alice", SourceIP: "192.0.2.10"}) || !command.OccurredAt.Equal(now) || command.Source != "configs/proxy-pool.yaml" { @@ -712,6 +734,10 @@ type recordingRuntimeNotifier struct { validationErr error } +type recordingSnapshotRefreshNotifier struct{ notifications int } + +func (notifier *recordingSnapshotRefreshNotifier) NotifySnapshotRefresh() { notifier.notifications++ } + func (notifier *recordingRuntimeNotifier) Notify() { notifier.notifications++ } func (notifier *recordingRuntimeNotifier) ValidateConfiguration(context.Context, *config.Config) error { diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index 5fba07f..e9e0358 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -176,6 +176,7 @@ func runWithWorkerFactory( if err != nil { return fmt.Errorf("%w: build Provider supervisor: %w", ErrStartup, err) } + snapshotRefresh := worker.NewSnapshotRefreshBroker() dependencies := controllerRuntime.Dependencies{} if loaded.Value.Distribution.Enabled { @@ -208,7 +209,7 @@ func runWithWorkerFactory( } service, serviceErr := admin.NewApplicationService(admin.ApplicationDependencies{ State: opened.state, Operations: statusReader, Configuration: loader, Publisher: configurationStore, - Runtime: supervisor, + Runtime: supervisor, SnapshotRefresh: snapshotRefresh, }, admin.ApplicationOptions{Now: options.Now, FingerprintKey: options.FingerprintKey}) if serviceErr != nil { return fmt.Errorf("%w: build admin service: %w", ErrStartup, serviceErr) @@ -290,6 +291,7 @@ func runWithWorkerFactory( DrainTickets: drainTickets, RoutingSource: routingSource, Credentials: opened.credentials, + SnapshotRefresh: snapshotRefresh, }) if serviceErr != nil { return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr) diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index 3aa7120..669e398 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -63,6 +63,7 @@ type Options struct { DrainTickets ownershipDomain.DrainTicketStore RoutingSource RoutingSource Credentials platformCredentials.Store + SnapshotRefresh SnapshotRefreshSubscriber SessionID func() (string, error) } @@ -158,7 +159,15 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err if err != nil { return nil, errors.Join(ErrInvalidCommand, err) } - refreshed, refreshErr := NewRefreshingSnapshotSource(source, snapshotRefreshEvery(options.MaxStaleAge)) + var refreshed *RefreshingSnapshotSource + var refreshErr error + if options.SnapshotRefresh == nil { + refreshed, refreshErr = NewRefreshingSnapshotSource(source, snapshotRefreshEvery(options.MaxStaleAge)) + } else { + refreshed, refreshErr = NewRefreshingSnapshotSourceWithRefreshEvents( + source, snapshotRefreshEvery(options.MaxStaleAge), options.SnapshotRefresh, + ) + } if refreshErr != nil { return nil, errors.Join(ErrInvalidCommand, refreshErr) } diff --git a/internal/controller/worker/snapshot_refresh.go b/internal/controller/worker/snapshot_refresh.go new file mode 100644 index 0000000..df6f707 --- /dev/null +++ b/internal/controller/worker/snapshot_refresh.go @@ -0,0 +1,62 @@ +package worker + +import "sync" + +// SnapshotRefreshNotifier requests every subscribed Worker snapshot stream to +// issue a new full snapshot. Notifications are coalesced per subscriber. +type SnapshotRefreshNotifier interface { + NotifySnapshotRefresh() +} + +// SnapshotRefreshSubscriber receives controller-local full-snapshot refresh +// requests. The cancel function releases the subscription. +type SnapshotRefreshSubscriber interface { + SubscribeSnapshotRefresh() (<-chan struct{}, func()) +} + +// SnapshotRefreshBroker fans a management-plane change out to every local +// Worker snapshot stream. It contains no proxy or client data. +type SnapshotRefreshBroker struct { + mu sync.Mutex + subscribers map[chan struct{}]struct{} +} + +func NewSnapshotRefreshBroker() *SnapshotRefreshBroker { + return &SnapshotRefreshBroker{subscribers: make(map[chan struct{}]struct{})} +} + +func (broker *SnapshotRefreshBroker) NotifySnapshotRefresh() { + if broker == nil { + return + } + broker.mu.Lock() + defer broker.mu.Unlock() + for subscriber := range broker.subscribers { + select { + case subscriber <- struct{}{}: + default: + } + } +} + +func (broker *SnapshotRefreshBroker) SubscribeSnapshotRefresh() (<-chan struct{}, func()) { + if broker == nil { + return nil, func() {} + } + subscriber := make(chan struct{}, 1) + broker.mu.Lock() + broker.subscribers[subscriber] = struct{}{} + broker.mu.Unlock() + + var once sync.Once + return subscriber, func() { + once.Do(func() { + broker.mu.Lock() + delete(broker.subscribers, subscriber) + broker.mu.Unlock() + }) + } +} + +var _ SnapshotRefreshNotifier = (*SnapshotRefreshBroker)(nil) +var _ SnapshotRefreshSubscriber = (*SnapshotRefreshBroker)(nil) diff --git a/internal/controller/worker/snapshot_source.go b/internal/controller/worker/snapshot_source.go index f820afe..02d57af 100644 --- a/internal/controller/worker/snapshot_source.go +++ b/internal/controller/worker/snapshot_source.go @@ -29,15 +29,31 @@ type SnapshotSource interface { // 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 + source SnapshotSource + refreshEvery time.Duration + refreshEvents SnapshotRefreshSubscriber } func NewRefreshingSnapshotSource(source SnapshotSource, refreshEvery time.Duration) (*RefreshingSnapshotSource, error) { + return NewRefreshingSnapshotSourceWithRefreshEvents(source, refreshEvery) +} + +func NewRefreshingSnapshotSourceWithRefreshEvents( + source SnapshotSource, + refreshEvery time.Duration, + refreshEvents ...SnapshotRefreshSubscriber, +) (*RefreshingSnapshotSource, error) { + if len(refreshEvents) > 1 || len(refreshEvents) == 1 && refreshEvents[0] == nil { + return nil, ErrSnapshotsUnavailable + } if source == nil || refreshEvery <= 0 { return nil, ErrSnapshotsUnavailable } - return &RefreshingSnapshotSource{source: source, refreshEvery: refreshEvery}, nil + result := &RefreshingSnapshotSource{source: source, refreshEvery: refreshEvery} + if len(refreshEvents) == 1 { + result.refreshEvents = refreshEvents[0] + } + return result, nil } func snapshotRefreshEvery(validFor time.Duration) time.Duration { @@ -57,13 +73,22 @@ func (source *RefreshingSnapshotSource) Watch( if source == nil || source.source == nil || ctx == nil { return nil, ErrSnapshotsUnavailable } + var refreshEvents <-chan struct{} + cancelRefresh := func() {} + if source.refreshEvents != nil { + refreshEvents, cancelRefresh = source.refreshEvents.SubscribeSnapshotRefresh() + if refreshEvents == nil || cancelRefresh == nil { + return nil, ErrSnapshotsUnavailable + } + } first, err := source.next(ctx, request) if err != nil { + cancelRefresh() return nil, err } updates := make(chan *controlplanev1.WorkerSnapshot, 1) updates <- first - go source.refresh(ctx, updates, request, first) + go source.refresh(ctx, updates, request, first, refreshEvents, cancelRefresh) return updates, nil } @@ -72,8 +97,11 @@ func (source *RefreshingSnapshotSource) refresh( updates chan<- *controlplanev1.WorkerSnapshot, request SnapshotWatchRequest, current *controlplanev1.WorkerSnapshot, + refreshEvents <-chan struct{}, + cancelRefresh func(), ) { defer close(updates) + defer cancelRefresh() version := current.GetVersion() checksum := append([]byte(nil), current.GetChecksum()...) timer := time.NewTimer(source.delayFor(current)) @@ -83,20 +111,21 @@ func (source *RefreshingSnapshotSource) refresh( 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)) - } + case <-refreshEvents: + } + 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)) } } } diff --git a/internal/controller/worker/snapshot_source_test.go b/internal/controller/worker/snapshot_source_test.go index fccd79c..5c5410d 100644 --- a/internal/controller/worker/snapshot_source_test.go +++ b/internal/controller/worker/snapshot_source_test.go @@ -65,6 +65,71 @@ func TestRefreshingSnapshotSourceIssuesSequentialFullSnapshots(t *testing.T) { } } +func TestRefreshingSnapshotSourceRefreshesImmediatelyAfterManagementNotification(t *testing.T) { + base := &refreshingSnapshotSourceStub{} + broker := NewSnapshotRefreshBroker() + source, err := NewRefreshingSnapshotSourceWithRefreshEvents(base, time.Hour, broker) + if err != nil { + t.Fatalf("NewRefreshingSnapshotSourceWithRefreshEvents() = %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) + } + if first := receiveSnapshot(t, updates); first.GetVersion() != 5 { + t.Fatalf("initial version = %d, want 5", first.GetVersion()) + } + broker.NotifySnapshotRefresh() + if refreshed := receiveSnapshot(t, updates); refreshed.GetVersion() != 6 { + t.Fatalf("refreshed version = %d, want 6", refreshed.GetVersion()) + } + if calls := base.Calls(); len(calls) != 2 || calls[0] != 4 || calls[1] != 5 { + t.Fatalf("base requested versions = %v", calls) + } +} + +func TestSnapshotRefreshBrokerBroadcastsCoalescedNotifications(t *testing.T) { + broker := NewSnapshotRefreshBroker() + first, unsubscribeFirst := broker.SubscribeSnapshotRefresh() + second, unsubscribeSecond := broker.SubscribeSnapshotRefresh() + defer unsubscribeSecond() + + broker.NotifySnapshotRefresh() + broker.NotifySnapshotRefresh() + select { + case <-first: + case <-time.After(time.Second): + t.Fatal("first subscriber did not receive notification") + } + select { + case <-second: + case <-time.After(time.Second): + t.Fatal("second subscriber did not receive notification") + } + select { + case <-first: + t.Fatal("first subscriber received an uncoalesced notification") + default: + } + + unsubscribeFirst() + broker.NotifySnapshotRefresh() + select { + case <-first: + t.Fatal("unsubscribed receiver received a notification") + default: + } + select { + case <-second: + case <-time.After(time.Second): + t.Fatal("remaining subscriber did not receive notification") + } +} + func TestRefreshingSnapshotSourceRefreshesBeforeShortSnapshotExpiry(t *testing.T) { source, err := NewRefreshingSnapshotSource(&refreshingSnapshotSourceStub{}, 2*time.Second) if err != nil { diff --git a/progress.md b/progress.md index 68e0aec..7f30466 100644 --- a/progress.md +++ b/progress.md @@ -2,6 +2,11 @@ ## 2026-08-02 +- 已实现管理态驱动的即时完整快照刷新：`SnapshotRefreshBroker` 以每个 Worker 流一格的 + 非阻塞通道广播，连续变更自动合并。Admin 成功提交 Upstream 启停、Routing 切换和配置 + 发布时触发通知；Worker 保留原有定时刷新作为失效保护。Routing 切换不按旧 Upstream + 强制 Drain，共享上游不会受误伤；Sequential 的后续预留直接使用新 CurrentUpstream， + 旧 Proxy 和活跃连接自然排空。Worker/Admin/Bootstrap 定向测试已通过。 - Drain 可观测性已落地：`proxy_pool_controller_drain_candidates_total` 与 `proxy_pool_controller_drains_started_total` 只允许 `unhealthy` 和 `upstream_disabled` 两种 reason。健康异常与上游停用 Reaper 复用领域接口上报， diff --git a/task_plan.md b/task_plan.md index 5cb7817..9f33dc7 100644 --- a/task_plan.md +++ b/task_plan.md @@ -45,7 +45,9 @@ 完整排除 Snapshot 屏障绑定；已在 ACK 与 Runtime 零计数之间完成 Redis 原子自动收尾。 配置与管理态共同决定的上游停用已接入同一 `BeginDrain` 编排：候选绑定上游 revision， Redis 原子复核策略、归属与 assignment epoch；两类自动 Drain 均已接入低基数 - Prometheus 指标。后续补齐更完整的运行态可观测闭环。 + Prometheus 指标。Admin 成功提交的 Upstream 启停、Routing 切换和配置发布现会经 + 公用广播器立即刷新本进程所有 Worker 完整 Snapshot；跨 Controller 副本仍以定时刷新 + 收敛。后续补齐更完整的运行态可观测闭环。 ## 串并行关系