From c9fa32bd4197d5ad66b997c408a54bf938c379d2 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 18:30:43 +0800 Subject: [PATCH] feat: run bounded health scheduling ticks --- .../controller/health/scheduler_runner.go | 128 ++++++++++++++++++ .../health/scheduler_runner_test.go | 105 ++++++++++++++ 2 files changed, 233 insertions(+) create mode 100644 internal/controller/health/scheduler_runner.go create mode 100644 internal/controller/health/scheduler_runner_test.go diff --git a/internal/controller/health/scheduler_runner.go b/internal/controller/health/scheduler_runner.go new file mode 100644 index 0000000..90fb5de --- /dev/null +++ b/internal/controller/health/scheduler_runner.go @@ -0,0 +1,128 @@ +package health + +import ( + "context" + "errors" + "time" +) + +var ( + ErrInvalidSchedulerRunner = errors.New("invalid health scheduler runner") + ErrInvalidDueSource = errors.New("invalid health due source response") + ErrInvalidTaskSink = errors.New("invalid health task sink response") +) + +// DueSource is implemented by the Controller's future Redis due-index. Both +// calls are bounded and stay outside the Gateway request path. +type DueSource interface { + InFlight(context.Context, time.Time) (int, error) + DueCandidates(context.Context, time.Time, int) ([]Candidate, error) +} + +// TaskSink atomically offers an already bounded batch to the shared leased +// task store. It must leave unaccepted candidates eligible for a later tick. +type TaskSink interface { + Offer(context.Context, []PlannedTask) (int, error) +} + +type SchedulerRunnerOptions struct { + PollInterval time.Duration + BatchSize int + Now func() time.Time +} + +// SchedulerRunner performs one bounded tick at a time. Run has one ticker +// loop, and Tick performs no concurrent fan-out. +type SchedulerRunner struct { + planner *Planner + source DueSource + sink TaskSink + options SchedulerRunnerOptions +} + +func NewSchedulerRunner( + planner *Planner, + source DueSource, + sink TaskSink, + options SchedulerRunnerOptions, +) (*SchedulerRunner, error) { + if planner == nil || nilInterface(source) || nilInterface(sink) || options.PollInterval <= 0 || + options.BatchSize <= 0 || options.Now == nil { + return nil, ErrInvalidSchedulerRunner + } + return &SchedulerRunner{planner: planner, source: source, sink: sink, options: options}, nil +} + +type TickResult struct { + Planned int + Offered int +} + +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 { + return TickResult{}, ErrInvalidSchedulerRunner + } + if err := ctx.Err(); err != nil { + return TickResult{}, err + } + now := runner.options.Now() + if now.IsZero() { + return TickResult{}, ErrInvalidSchedulerRunner + } + inFlight, err := runner.source.InFlight(ctx, now.UTC()) + if err != nil { + return TickResult{}, err + } + if inFlight < 0 { + return TickResult{}, ErrInvalidDueSource + } + if inFlight >= runner.planner.policy.MaxInFlight { + return TickResult{}, nil + } + candidates, err := runner.source.DueCandidates(ctx, now.UTC(), runner.options.BatchSize) + if err != nil { + return TickResult{}, err + } + if len(candidates) > runner.options.BatchSize { + return TickResult{}, ErrInvalidDueSource + } + plans, err := runner.planner.Plan(now.UTC(), inFlight, runner.options.BatchSize, candidates) + if err != nil { + return TickResult{}, err + } + result := TickResult{Planned: len(plans)} + if len(plans) == 0 { + return result, nil + } + offered, err := runner.sink.Offer(ctx, plans) + if err != nil { + return TickResult{}, err + } + if offered < 0 || offered > len(plans) { + return TickResult{}, ErrInvalidTaskSink + } + result.Offered = offered + return result, nil +} + +func (runner *SchedulerRunner) Run(ctx context.Context) error { + if ctx == nil || runner == nil { + return ErrInvalidSchedulerRunner + } + if _, err := runner.Tick(ctx); err != nil { + return err + } + ticker := time.NewTicker(runner.options.PollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + if _, err := runner.Tick(ctx); err != nil { + return err + } + } + } +} diff --git a/internal/controller/health/scheduler_runner_test.go b/internal/controller/health/scheduler_runner_test.go new file mode 100644 index 0000000..c36149f --- /dev/null +++ b/internal/controller/health/scheduler_runner_test.go @@ -0,0 +1,105 @@ +package health + +import ( + "context" + "errors" + "testing" + "time" + + healthDomain "proxy-pool/internal/domain/health" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestSchedulerRunnerUsesBoundedDueBatchAndSharedInFlightCount(t *testing.T) { + planner, err := NewPlanner(SchedulePolicy{ + Interval: time.Minute, MaxInFlight: 3, Timeout: time.Second, MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + now := time.Date(2026, 7, 31, 18, 0, 0, 0, time.UTC) + source := &dueSourceStub{inFlight: 1, candidates: []Candidate{ + {ProxyID: "fetched", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}, + {ProxyID: "available", State: proxyDomain.StateAvailable, Level: healthDomain.LevelBasic, DueAt: now}, + }} + sink := &taskSinkStub{} + runner, err := NewSchedulerRunner(planner, source, sink, SchedulerRunnerOptions{ + PollInterval: time.Second, BatchSize: 2, Now: func() time.Time { return now }, + }) + if err != nil { + t.Fatalf("NewSchedulerRunner(): %v", err) + } + result, err := runner.Tick(context.Background()) + if err != nil || result.Planned != 2 || result.Offered != 2 || source.limit != 2 || len(sink.tasks) != 2 || + sink.tasks[0].Priority != PriorityFetched { + t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + } +} + +func TestSchedulerRunnerSkipsDueReadAtSharedCapacity(t *testing.T) { + planner, err := NewPlanner(SchedulePolicy{ + Interval: time.Minute, MaxInFlight: 1, Timeout: time.Second, MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + source := &dueSourceStub{inFlight: 1} + runner, err := NewSchedulerRunner(planner, source, &taskSinkStub{}, SchedulerRunnerOptions{ + PollInterval: time.Second, BatchSize: 1, Now: time.Now, + }) + if err != nil { + t.Fatalf("NewSchedulerRunner(): %v", err) + } + result, err := runner.Tick(context.Background()) + if err != nil || result != (TickResult{}) || source.limit != 0 { + t.Fatalf("Tick(at capacity) = (%+v, %v); source=%+v", result, err, source) + } +} + +func TestSchedulerRunnerRejectsOversizedSourceAndSinkResponses(t *testing.T) { + planner, err := NewPlanner(SchedulePolicy{ + Interval: time.Minute, MaxInFlight: 2, Timeout: time.Second, MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + now := time.Now() + runner, err := NewSchedulerRunner(planner, &dueSourceStub{candidates: []Candidate{ + {ProxyID: "one", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}, + {ProxyID: "two", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}, + }}, &taskSinkStub{}, SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 1, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewSchedulerRunner(): %v", err) + } + if _, err := runner.Tick(context.Background()); !errors.Is(err, ErrInvalidDueSource) { + t.Fatalf("Tick(oversized source) error = %v, want ErrInvalidDueSource", err) + } +} + +type dueSourceStub struct { + inFlight int + candidates []Candidate + limit int +} + +func (source *dueSourceStub) InFlight(context.Context, time.Time) (int, error) { + return source.inFlight, nil +} + +func (source *dueSourceStub) DueCandidates(_ context.Context, _ time.Time, limit int) ([]Candidate, error) { + source.limit = limit + return source.candidates, nil +} + +type taskSinkStub struct { + tasks []PlannedTask + offer int +} + +func (sink *taskSinkStub) Offer(_ context.Context, tasks []PlannedTask) (int, error) { + sink.tasks = append([]PlannedTask(nil), tasks...) + if sink.offer != 0 { + return sink.offer, nil + } + return len(tasks), nil +}