feat: refresh worker snapshots after routing changes

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 分场景压测，以及

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 下发，仅保留在

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 与补池

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）
 

internal/controller/admin/service.go

@@ -37,6 +37,12 @@ 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 {
@@ -45,6 +51,7 @@ type ApplicationDependencies struct {
 	Configuration   ConfigurationLoader
 	Publisher       ConfigurationPublisher
 	Runtime         RuntimeController
+	SnapshotRefresh SnapshotRefreshNotifier
 }
 
 type ApplicationOptions struct {
@@ -80,6 +87,7 @@ type ApplicationService struct {
 	configuration   ConfigurationLoader
 	publisher       ConfigurationPublisher
 	runtime         RuntimeController
+	snapshotRefresh SnapshotRefreshNotifier
 	now             func() time.Time
 	fingerprintKey  []byte
 }
@@ -98,6 +106,7 @@ func NewApplicationService(dependencies ApplicationDependencies, options Applica
 		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
 }
 

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,
+		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 {

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)

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)
 		}

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)

internal/controller/worker/snapshot_source.go

@@ -31,13 +31,29 @@ type SnapshotSource interface {
 type RefreshingSnapshotSource struct {
 	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,6 +111,8 @@ func (source *RefreshingSnapshotSource) refresh(
 		case <-ctx.Done():
 			return
 		case <-timer.C:
+		case <-refreshEvents:
+		}
 		request.LastAppliedVersion = version
 		request.LastChecksum = append(request.LastChecksum[:0], checksum...)
 		next, err := source.next(ctx, request)
@@ -98,7 +128,6 @@ func (source *RefreshingSnapshotSource) refresh(
 			timer.Reset(source.delayFor(next))
 		}
 	}
-	}
 }
 
 func (source *RefreshingSnapshotSource) next(

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 {

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 复用领域接口上报，

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 副本仍以定时刷新
+    收敛。后续补齐更完整的运行态可观测闭环。
 
 ## 串并行关系