diff --git a/README.md b/README.md index 6fb5306..07e149c 100644 --- a/README.md +++ b/README.md @@ -168,7 +168,9 @@ Checker 的参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、 `PROXY_POOL_CHECKER_ID`、`PROXY_POOL_CHECKER_INSTANCE_ID` 与 `PROXY_POOL_CHECKER_MAX_IN_FLIGHT` 提供。它不会访问 Redis/PostgreSQL；生产 Controller 在启用控制面时装配 Redis 共享任务 broker，并按启用的 Upstream 调度 -HTTP/HTTPS BASIC 检查。EGRESS 与 TARGET 尚未进入生产调度，loadgen 命令也尚未实现。 +HTTP/HTTPS BASIC 检查。调度监督器每轮读取已发布配置，因此 reload 后的上游启停、 +检查间隔、抖动、超时、重试次数和 `maxInFlight` 会在下一轮生效；新启用的上游无需 +重启 Controller。EGRESS 与 TARGET 尚未进入生产调度，loadgen 命令也尚未实现。 ## 关键配置与入口 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index adbfefa..6968dfe 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -202,7 +202,9 @@ Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-pla `PROXY_POOL_*` 环境变量提供。mTLS 模式下该命令读取 `checkerTLS`，明文 fixture 模式只接受回环 Controller 地址。Checker 只从 gRPC 领取任务并批量上报事实，不读取 Redis/PostgreSQL；Controller 在生产启动拓扑中装配 Redis 共享任务队列，当前调度 -HTTP/HTTPS BASIC 检查。EGRESS 和 TARGET 的生产调度仍在后续实施范围。 +HTTP/HTTPS BASIC 检查。调度监督器在每轮从已发布配置读取启用的上游；Admin reload +发布后，上游启停和有效 `check` 策略会在下一轮生效，新启用的上游无需重启 Controller。 +EGRESS 和 TARGET 的生产调度仍在后续实施范围。 `maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway 在本地维护容量为 `65536` 的非阻塞 Outcome 队列，默认微批上限为 `512`，实际取二者中 diff --git a/docs/design/architecture.md b/docs/design/architecture.md index e1459da..1ec159a 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -436,6 +436,8 @@ flowchart LR 热更新规则： - 删除/禁用 Upstream：停止 Fetch 和新分配，已有 Proxy/连接 Drain。 +- Checker 调度监督器在下一轮停止禁用 Upstream 的新任务，并发现新启用 Upstream；有效 + `check` 策略在同一轮重新解析，已有租约由租约期自然收敛。 - 减小 maxSize：不强杀连接，停止补池并按策略自然缩容。 - Routing 立即对新请求生效；旧请求持有旧 Snapshot 完成。 - 修改 API 地址或凭据版本会重建 Provider Adapter，但不会把错误计成 Empty。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index c78a0a0..3df1ec8 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -289,6 +289,10 @@ EGRESS 已具备任务 URL 传输、HTTP/HTTPS 探测和全局事实回传契约 SOCKS5、EGRESS/TARGET 多维任务索引及部署运行态仍未实现， 因此本任务保持未完成。 +补充进度（2026-08-02）：BASIC 调度已改为配置驱动监督器。它每轮读取已发布快照并复用 +有界派发逻辑，所以 reload 后已启用上游的策略变更、停用，以及新启用上游都无需重启 +Controller 即可生效；Redis 任务存储仍仅承载 BASIC，未扩展 EGRESS/TARGET 的多维索引。 + ## Task 12: Machine-readable Contracts **Files:** `api/openapi/proxy-pool.yaml`, `api/proto/controlplane/v1/controlplane.proto`, diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index a06a897..9aede1a 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -307,7 +307,7 @@ func runWithWorkerFactory( } runners = append(runners, runner) if tasks, ok := opened.activity.(healthTaskRuntime); ok && !nilInterface(tasks) { - schedulers, schedulerErr := newHealthSchedulers(loaded.Value, tasks, options.Now) + schedulers, schedulerErr := newHealthSchedulers(configurationStore, tasks, options.Now) if schedulerErr != nil { return fmt.Errorf("%w: build Checker health schedulers: %w", ErrStartup, schedulerErr) } @@ -323,50 +323,34 @@ func runWithWorkerFactory( } func newHealthSchedulers( - configuration *config.Config, + configuration controllerHealth.ConfigurationSource, tasks healthTaskRuntime, now func() time.Time, ) ([]lifecycle.Runner, error) { - if configuration == nil || nilInterface(tasks) || now == nil { + if nilInterface(configuration) || nilInterface(tasks) || now == nil { return nil, ErrInvalidOptions } - names := make([]string, 0, len(configuration.Upstreams)) - for name, upstream := range configuration.Upstreams { + current := configuration.Current() + if current == nil { + return nil, ErrInvalidOptions + } + hasEnabledUpstream := false + for _, upstream := range current.Upstreams { if upstream.Enabled { - names = append(names, name) + hasEnabledUpstream = true + break } } - sort.Strings(names) - runners := make([]lifecycle.Runner, 0, len(names)) - for _, name := range names { - check := config.EffectiveCheck(configuration.Defaults.Check, configuration.Upstreams[name].Check) - planner, err := controllerHealth.NewPlanner(controllerHealth.SchedulePolicy{ - Interval: check.Interval.Value(), Jitter: check.Jitter, MaxInFlight: check.MaxInFlight, - Timeout: check.Timeout.Value(), MaxAttempts: check.MaxAttempts, - }) - if err != nil { - return nil, err - } - source, err := controllerHealth.NewUpstreamDueSource(tasks, name) - if err != nil { - return nil, err - } - batchSize := checkSchedulerBatchSize - if check.MaxInFlight < batchSize { - batchSize = check.MaxInFlight - } - runner, err := controllerHealth.NewSchedulerRunner(planner, source, tasks, controllerHealth.SchedulerRunnerOptions{ - PollInterval: checkSchedulerPollInterval, BatchSize: batchSize, Now: now, - }) - if err != nil { - return nil, err - } - runners = append(runners, runner) - } - if len(runners) == 0 { + if !hasEnabledUpstream { return nil, ErrInvalidOptions } - return runners, nil + supervisor, err := controllerHealth.NewConfiguredSchedulerSupervisor(configuration, tasks, tasks, controllerHealth.SchedulerRunnerOptions{ + PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: now, + }) + if err != nil { + return nil, err + } + return []lifecycle.Runner{supervisor}, nil } func newCheckerIdentity(controlPlane config.ControlPlane) (controllerHealth.CheckerIdentityAuthorizer, error) { diff --git a/internal/controller/bootstrap/bootstrap_test.go b/internal/controller/bootstrap/bootstrap_test.go index 457ebaf..31cda9b 100644 --- a/internal/controller/bootstrap/bootstrap_test.go +++ b/internal/controller/bootstrap/bootstrap_test.go @@ -319,7 +319,11 @@ func TestNewHealthSchedulersCreatesOneRunnerPerEnabledUpstream(t *testing.T) { disabled := configuration.Upstreams["provider-b"] disabled.Enabled = false configuration.Upstreams["provider-b"] = disabled - runners, err := newHealthSchedulers(configuration, healthTaskRuntimeStub{}, time.Now) + store, err := config.NewStore(configuration) + if err != nil { + t.Fatalf("config.NewStore(): %v", err) + } + runners, err := newHealthSchedulers(store, healthTaskRuntimeStub{}, time.Now) if err != nil || len(runners) != 1 { t.Fatalf("newHealthSchedulers() = (%d runners, %v)", len(runners), err) } diff --git a/internal/controller/health/scheduler_runner.go b/internal/controller/health/scheduler_runner.go index 8df67b3..c6c13a5 100644 --- a/internal/controller/health/scheduler_runner.go +++ b/internal/controller/health/scheduler_runner.go @@ -3,7 +3,11 @@ package health import ( "context" "errors" + "sort" + "strings" "time" + + "proxy-pool/internal/config" ) var ( @@ -73,13 +77,64 @@ func NewSchedulerRunner( sink TaskSink, options SchedulerRunnerOptions, ) (*SchedulerRunner, error) { - if planner == nil || nilInterface(source) || nilInterface(sink) || options.PollInterval <= 0 || - options.BatchSize <= 0 || options.Now == nil { + if planner == nil || !validSchedulerRunnerParts(source, sink, options) { return nil, ErrInvalidSchedulerRunner } return &SchedulerRunner{planner: planner, source: source, sink: sink, options: options}, nil } +// ConfiguredSchedulerRunner resolves the current effective upstream check +// policy on every tick. It keeps a running Controller aligned with validated +// config reloads without putting configuration reads in the Gateway path. +type ConfiguredSchedulerRunner struct { + configuration ConfigurationSource + upstreamID string + source DueSource + sink TaskSink + options SchedulerRunnerOptions +} + +// ConfiguredSchedulerSupervisor discovers enabled upstreams from each +// published configuration snapshot. It lets newly enabled upstreams begin +// bounded BASIC scheduling without a Controller restart. +type ConfiguredSchedulerSupervisor struct { + configuration ConfigurationSource + source UpstreamTaskSource + sink TaskSink + options SchedulerRunnerOptions +} + +func NewConfiguredSchedulerRunner( + configuration ConfigurationSource, + upstreamID string, + source DueSource, + sink TaskSink, + options SchedulerRunnerOptions, +) (*ConfiguredSchedulerRunner, error) { + if nilInterface(configuration) || strings.TrimSpace(upstreamID) != upstreamID || upstreamID == "" || + !validSchedulerRunnerParts(source, sink, options) { + return nil, ErrInvalidSchedulerRunner + } + return &ConfiguredSchedulerRunner{ + configuration: configuration, upstreamID: upstreamID, source: source, sink: sink, options: options, + }, nil +} + +func NewConfiguredSchedulerSupervisor( + configuration ConfigurationSource, + source UpstreamTaskSource, + sink TaskSink, + options SchedulerRunnerOptions, +) (*ConfiguredSchedulerSupervisor, error) { + if nilInterface(configuration) || nilInterface(source) || nilInterface(sink) || options.PollInterval <= 0 || + options.BatchSize <= 0 || options.Now == nil { + return nil, ErrInvalidSchedulerRunner + } + return &ConfiguredSchedulerSupervisor{ + configuration: configuration, source: source, sink: sink, options: options, + }, nil +} + type TickResult struct { Planned int Offered int @@ -87,34 +142,101 @@ type TickResult struct { func (runner *SchedulerRunner) Tick(ctx context.Context) (TickResult, error) { if ctx == nil || runner == nil || runner.planner == nil || nilInterface(runner.source) || - nilInterface(runner.sink) || runner.options.Now == nil { + !validSchedulerRunnerParts(runner.source, runner.sink, runner.options) { + return TickResult{}, ErrInvalidSchedulerRunner + } + return tickWithPlanner(ctx, runner.planner, runner.source, runner.sink, runner.options) +} + +func (runner *ConfiguredSchedulerRunner) Tick(ctx context.Context) (TickResult, error) { + if ctx == nil || runner == nil || nilInterface(runner.configuration) || runner.upstreamID == "" || + !validSchedulerRunnerParts(runner.source, runner.sink, runner.options) { return TickResult{}, ErrInvalidSchedulerRunner } if err := ctx.Err(); err != nil { return TickResult{}, err } - now := runner.options.Now() + planner, configured, err := configuredPlanner(runner.configuration, runner.upstreamID) + if err != nil { + return TickResult{}, err + } + if !configured { + return TickResult{}, nil + } + return tickWithPlanner(ctx, planner, runner.source, runner.sink, schedulerOptionsForPlanner(runner.options, planner)) +} + +func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (TickResult, error) { + if ctx == nil || supervisor == nil || nilInterface(supervisor.configuration) || nilInterface(supervisor.source) || + nilInterface(supervisor.sink) || supervisor.options.PollInterval <= 0 || supervisor.options.BatchSize <= 0 || supervisor.options.Now == nil { + return TickResult{}, ErrInvalidSchedulerRunner + } + if err := ctx.Err(); err != nil { + return TickResult{}, err + } + configuration := supervisor.configuration.Current() + if configuration == nil { + return TickResult{}, ErrInvalidSchedulerRunner + } + names := enabledUpstreamNames(configuration) + result := TickResult{} + for _, name := range names { + planner, configured, err := configuredPlannerForConfig(configuration, name) + if err != nil { + return TickResult{}, err + } + if !configured { + continue + } + source, err := NewUpstreamDueSource(supervisor.source, name) + if err != nil { + return TickResult{}, err + } + tick, err := tickWithPlanner(ctx, planner, source, supervisor.sink, schedulerOptionsForPlanner(supervisor.options, planner)) + if err != nil { + return TickResult{}, err + } + result.Planned += tick.Planned + result.Offered += tick.Offered + } + return result, nil +} + +func tickWithPlanner( + ctx context.Context, + planner *Planner, + source DueSource, + sink TaskSink, + options SchedulerRunnerOptions, +) (TickResult, error) { + if ctx == nil || planner == nil || !validSchedulerRunnerParts(source, sink, options) { + return TickResult{}, ErrInvalidSchedulerRunner + } + if err := ctx.Err(); err != nil { + return TickResult{}, err + } + now := options.Now() if now.IsZero() { return TickResult{}, ErrInvalidSchedulerRunner } - inFlight, err := runner.source.InFlight(ctx, now.UTC()) + inFlight, err := source.InFlight(ctx, now.UTC()) if err != nil { return TickResult{}, err } if inFlight < 0 { return TickResult{}, ErrInvalidDueSource } - if inFlight >= runner.planner.policy.MaxInFlight { + if inFlight >= planner.policy.MaxInFlight { return TickResult{}, nil } - candidates, err := runner.source.DueCandidates(ctx, now.UTC(), runner.options.BatchSize) + candidates, err := source.DueCandidates(ctx, now.UTC(), options.BatchSize) if err != nil { return TickResult{}, err } - if len(candidates) > runner.options.BatchSize { + if len(candidates) > options.BatchSize { return TickResult{}, ErrInvalidDueSource } - plans, err := runner.planner.Plan(now.UTC(), inFlight, runner.options.BatchSize, candidates) + plans, err := planner.Plan(now.UTC(), inFlight, options.BatchSize, candidates) if err != nil { return TickResult{}, err } @@ -122,7 +244,7 @@ func (runner *SchedulerRunner) Tick(ctx context.Context) (TickResult, error) { if len(plans) == 0 { return result, nil } - offered, err := runner.sink.Offer(ctx, plans) + offered, err := sink.Offer(ctx, plans) if err != nil { return TickResult{}, err } @@ -134,20 +256,90 @@ func (runner *SchedulerRunner) Tick(ctx context.Context) (TickResult, error) { } func (runner *SchedulerRunner) Run(ctx context.Context) error { - if ctx == nil || runner == nil { + if runner == nil { return ErrInvalidSchedulerRunner } - if _, err := runner.Tick(ctx); err != nil { + return runScheduler(ctx, runner.options.PollInterval, runner.Tick) +} + +func (runner *ConfiguredSchedulerRunner) Run(ctx context.Context) error { + if runner == nil { + return ErrInvalidSchedulerRunner + } + return runScheduler(ctx, runner.options.PollInterval, runner.Tick) +} + +func (supervisor *ConfiguredSchedulerSupervisor) Run(ctx context.Context) error { + if supervisor == nil { + return ErrInvalidSchedulerRunner + } + return runScheduler(ctx, supervisor.options.PollInterval, supervisor.Tick) +} + +func validSchedulerRunnerParts(source DueSource, sink TaskSink, options SchedulerRunnerOptions) bool { + return !nilInterface(source) && !nilInterface(sink) && options.PollInterval > 0 && options.BatchSize > 0 && options.Now != nil +} + +func configuredPlanner(configuration ConfigurationSource, upstreamID string) (*Planner, bool, error) { + current := configuration.Current() + if current == nil { + return nil, false, ErrInvalidSchedulerRunner + } + return configuredPlannerForConfig(current, upstreamID) +} + +func configuredPlannerForConfig(current *config.Config, upstreamID string) (*Planner, bool, error) { + if current == nil { + return nil, false, ErrInvalidSchedulerRunner + } + upstream, exists := current.Upstreams[upstreamID] + if !exists || !upstream.Enabled { + return nil, false, nil + } + check := config.EffectiveCheck(current.Defaults.Check, upstream.Check) + planner, err := NewPlanner(SchedulePolicy{ + Interval: check.Interval.Value(), Jitter: check.Jitter, MaxInFlight: check.MaxInFlight, + Timeout: check.Timeout.Value(), MaxAttempts: check.MaxAttempts, + }) + if err != nil { + return nil, false, err + } + return planner, true, nil +} + +func schedulerOptionsForPlanner(options SchedulerRunnerOptions, planner *Planner) SchedulerRunnerOptions { + if planner.policy.MaxInFlight < options.BatchSize { + options.BatchSize = planner.policy.MaxInFlight + } + return options +} + +func enabledUpstreamNames(configuration *config.Config) []string { + names := make([]string, 0, len(configuration.Upstreams)) + for name, upstream := range configuration.Upstreams { + if upstream.Enabled { + names = append(names, name) + } + } + sort.Strings(names) + return names +} + +func runScheduler(ctx context.Context, pollInterval time.Duration, tick func(context.Context) (TickResult, error)) error { + if ctx == nil || pollInterval <= 0 || tick == nil { + return ErrInvalidSchedulerRunner + } + if _, err := tick(ctx); err != nil { return err } - ticker := time.NewTicker(runner.options.PollInterval) + ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return ctx.Err() case <-ticker.C: - if _, err := runner.Tick(ctx); err != nil { + if _, err := tick(ctx); err != nil { return err } } diff --git a/internal/controller/health/scheduler_runner_test.go b/internal/controller/health/scheduler_runner_test.go index c36149f..9f7a1f0 100644 --- a/internal/controller/health/scheduler_runner_test.go +++ b/internal/controller/health/scheduler_runner_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "proxy-pool/internal/config" healthDomain "proxy-pool/internal/domain/health" proxyDomain "proxy-pool/internal/domain/proxy" ) @@ -76,21 +77,132 @@ func TestSchedulerRunnerRejectsOversizedSourceAndSinkResponses(t *testing.T) { } } +func TestConfiguredSchedulerRunnerFollowsUpstreamConfigChanges(t *testing.T) { + now := time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC) + configuration := &config.Config{ + Defaults: config.Defaults{Check: config.Check{ + Interval: config.Duration(time.Minute), + MaxInFlight: 2, + Timeout: config.Duration(time.Second), + MaxAttempts: 1, + }}, + Upstreams: map[string]config.Upstream{ + "provider-a": {Enabled: true}, + }, + } + source := &dueSourceStub{candidates: []Candidate{{ + ProxyID: "proxy-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now, + }}} + sink := &taskSinkStub{} + runner, err := NewConfiguredSchedulerRunner(&configurationSourceStub{configuration: configuration}, "provider-a", source, sink, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 16, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredSchedulerRunner(): %v", err) + } + + result, err := runner.Tick(context.Background()) + if err != nil || result != (TickResult{Planned: 1, Offered: 1}) || len(sink.tasks) != 1 || sink.tasks[0].Attempts != 1 || + source.limit != 2 { + t.Fatalf("initial Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + } + + updated := configuration.Upstreams["provider-a"] + updated.Check.MaxAttempts = 3 + configuration.Upstreams["provider-a"] = updated + result, err = runner.Tick(context.Background()) + if err != nil || result != (TickResult{Planned: 1, Offered: 1}) || len(sink.tasks) != 1 || sink.tasks[0].Attempts != 3 { + t.Fatalf("updated Tick() = (%+v, %v); sink=%+v", result, err, sink.tasks) + } + + updated.Enabled = false + configuration.Upstreams["provider-a"] = updated + inFlightCalls, dueCalls := source.inFlightCalls, source.dueCalls + result, err = runner.Tick(context.Background()) + if err != nil || result != (TickResult{}) || source.inFlightCalls != inFlightCalls || source.dueCalls != dueCalls { + t.Fatalf("disabled Tick() = (%+v, %v); source=%+v", result, err, source) + } +} + +func TestConfiguredSchedulerSupervisorDiscoversNewEnabledUpstream(t *testing.T) { + now := time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC) + configuration := &config.Config{ + Defaults: config.Defaults{Check: config.Check{ + Interval: config.Duration(time.Minute), + MaxInFlight: 2, + Timeout: config.Duration(time.Second), + MaxAttempts: 1, + }}, + Upstreams: map[string]config.Upstream{ + "provider-a": {Enabled: true}, + }, + } + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": {candidates: []Candidate{{ + ProxyID: "proxy-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now, + }}}, + "provider-b": {candidates: []Candidate{{ + ProxyID: "proxy-b", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now, + }}}, + }} + sink := &taskSinkStub{} + supervisor, err := NewConfiguredSchedulerSupervisor(&configurationSourceStub{configuration: configuration}, source, sink, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 16, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor(): %v", err) + } + if result, err := supervisor.Tick(context.Background()); err != nil || result != (TickResult{Planned: 1, Offered: 1}) || + source.sources["provider-a"].dueCalls != 1 || source.sources["provider-b"].dueCalls != 0 { + t.Fatalf("initial Tick() = (%+v, %v); source=%+v", result, err, source) + } + + configuration.Upstreams["provider-a"] = config.Upstream{Enabled: false} + configuration.Upstreams["provider-b"] = config.Upstream{Enabled: true, Check: config.Check{MaxAttempts: 3, MaxInFlight: 1}} + result, err := supervisor.Tick(context.Background()) + if err != nil || result != (TickResult{Planned: 1, Offered: 1}) || source.sources["provider-b"].dueCalls != 1 || + len(sink.tasks) != 1 || sink.tasks[0].Candidate.ProxyID != "proxy-b" || sink.tasks[0].Attempts != 3 || source.sources["provider-b"].limit != 1 { + t.Fatalf("updated Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + } +} + type dueSourceStub struct { - inFlight int - candidates []Candidate - limit int + inFlight int + candidates []Candidate + limit int + inFlightCalls int + dueCalls int } func (source *dueSourceStub) InFlight(context.Context, time.Time) (int, error) { + source.inFlightCalls++ return source.inFlight, nil } func (source *dueSourceStub) DueCandidates(_ context.Context, _ time.Time, limit int) ([]Candidate, error) { + source.dueCalls++ source.limit = limit return source.candidates, nil } +type upstreamTaskSourceStub struct { + sources map[string]*dueSourceStub +} + +func (source *upstreamTaskSourceStub) InFlightForUpstream(ctx context.Context, upstreamID string, now time.Time) (int, error) { + item, exists := source.sources[upstreamID] + if !exists { + return 0, errors.New("missing upstream source") + } + return item.InFlight(ctx, now) +} + +func (source *upstreamTaskSourceStub) DueCandidatesForUpstream(ctx context.Context, upstreamID string, now time.Time, limit int) ([]Candidate, error) { + item, exists := source.sources[upstreamID] + if !exists { + return nil, errors.New("missing upstream source") + } + return item.DueCandidates(ctx, now, limit) +} + type taskSinkStub struct { tasks []PlannedTask offer int