diff --git a/README.md b/README.md index 8ecb782..7855de6 100644 --- a/README.md +++ b/README.md @@ -115,7 +115,8 @@ flowchart LR Redis 会话栅栏，以及 Gateway Outcome 上报的有界队列、序列确认与重试； Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 Checker 探测和 - Observation 状态归并。 + Observation 状态归并；Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤和 + Snapshot 即时刷新。 - **部分完成**：Docker Compose/Kubernetes 运行时 mTLS Overlay。 - **待完成**：CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index a4184e3..15403cc 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -104,9 +104,13 @@ Worker 维度；该测试纳入 Linux CI 的 race 范围，本机因 `CGO_ENABLE - [ ] Cover four-empty-then-success, five-empty, A-to-B-only, disabled references, end behavior, and explicit onUnavailable. -当前进度（2026-07-29）：领域构造器与严格配置已统一 Sequential 至少两个 -Upstream、`endBehavior` 默认 `stop`，并覆盖列表末端停止；disabled candidate、 -跨实例恢复和 `onUnavailable` 运行链仍待完成。 +当前进度（2026-08-02）：领域构造器与严格配置已统一 Sequential 至少两个 +Upstream、`endBehavior` 默认 `stop`，并覆盖列表末端停止。Provider Stats 现为每次 +连续空结果分配单调代次；Controller 的公共 Sequential 协调器只接收有界通知，在独立 +循环中读取权威配置、管理快照和 Stats，并以既有 `ExpectedCurrent` CAS 自动切换。 +它会跳过禁用 Upstream，支持 `loop`/`stayLast`，同一空结果代次不会在循环后重复切换， +成功后立即广播完整 Snapshot。Gateway 已执行 reject/wait/direct；控制面将末端 `stop` +持久化为禁用路由的原子管理态变更仍待实现。 ## Task 5: Provider Fetch Classification and Scheduling diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 6f4ab6b..eebfb6a 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -124,8 +124,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 生产任务调度与 REMOVE 编排仍待实现。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 -10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS - 和 disabled candidate 语义。 +10. 将 reject/wait/direct 接入 Distribution 运行链；Sequential 已接入 Provider 空结果 + 代次、管理库 CAS 和禁用候选过滤，仍需补齐末端 `stop` 的原子路由停用管理态。 ## 4. 容量结论 100,000 QPS 是集群设计输入，不是本次验证结果。只有实现上述运行时，并在 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 538d3e9..8a0008a 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -20,7 +20,7 @@ | ROUTE-001 | Routing 自上而下匹配，首条命中停止 | 3534-3798, 5825-6467 | `rule.go` 与不可变/首命中单测 | | ROUTE-002 | Routing 与 Upstream 生命周期解耦 | 3534-3798 | 包依赖与配置模型 | | ROUTE-003 | 支持 sequential、random、roundRobin、weighted、leastConnections | 5825-6467 | 五种领域策略、同版本 Gateway Snapshot 派发和定向测试已完成；Distribution 接线待完成 | -| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；持久化恢复和跨实例 CAS 待完成 | +| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；Provider Stats 对连续空结果分配单调代次，Controller `SequentialCoordinator` 在独立有界循环中读取权威配置/管理快照，并通过管理库 `ExpectedCurrent` CAS 自动切换。禁用候选会跳过，重复 Tick、循环后旧代次和并发 Tick 均不会再次切换；成功即广播完整 Snapshot | | ROUTE-005 | 空计数属于 Upstream，当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 | | 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 接线和默认化策略待完成 | diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index a8e2fa5..f5faf37 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -19,6 +19,7 @@ import ( "proxy-pool/internal/controller/operations" "proxy-pool/internal/controller/pool" "proxy-pool/internal/controller/provider" + controllerRouting "proxy-pool/internal/controller/routing" controllerRuntime "proxy-pool/internal/controller/runtime" "proxy-pool/internal/controller/worker" "proxy-pool/internal/domain/activitypool" @@ -178,6 +179,7 @@ func runWithWorkerFactory( return fmt.Errorf("%w: build Provider supervisor: %w", ErrStartup, err) } snapshotRefresh := worker.NewSnapshotRefreshBroker() + var sequentialCoordinator lifecycle.Runner dependencies := controllerRuntime.Dependencies{} var checkerMetrics healthDomain.TaskMetricsObserver @@ -269,6 +271,22 @@ func runWithWorkerFactory( return fmt.Errorf("%w: commit startup configuration: %w", ErrStartup, applyErr) } dependencies.AdminService = service + if stats, ok := opened.providerResults.(provider.StatsReader); ok && !nilInterface(stats) { + coordinator, coordinatorErr := controllerRouting.NewSequentialCoordinator( + configurationStore, + opened.state, + stats, + snapshotRefresh, + controllerRouting.Options{Now: options.Now}, + ) + if coordinatorErr != nil { + return fmt.Errorf("%w: build Sequential routing coordinator: %w", ErrStartup, coordinatorErr) + } + if registrar, ok := opened.providerResults.(provider.ResultObserverRegistrar); ok && !nilInterface(registrar) { + registrar.AddResultObserver(coordinator) + } + sequentialCoordinator = coordinator + } } runners := make([]lifecycle.Runner, 0, 3+len(loaded.Value.Upstreams)) if hasHTTPRuntime(loaded.Value) { @@ -396,6 +414,9 @@ func runWithWorkerFactory( runners = append(runners, reaper) } } + if sequentialCoordinator != nil { + runners = append(runners, sequentialCoordinator) + } runners = append(runners, supervisor) group, err := lifecycle.NewGroup(runners...) if err != nil { diff --git a/internal/controller/provider/stats.go b/internal/controller/provider/stats.go index 580c583..c1771a1 100644 --- a/internal/controller/provider/stats.go +++ b/internal/controller/provider/stats.go @@ -14,6 +14,7 @@ var ErrInvalidStatsRecorder = errors.New("invalid Provider stats recorder") type Stats struct { UpstreamID string ConsecutiveEmptyFetch int64 + EmptyGeneration uint64 FetchErrorCount int64 } @@ -26,10 +27,11 @@ type StatsRetainer interface { } type StatsRecorder struct { - mu sync.Mutex - maximum int - byID map[string]Stats - observers []ResultObserver + mu sync.Mutex + maximum int + byID map[string]Stats + observers []ResultObserver + nextEmptyGeneration uint64 } func NewStatsRecorder(maximum int) (*StatsRecorder, error) { @@ -56,6 +58,10 @@ func (recorder *StatsRecorder) Record(result Result) { } switch result.Class { case upstream.FetchEmpty: + if stats.ConsecutiveEmptyFetch == 0 && recorder.nextEmptyGeneration < math.MaxUint64 { + recorder.nextEmptyGeneration++ + stats.EmptyGeneration = recorder.nextEmptyGeneration + } if stats.ConsecutiveEmptyFetch < math.MaxInt64 { stats.ConsecutiveEmptyFetch++ } diff --git a/internal/controller/provider/stats_test.go b/internal/controller/provider/stats_test.go index f4e009c..fd7578c 100644 --- a/internal/controller/provider/stats_test.go +++ b/internal/controller/provider/stats_test.go @@ -29,6 +29,26 @@ func TestStatsRecorderTracksEmptyResetAndErrors(t *testing.T) { } } +func TestStatsRecorderAssignsOneGenerationToEachEmptyEpisode(t *testing.T) { + recorder, err := NewStatsRecorder(1) + if err != nil { + t.Fatalf("NewStatsRecorder() = %v", err) + } + recorder.Record(Result{UpstreamID: "provider-a", Class: upstream.FetchEmpty}) + recorder.Record(Result{UpstreamID: "provider-a", Class: upstream.FetchEmpty}) + first := recorder.ReadProviderStats([]string{"provider-a"})[0] + if first.ConsecutiveEmptyFetch != 2 || first.EmptyGeneration == 0 { + t.Fatalf("first empty episode = %+v, want count=2 and generation", first) + } + + recorder.Record(Result{UpstreamID: "provider-a", Class: upstream.FetchValid}) + recorder.Record(Result{UpstreamID: "provider-a", Class: upstream.FetchEmpty}) + second := recorder.ReadProviderStats([]string{"provider-a"})[0] + if second.ConsecutiveEmptyFetch != 1 || second.EmptyGeneration <= first.EmptyGeneration { + t.Fatalf("second empty episode = %+v after %+v, want reset count and newer generation", second, first) + } +} + func TestStatsRecorderIsBoundedAndConcurrent(t *testing.T) { recorder, err := NewStatsRecorder(1) if err != nil { diff --git a/internal/controller/routing/sequential.go b/internal/controller/routing/sequential.go new file mode 100644 index 0000000..e2b9fd5 --- /dev/null +++ b/internal/controller/routing/sequential.go @@ -0,0 +1,302 @@ +// Package routing coordinates Controller-side automatic Sequential switching. +// It consumes bounded Provider result notifications and keeps the actual state +// transition in the existing management-store compare-and-swap boundary. +package routing + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "reflect" + "strconv" + "sync" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/controller/provider" + "proxy-pool/internal/domain/adminstate" + "proxy-pool/internal/domain/upstream" +) + +var ErrInvalidSequentialCoordinator = errors.New("invalid sequential routing coordinator") + +const ( + defaultPollInterval = time.Second + autoSwitchActor = "proxy-controller" + autoSwitchReason = "consecutive empty provider fetches reached routing threshold" +) + +// ConfigurationSource supplies one immutable configuration and its matching +// authoritative management revision. +type ConfigurationSource interface { + Snapshot() (*config.Config, uint64) +} + +// StateStore contains the only mutation boundary for routing current-upstream +// state. Implementations enforce ExpectedCurrent as a compare-and-swap fence. +type StateStore interface { + Snapshot(context.Context) (adminstate.Snapshot, error) + SwitchRouting(context.Context, adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error) +} + +type SnapshotRefreshNotifier interface { + NotifySnapshotRefresh() +} + +type Options struct { + Now func() time.Time + PollInterval time.Duration +} + +type TickResult struct { + Switched int +} + +// SequentialCoordinator observes Provider fetch outcomes without blocking the +// fetch path. Tick serializes bounded scans and only records a generation as +// processed after it has made, or intentionally declined, a decision. +type SequentialCoordinator struct { + configuration ConfigurationSource + state StateStore + stats provider.StatsReader + refresh SnapshotRefreshNotifier + now func() time.Time + pollInterval time.Duration + notify chan struct{} + + mu sync.Mutex + processed map[string]uint64 +} + +var _ provider.ResultObserver = (*SequentialCoordinator)(nil) + +func NewSequentialCoordinator( + configuration ConfigurationSource, + state StateStore, + stats provider.StatsReader, + refresh SnapshotRefreshNotifier, + options Options, +) (*SequentialCoordinator, error) { + if nilInterface(configuration) || nilInterface(state) || nilInterface(stats) || options.Now == nil { + return nil, ErrInvalidSequentialCoordinator + } + if options.PollInterval == 0 { + options.PollInterval = defaultPollInterval + } + if options.PollInterval <= 0 { + return nil, ErrInvalidSequentialCoordinator + } + return &SequentialCoordinator{ + configuration: configuration, + state: state, + stats: stats, + refresh: refresh, + now: options.Now, + pollInterval: options.PollInterval, + notify: make(chan struct{}, 1), + processed: make(map[string]uint64), + }, nil +} + +// ObserveProviderResult deliberately performs no store calls. Every validated +// result can affect an empty episode, but one buffered notification is enough +// because Tick reads the latest bounded Stats snapshot. +func (coordinator *SequentialCoordinator) ObserveProviderResult(result provider.Result) { + if coordinator == nil || !validResultClass(result.Class) || result.UpstreamID == "" { + return + } + select { + case coordinator.notify <- struct{}{}: + default: + } +} + +func (coordinator *SequentialCoordinator) Run(ctx context.Context) error { + if coordinator == nil || ctx == nil || coordinator.pollInterval <= 0 || coordinator.now == nil || coordinator.notify == nil { + return ErrInvalidSequentialCoordinator + } + ticker := time.NewTicker(coordinator.pollInterval) + defer ticker.Stop() + for { + // A temporary management-store failure must not terminate Provider + // reconciliation. The retained empty generation is evaluated on retry. + _, _ = coordinator.Tick(ctx) + select { + case <-ctx.Done(): + return ctx.Err() + case <-coordinator.notify: + case <-ticker.C: + } + } +} + +func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult, error) { + if coordinator == nil || ctx == nil || nilInterface(coordinator.configuration) || nilInterface(coordinator.state) || + nilInterface(coordinator.stats) || coordinator.now == nil { + return TickResult{}, ErrInvalidSequentialCoordinator + } + if err := ctx.Err(); err != nil { + return TickResult{}, err + } + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + + configuration, revision := coordinator.configuration.Snapshot() + if configuration == nil || revision == 0 { + return TickResult{}, ErrInvalidSequentialCoordinator + } + snapshot, err := coordinator.state.Snapshot(ctx) + if err != nil { + return TickResult{}, err + } + if snapshot.Config == nil || snapshot.Config.Revision != revision { + return TickResult{}, ErrInvalidSequentialCoordinator + } + + upstreams := upstreamStates(snapshot.Upstreams) + routings := routingStates(snapshot.Routings) + result := TickResult{} + for _, route := range configuration.Routing { + if !route.Enabled || route.Strategy.Type != "sequential" || route.Strategy.SwitchAfterEmptyFetch <= 0 { + continue + } + state, exists := routings[route.Name] + if !exists || !state.Enabled || !upstreamEnabled(configuration, upstreams, state.CurrentUpstream) { + continue + } + read := coordinator.stats.ReadProviderStats([]string{state.CurrentUpstream}) + if len(read) != 1 { + return result, ErrInvalidSequentialCoordinator + } + stats := read[0] + if stats.UpstreamID != state.CurrentUpstream || stats.ConsecutiveEmptyFetch < int64(route.Strategy.SwitchAfterEmptyFetch) || + stats.EmptyGeneration == 0 { + coordinator.clearProcessed(route.Name, state.CurrentUpstream) + continue + } + key := processedKey(route.Name, state.CurrentUpstream) + if coordinator.processed[key] == stats.EmptyGeneration { + continue + } + target, switchable := nextUpstream(route, state.CurrentUpstream, upstreams, configuration) + if !switchable { + coordinator.processed[key] = stats.EmptyGeneration + continue + } + mutation, mutateErr := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{ + RequestID: requestID(route.Name, state.CurrentUpstream, stats.EmptyGeneration), + Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(), + Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Target: target, Reason: autoSwitchReason, + }) + if mutateErr != nil { + if errors.Is(mutateErr, adminstate.ErrConflict) { + coordinator.processed[key] = stats.EmptyGeneration + continue + } + return result, mutateErr + } + coordinator.processed[key] = stats.EmptyGeneration + if mutation.Changed { + result.Switched++ + if coordinator.refresh != nil { + coordinator.refresh.NotifySnapshotRefresh() + } + } + } + return result, nil +} + +func nextUpstream( + route config.Routing, + current string, + states map[string]adminstate.UpstreamState, + configuration *config.Config, +) (string, bool) { + eligible := make([]string, 0, len(route.Upstreams)) + currentIndex := -1 + for _, upstream := range route.Upstreams { + if !upstreamEnabled(configuration, states, upstream) { + continue + } + if upstream == current { + currentIndex = len(eligible) + } + eligible = append(eligible, upstream) + } + if currentIndex < 0 || len(eligible) < 2 { + return "", false + } + if currentIndex+1 < len(eligible) { + return eligible[currentIndex+1], true + } + switch route.Strategy.EndBehavior { + case "loop": + return eligible[0], eligible[0] != current + case "stayLast", "", "stop": + return "", false + default: + return "", false + } +} + +func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState { + result := make(map[string]adminstate.UpstreamState, len(values)) + for _, value := range values { + result[value.Name] = value + } + return result +} + +func routingStates(values []adminstate.RoutingState) map[string]adminstate.RoutingState { + result := make(map[string]adminstate.RoutingState, len(values)) + for _, value := range values { + result[value.Name] = value + } + return result +} + +func upstreamEnabled(configuration *config.Config, states map[string]adminstate.UpstreamState, name string) bool { + configured, exists := configuration.Upstreams[name] + if !exists || !configured.Enabled { + return false + } + state, exists := states[name] + return exists && state.Enabled +} + +func (coordinator *SequentialCoordinator) clearProcessed(routeName, upstream string) { + delete(coordinator.processed, processedKey(routeName, upstream)) +} + +func processedKey(routeName, upstream string) string { + return routeName + "\x00" + upstream +} + +func requestID(routeName, upstream string, generation uint64) string { + payload := routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10) + digest := sha256.Sum256([]byte(payload)) + return "auto-sequential-" + hex.EncodeToString(digest[:16]) +} + +func validResultClass(class upstream.FetchClass) bool { + switch class { + case upstream.FetchValid, upstream.FetchEmpty, upstream.FetchDuplicateOnly, upstream.FetchError: + return true + default: + return false + } +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/controller/routing/sequential_test.go b/internal/controller/routing/sequential_test.go new file mode 100644 index 0000000..1958af9 --- /dev/null +++ b/internal/controller/routing/sequential_test.go @@ -0,0 +1,274 @@ +package routing + +import ( + "context" + "sync" + "testing" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/controller/provider" + "proxy-pool/internal/domain/adminstate" +) + +func TestSequentialCoordinatorSwitchesOnlyAtConfiguredThreshold(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-a": {UpstreamID: "provider-a", ConsecutiveEmptyFetch: 4, EmptyGeneration: 1}, + }} + refresh := &refreshRecorder{} + coordinator := newCoordinator(t, routingConfiguration("stop", []string{"provider-a", "provider-b"}), state, stats, refresh) + + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 0 || state.current() != "provider-a" { + t.Fatalf("Tick(before threshold) = (%+v, %v), current=%q", result, err, state.current()) + } + stats.set("provider-a", provider.Stats{UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 1}) + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 1 || state.current() != "provider-b" { + t.Fatalf("Tick(at threshold) = (%+v, %v), current=%q", result, err, state.current()) + } + if refresh.count != 1 || state.switches != 1 { + t.Fatalf("refresh=%d switches=%d, want 1", refresh.count, state.switches) + } +} + +func TestSequentialCoordinatorConsumesStatsRecorderNotifications(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) + stats, err := provider.NewStatsRecorder(2) + if err != nil { + t.Fatalf("NewStatsRecorder() = %v", err) + } + refresh := &refreshRecorder{} + coordinator, err := NewSequentialCoordinator( + staticConfiguration{configuration: routingConfiguration("stop", []string{"provider-a", "provider-b"}), revision: 7}, + state, + stats, + refresh, + Options{Now: time.Now}, + ) + if err != nil { + t.Fatalf("NewSequentialCoordinator() = %v", err) + } + stats.AddResultObserver(coordinator) + for range 5 { + stats.Record(provider.Result{UpstreamID: "provider-a", Class: "empty"}) + } + select { + case <-coordinator.notify: + case <-time.After(time.Second): + t.Fatal("StatsRecorder did not notify SequentialCoordinator") + } + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 1 || state.current() != "provider-b" || refresh.count != 1 { + t.Fatalf("Tick() = (%+v, %v), current=%q refresh=%d", result, err, state.current(), refresh.count) + } +} + +func TestSequentialCoordinatorSkipsDisabledCandidatesAndStaysAtConfiguredEnd(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{ + "provider-a": true, "provider-b": false, "provider-c": true, + }) + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-a": {UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 1}, + "provider-c": {UpstreamID: "provider-c", ConsecutiveEmptyFetch: 5, EmptyGeneration: 2}, + }} + coordinator := newCoordinator(t, routingConfiguration("stayLast", []string{"provider-a", "provider-b", "provider-c"}), state, stats, &refreshRecorder{}) + + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 1 || state.current() != "provider-c" { + t.Fatalf("Tick(skip disabled) = (%+v, %v), current=%q", result, err, state.current()) + } + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 0 || state.current() != "provider-c" { + t.Fatalf("Tick(stay last) = (%+v, %v), current=%q", result, err, state.current()) + } +} + +func TestSequentialCoordinatorDoesNotReuseAnEmptyGenerationAfterLoop(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-a": {UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 1}, + "provider-b": {UpstreamID: "provider-b", ConsecutiveEmptyFetch: 5, EmptyGeneration: 2}, + }} + coordinator := newCoordinator(t, routingConfiguration("loop", []string{"provider-a", "provider-b"}), state, stats, &refreshRecorder{}) + + for _, want := range []string{"provider-b", "provider-a"} { + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 1 || state.current() != want { + t.Fatalf("Tick() = (%+v, %v), current=%q, want %q", result, err, state.current(), want) + } + } + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 0 || state.current() != "provider-a" { + t.Fatalf("Tick(reused generation) = (%+v, %v), current=%q", result, err, state.current()) + } + stats.set("provider-a", provider.Stats{UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 3}) + if result, err := coordinator.Tick(context.Background()); err != nil || result.Switched != 1 || state.current() != "provider-b" { + t.Fatalf("Tick(new generation) = (%+v, %v), current=%q", result, err, state.current()) + } +} + +func TestSequentialCoordinatorUsesCompareAndSwapUnderConcurrentTicks(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-a": {UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 1}, + }} + coordinator := newCoordinator(t, routingConfiguration("stop", []string{"provider-a", "provider-b"}), state, stats, &refreshRecorder{}) + + var wait sync.WaitGroup + for range 32 { + wait.Add(1) + go func() { + defer wait.Done() + if _, err := coordinator.Tick(context.Background()); err != nil { + t.Errorf("Tick() = %v", err) + } + }() + } + wait.Wait() + if state.switches != 1 || state.current() != "provider-b" { + t.Fatalf("switches=%d current=%q, want one switch to provider-b", state.switches, state.current()) + } +} + +func TestSequentialCoordinatorFailsClosedWhenConfigurationRevisionIsStale(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) + state.snapshot.Config.Revision = 8 + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-a": {UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 1}, + }} + coordinator := newCoordinator(t, routingConfiguration("stop", []string{"provider-a", "provider-b"}), state, stats, &refreshRecorder{}) + if _, err := coordinator.Tick(context.Background()); err == nil || state.current() != "provider-a" { + t.Fatalf("Tick(stale revision) error=%v current=%q, want error and unchanged state", err, state.current()) + } +} + +func TestSequentialCoordinatorRejectsMalformedStatsAndCoalescesNotifications(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) + configuration := staticConfiguration{configuration: routingConfiguration("stop", []string{"provider-a", "provider-b"}), revision: 7} + coordinator, err := NewSequentialCoordinator(configuration, state, emptyStats{}, &refreshRecorder{}, Options{Now: time.Now}) + if err != nil { + t.Fatalf("NewSequentialCoordinator() = %v", err) + } + if _, err := coordinator.Tick(context.Background()); err == nil { + t.Fatal("Tick() with malformed StatsReader response error = nil") + } + for range 4 { + coordinator.ObserveProviderResult(provider.Result{UpstreamID: "provider-a", Class: "empty"}) + } + select { + case <-coordinator.notify: + default: + t.Fatal("ObserveProviderResult() did not notify") + } + select { + case <-coordinator.notify: + t.Fatal("ObserveProviderResult() did not coalesce notifications") + default: + } +} + +func newCoordinator( + t *testing.T, + configuration *config.Config, + state *routingState, + stats *routingStats, + refresh *refreshRecorder, +) *SequentialCoordinator { + t.Helper() + coordinator, err := NewSequentialCoordinator(staticConfiguration{configuration: configuration, revision: 7}, state, stats, refresh, Options{ + Now: func() time.Time { return time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("NewSequentialCoordinator() = %v", err) + } + return coordinator +} + +func routingConfiguration(endBehavior string, upstreams []string) *config.Config { + configured := make(map[string]config.Upstream, len(upstreams)) + for _, upstream := range upstreams { + configured[upstream] = config.Upstream{Enabled: true} + } + return &config.Config{ + Routing: []config.Routing{{ + Name: "checkout", Enabled: true, Purpose: "gateway", Upstreams: append([]string(nil), upstreams...), + Strategy: config.Strategy{Type: "sequential", SwitchAfterEmptyFetch: 5, EndBehavior: endBehavior}, + }}, + Upstreams: configured, + } +} + +type staticConfiguration struct { + configuration *config.Config + revision uint64 +} + +func (source staticConfiguration) Snapshot() (*config.Config, uint64) { + return source.configuration, source.revision +} + +type routingState struct { + mu sync.Mutex + snapshot adminstate.Snapshot + switches int +} + +func newRoutingState(current string, enabled map[string]bool) *routingState { + upstreams := make([]adminstate.UpstreamState, 0, len(enabled)) + for name, isEnabled := range enabled { + upstreams = append(upstreams, adminstate.UpstreamState{Name: name, Enabled: isEnabled}) + } + return &routingState{snapshot: adminstate.Snapshot{ + Config: &adminstate.ConfigRevision{Revision: 7}, Upstreams: upstreams, + Routings: []adminstate.RoutingState{{Name: "checkout", Enabled: true, CurrentUpstream: current}}, + }} +} + +func (state *routingState) Snapshot(context.Context) (adminstate.Snapshot, error) { + state.mu.Lock() + defer state.mu.Unlock() + return state.snapshot, nil +} + +func (state *routingState) SwitchRouting(_ context.Context, command adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error) { + state.mu.Lock() + defer state.mu.Unlock() + routing := &state.snapshot.Routings[0] + if command.Name != routing.Name || command.ExpectedCurrent != routing.CurrentUpstream { + return adminstate.MutationResult{}, adminstate.ErrConflict + } + routing.CurrentUpstream = command.Target + state.switches++ + return adminstate.MutationResult{RequestID: command.RequestID, Changed: true, Revision: uint64(state.switches)}, nil +} + +func (state *routingState) current() string { + state.mu.Lock() + defer state.mu.Unlock() + return state.snapshot.Routings[0].CurrentUpstream +} + +type routingStats struct { + mu sync.Mutex + byUpstream map[string]provider.Stats +} + +type emptyStats struct{} + +func (emptyStats) ReadProviderStats([]string) []provider.Stats { return nil } + +func (stats *routingStats) ReadProviderStats(upstreams []string) []provider.Stats { + stats.mu.Lock() + defer stats.mu.Unlock() + result := make([]provider.Stats, len(upstreams)) + for index, upstream := range upstreams { + result[index] = stats.byUpstream[upstream] + result[index].UpstreamID = upstream + } + return result +} + +func (stats *routingStats) set(upstream string, value provider.Stats) { + stats.mu.Lock() + defer stats.mu.Unlock() + stats.byUpstream[upstream] = value +} + +type refreshRecorder struct{ count int } + +func (recorder *refreshRecorder) NotifySnapshotRefresh() { recorder.count++ }