feat: reload health scheduling configuration

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 命令也尚未实现。
 
 ## 关键配置与入口
 

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`，实际取二者中

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。

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`,

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))
+	current := configuration.Current()
-	for name, upstream := range configuration.Upstreams {
+	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)
+	if !hasEnabledUpstream {
-	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 {
 		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) {

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

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 ||
+	if planner == nil || !validSchedulerRunnerParts(source, sink, options) {
-		options.BatchSize <= 0 || options.Now == nil {
 		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
 			}
 		}

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
+	inFlight      int
-	candidates []Candidate
+	candidates    []Candidate
-	limit      int
+	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