diff --git a/internal/controller/health/scheduler_runner.go b/internal/controller/health/scheduler_runner.go index c2622f4..2fb92e2 100644 --- a/internal/controller/health/scheduler_runner.go +++ b/internal/controller/health/scheduler_runner.go @@ -295,7 +295,10 @@ func (supervisor *ConfiguredSchedulerSupervisor) upstreamDueGroups( egressURLs []string, targets []configuredTargetProfile, ) ([]DueSource, error) { - groups := []DueSource{basic} + // EGRESS and TARGET due reads lazily materialize references from BASIC due + // entries. Run them before BASIC so a small batch cannot consume the only + // source reference before these independent checks are established. + groups := make([]DueSource, 0, 1+len(egressURLs)+len(targets)) if egress, supported := supervisor.source.(EgressUpstreamTaskSource); supported { for _, targetURL := range egressURLs { source, err := newEgressUpstreamDueSource(egress, upstreamID, targetURL) @@ -314,6 +317,7 @@ func (supervisor *ConfiguredSchedulerSupervisor) upstreamDueGroups( groups = append(groups, source) } } + groups = append(groups, basic) return groups, nil } diff --git a/internal/controller/health/scheduler_runner_test.go b/internal/controller/health/scheduler_runner_test.go index 0e56296..f0fad9d 100644 --- a/internal/controller/health/scheduler_runner_test.go +++ b/internal/controller/health/scheduler_runner_test.go @@ -188,9 +188,50 @@ func TestConfiguredSchedulerSupervisorSchedulesBoundedEgressGroups(t *testing.T) t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) } result, err := supervisor.Tick(context.Background()) - if err != nil || result != (TickResult{Planned: 3, Offered: 3}) || len(sink.tasks) != 1 || - sink.tasks[0].Candidate.Level != healthDomain.LevelEgress || source.sources["provider-a"].egressCalls != 2 { - t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + if err != nil || result != (TickResult{Planned: 3, Offered: 3}) || !sink.offeredLevel(healthDomain.LevelEgress) || + source.sources["provider-a"].egressCalls != 2 { + t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.batches) + } +} + +func TestConfiguredSchedulerSupervisorMaterializesEveryEgressGroupBeforeBasic(t *testing.T) { + now := time.Date(2026, 8, 2, 11, 15, 0, 0, time.UTC) + firstURL := "https://egress-one.example/identity" + secondURL := "https://egress-two.example/identity" + configuration := &config.Config{ + Defaults: config.Defaults{Check: config.Check{ + Interval: config.Duration(time.Minute), MaxInFlight: 1, Timeout: config.Duration(time.Second), MaxAttempts: 1, + URLs: []string{firstURL, secondURL}, + }}, + Upstreams: map[string]config.Upstream{"provider-a": {Enabled: true}}, + } + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": {candidates: []Candidate{{ + ProxyID: "basic", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now, + }}, egressCandidates: map[string][]Candidate{ + firstURL: {{ProxyID: "egress-one", State: proxyDomain.StateFetched, Level: healthDomain.LevelEgress, TargetURL: firstURL, DueAt: now}}, + secondURL: {{ProxyID: "egress-two", State: proxyDomain.StateFetched, Level: healthDomain.LevelEgress, TargetURL: secondURL, DueAt: now}}, + }}, + }} + sink := &taskSinkStub{} + supervisor, err := NewConfiguredSchedulerSupervisor(&configurationSourceStub{configuration: configuration}, source, sink, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 1, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + for tick := 0; tick < 2; tick++ { + if result, err := supervisor.Tick(context.Background()); err != nil || result != (TickResult{Planned: 1, Offered: 1}) { + t.Fatalf("Tick(%d) = (%+v, %v)", tick, result, err) + } + } + item := source.sources["provider-a"] + if item.egressCalls != 2 || item.dueCalls != 0 || len(sink.batches) != 2 || + sink.batches[0][0].Candidate.Level != healthDomain.LevelEgress || sink.batches[1][0].Candidate.Level != healthDomain.LevelEgress { + t.Fatalf("first two ticks did not establish egress groups: source=%+v batches=%+v", item, sink.batches) + } + if result, err := supervisor.Tick(context.Background()); err != nil || result != (TickResult{Planned: 1, Offered: 1}) || + item.dueCalls != 1 || len(sink.batches) != 3 || sink.batches[2][0].Candidate.Level != healthDomain.LevelBasic { + t.Fatalf("basic Tick() = (%+v, %v), source=%+v batches=%+v", result, err, item, sink.batches) } } @@ -225,10 +266,9 @@ func TestConfiguredSchedulerSupervisorSchedulesRoutingTargetProfiles(t *testing. t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) } result, err := supervisor.Tick(context.Background()) - if err != nil || result != (TickResult{Planned: 2, Offered: 2}) || len(sink.tasks) != 1 || - sink.tasks[0].Candidate.Level != healthDomain.LevelTarget || sink.tasks[0].Candidate.RoutingName != routingName || + if err != nil || result != (TickResult{Planned: 2, Offered: 2}) || !sink.offeredTarget(routingName, targetURL) || source.sources["provider-a"].targetCalls != 1 { - t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.batches) } } @@ -359,14 +399,39 @@ func targetCandidateKey(routingName, targetURL string) string { } type taskSinkStub struct { - tasks []PlannedTask - offer int + tasks []PlannedTask + batches [][]PlannedTask + offer int } func (sink *taskSinkStub) Offer(_ context.Context, tasks []PlannedTask) (int, error) { sink.tasks = append([]PlannedTask(nil), tasks...) + sink.batches = append(sink.batches, append([]PlannedTask(nil), tasks...)) if sink.offer != 0 { return sink.offer, nil } return len(tasks), nil } + +func (sink *taskSinkStub) offeredLevel(level healthDomain.Level) bool { + for _, batch := range sink.batches { + for _, task := range batch { + if task.Candidate.Level == level { + return true + } + } + } + return false +} + +func (sink *taskSinkStub) offeredTarget(routingName, targetURL string) bool { + for _, batch := range sink.batches { + for _, task := range batch { + if task.Candidate.Level == healthDomain.LevelTarget && task.Candidate.RoutingName == routingName && + task.Candidate.TargetURL == targetURL { + return true + } + } + } + return false +} diff --git a/progress.md b/progress.md index 4b9a920..c63cfeb 100644 --- a/progress.md +++ b/progress.md @@ -19,6 +19,8 @@ 前重建并下发递增版本，Gateway 在同一长连接内原子替换视图；构建失败或流中断时才由 SessionSupervisor 按原有退避重连。该机制复用 `RefreshingSnapshotSource`，不降低 Delta 的 fail-closed 校验，也不把自动 Drain/ACK 记为已完成。 +- Health Scheduler 现在先建立 EGRESS 与 Routing TARGET 的独立 due 引用，再投递 + BASIC 任务；在单项小批次下，短 TTL Proxy 不会因 BASIC 先出队而错过首次出口/目标探测。 ## 2026-07-30