diff --git a/internal/controller/health/scheduler.go b/internal/controller/health/scheduler.go new file mode 100644 index 0000000..a6c324f --- /dev/null +++ b/internal/controller/health/scheduler.go @@ -0,0 +1,186 @@ +package health + +import ( + "crypto/sha256" + "encoding/binary" + "errors" + "sort" + "strings" + "time" + + healthDomain "proxy-pool/internal/domain/health" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +var ( + ErrInvalidSchedulePolicy = errors.New("invalid health schedule policy") + ErrInvalidScheduleRequest = errors.New("invalid health schedule request") +) + +// SchedulePolicy is scoped to one effective upstream Check configuration. +// MaxInFlight is enforced by the eventual shared task store; Planner uses the +// observed count only to ensure one planning pass never exceeds that ceiling. +type SchedulePolicy struct { + Interval time.Duration + Jitter int + MaxInFlight int + Timeout time.Duration + MaxAttempts int +} + +// Candidate is a bounded record supplied by a due-index query. The Planner +// never scans proxies and never starts a goroutine for a candidate. +type Candidate struct { + ProxyID string + State proxyDomain.State + Level healthDomain.Level + RoutingName string + TargetURL string + DueAt time.Time +} + +type Priority uint8 + +const ( + PriorityFetched Priority = iota + 1 + PrioritySuspect + PriorityUnhealthy + PriorityAvailable +) + +// PlannedTask is transport-neutral work ready for a leased broker to assign +// to a Checker. The task ID and proxy credential material are added only by +// the Controller's broker after it atomically claims the task. +type PlannedTask struct { + Candidate Candidate + Priority Priority + Deadline time.Time + Attempts int +} + +// Planner is stateless and safe for concurrent callers. Its lack of internal +// queues makes maxInFlight and batch bounds explicit at the storage boundary. +type Planner struct { + policy SchedulePolicy +} + +func NewPlanner(policy SchedulePolicy) (*Planner, error) { + if policy.Interval <= 0 || policy.Jitter < 0 || policy.Jitter > 100 || policy.MaxInFlight <= 0 || + policy.Timeout <= 0 || policy.MaxAttempts <= 0 { + return nil, ErrInvalidSchedulePolicy + } + return &Planner{policy: policy}, nil +} + +// NextDue derives a stable, symmetric jitter from the check identity. A +// process restart therefore does not synchronize all proxy checks into one +// burst, while the same proxy/profile remains predictably distributed. +func (planner *Planner) NextDue(checkedAt time.Time, identity string) (time.Time, error) { + if planner == nil || checkedAt.IsZero() || strings.TrimSpace(identity) != identity || identity == "" { + return time.Time{}, ErrInvalidScheduleRequest + } + base := checkedAt.UTC().Add(planner.policy.Interval) + if planner.policy.Jitter == 0 { + return base, nil + } + interval := int64(planner.policy.Interval) + span := (interval/100)*int64(planner.policy.Jitter) + (interval%100)*int64(planner.policy.Jitter)/100 + if span <= 0 { + return base, nil + } + digest := sha256.Sum256([]byte(identity)) + value := binary.BigEndian.Uint64(digest[:8]) + offset := time.Duration(value % uint64(span+1)) + if value&1 == 0 { + return base.Add(offset), nil + } + return base.Add(-offset), nil +} + +// Plan returns at most min(maxTasks, MaxInFlight-inFlight) overdue tasks. +// Fetched inventory is prioritized over suspect inventory, followed by normal +// available inventory, so recovery work does not starve first-use validation. +func (planner *Planner) Plan(now time.Time, inFlight, maxTasks int, candidates []Candidate) ([]PlannedTask, error) { + if planner == nil || now.IsZero() || inFlight < 0 || maxTasks <= 0 { + return nil, ErrInvalidScheduleRequest + } + if inFlight >= planner.policy.MaxInFlight || len(candidates) == 0 { + return []PlannedTask{}, nil + } + limit := planner.policy.MaxInFlight - inFlight + if maxTasks < limit { + limit = maxTasks + } + eligible := make([]plannedCandidate, 0, min(limit, len(candidates))) + for _, candidate := range candidates { + priority, include, err := planner.validateCandidate(candidate) + if err != nil { + return nil, err + } + if !include || candidate.DueAt.After(now) { + continue + } + eligible = append(eligible, plannedCandidate{candidate: candidate, priority: priority}) + } + sort.Slice(eligible, func(left, right int) bool { + if eligible[left].priority != eligible[right].priority { + return eligible[left].priority < eligible[right].priority + } + if !eligible[left].candidate.DueAt.Equal(eligible[right].candidate.DueAt) { + return eligible[left].candidate.DueAt.Before(eligible[right].candidate.DueAt) + } + return candidateIdentity(eligible[left].candidate) < candidateIdentity(eligible[right].candidate) + }) + if len(eligible) > limit { + eligible = eligible[:limit] + } + deadline := now.UTC().Add(planner.policy.Timeout) + result := make([]PlannedTask, len(eligible)) + for index, item := range eligible { + result[index] = PlannedTask{ + Candidate: item.candidate, Priority: item.priority, Deadline: deadline, Attempts: planner.policy.MaxAttempts, + } + } + return result, nil +} + +type plannedCandidate struct { + candidate Candidate + priority Priority +} + +func (planner *Planner) validateCandidate(candidate Candidate) (Priority, bool, error) { + if strings.TrimSpace(candidate.ProxyID) != candidate.ProxyID || candidate.ProxyID == "" || candidate.DueAt.IsZero() { + return 0, false, ErrInvalidScheduleRequest + } + switch candidate.Level { + case healthDomain.LevelBasic, healthDomain.LevelEgress: + if candidate.RoutingName != "" || candidate.TargetURL != "" { + return 0, false, ErrInvalidScheduleRequest + } + case healthDomain.LevelTarget: + if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: candidate.RoutingName, TargetURL: candidate.TargetURL, + }); err != nil { + return 0, false, ErrInvalidScheduleRequest + } + default: + return 0, false, ErrInvalidScheduleRequest + } + switch candidate.State { + case proxyDomain.StateFetched: + return PriorityFetched, true, nil + case proxyDomain.StateSuspect: + return PrioritySuspect, true, nil + case proxyDomain.StateUnhealthy: + return PriorityUnhealthy, true, nil + case proxyDomain.StateAvailable: + return PriorityAvailable, true, nil + default: + return 0, false, nil + } +} + +func candidateIdentity(candidate Candidate) string { + return candidate.ProxyID + "\x00" + string(candidate.Level) + "\x00" + candidate.RoutingName + "\x00" + candidate.TargetURL +} diff --git a/internal/controller/health/scheduler_test.go b/internal/controller/health/scheduler_test.go new file mode 100644 index 0000000..bb36afe --- /dev/null +++ b/internal/controller/health/scheduler_test.go @@ -0,0 +1,88 @@ +package health + +import ( + "errors" + "testing" + "time" + + healthDomain "proxy-pool/internal/domain/health" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestPlannerPrioritizesAndBoundsOverdueCandidates(t *testing.T) { + planner, err := NewPlanner(SchedulePolicy{ + Interval: time.Minute, Jitter: 20, MaxInFlight: 3, Timeout: 4 * time.Second, MaxAttempts: 2, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + now := time.Date(2026, 7, 31, 17, 0, 0, 0, time.UTC) + plans, err := planner.Plan(now, 1, 10, []Candidate{ + {ProxyID: "available", State: proxyDomain.StateAvailable, Level: healthDomain.LevelBasic, DueAt: now.Add(-time.Second)}, + {ProxyID: "suspect", State: proxyDomain.StateSuspect, Level: healthDomain.LevelEgress, DueAt: now.Add(-2 * time.Second)}, + {ProxyID: "fetched", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now.Add(-3 * time.Second)}, + {ProxyID: "future", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now.Add(time.Second)}, + }) + if err != nil || len(plans) != 2 { + t.Fatalf("Plan() = (%+v, %v)", plans, err) + } + if plans[0].Candidate.ProxyID != "fetched" || plans[0].Priority != PriorityFetched || + plans[1].Candidate.ProxyID != "suspect" || plans[1].Priority != PrioritySuspect || + plans[0].Deadline != now.Add(4*time.Second) || plans[0].Attempts != 2 { + t.Fatalf("plans = %+v", plans) + } +} + +func TestPlannerSchedulesUnhealthyForRecovery(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) + } + now := time.Date(2026, 7, 31, 17, 0, 0, 0, time.UTC) + plans, err := planner.Plan(now, 0, 1, []Candidate{{ + ProxyID: "unhealthy", State: proxyDomain.StateUnhealthy, Level: healthDomain.LevelBasic, DueAt: now, + }}) + if err != nil || len(plans) != 1 || plans[0].Priority != PriorityUnhealthy { + t.Fatalf("Plan(unhealthy) = (%+v, %v)", plans, err) + } +} + +func TestPlannerDerivesStableBoundedJitter(t *testing.T) { + planner, err := NewPlanner(SchedulePolicy{ + Interval: time.Minute, Jitter: 20, MaxInFlight: 1, Timeout: time.Second, MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + checkedAt := time.Date(2026, 7, 31, 17, 0, 0, 0, time.UTC) + first, err := planner.NextDue(checkedAt, "proxy-a\x00BASIC") + if err != nil { + t.Fatalf("NextDue(first): %v", err) + } + second, err := planner.NextDue(checkedAt, "proxy-a\x00BASIC") + if err != nil || second != first { + t.Fatalf("NextDue(stable) = (%v, %v), want %v", second, err, first) + } + base := checkedAt.Add(time.Minute) + if delta := first.Sub(base); delta < -12*time.Second || delta > 12*time.Second { + t.Fatalf("NextDue jitter delta = %s, want [-12s, 12s]", delta) + } +} + +func TestPlannerRejectsInvalidCandidatesWithoutPartialPlan(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) + } + _, err = planner.Plan(time.Now(), 0, 1, []Candidate{{ + ProxyID: "proxy-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelTarget, + RoutingName: "route-a", TargetURL: "not-a-url", DueAt: time.Now(), + }}) + if !errors.Is(err, ErrInvalidScheduleRequest) { + t.Fatalf("Plan(invalid target) error = %v, want ErrInvalidScheduleRequest", err) + } +}