From 0f029c6127853dea84194708ac7eabfeda4a22e5 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 21:37:41 +0800 Subject: [PATCH] refactor: share health task contracts --- internal/controller/health/scheduler.go | 41 ++------ internal/controller/health/task_broker.go | 82 ++-------------- internal/domain/health/task.go | 113 ++++++++++++++++++++++ 3 files changed, 130 insertions(+), 106 deletions(-) create mode 100644 internal/domain/health/task.go diff --git a/internal/controller/health/scheduler.go b/internal/controller/health/scheduler.go index a6c324f..928698d 100644 --- a/internal/controller/health/scheduler.go +++ b/internal/controller/health/scheduler.go @@ -28,36 +28,19 @@ type SchedulePolicy struct { 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 +// Task contracts live in the health domain so every shared store can use the +// same boundary without importing Controller orchestration code. +type Candidate = healthDomain.Candidate +type Priority = healthDomain.Priority +type PlannedTask = healthDomain.PlannedTask const ( - PriorityFetched Priority = iota + 1 - PrioritySuspect - PriorityUnhealthy - PriorityAvailable + PriorityFetched = healthDomain.PriorityFetched + PrioritySuspect = healthDomain.PrioritySuspect + PriorityUnhealthy = healthDomain.PriorityUnhealthy + PriorityAvailable = healthDomain.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 { @@ -129,7 +112,7 @@ func (planner *Planner) Plan(now time.Time, inFlight, maxTasks int, candidates [ 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) + return healthDomain.CandidateIdentity(eligible[left].candidate) < healthDomain.CandidateIdentity(eligible[right].candidate) }) if len(eligible) > limit { eligible = eligible[:limit] @@ -180,7 +163,3 @@ func (planner *Planner) validateCandidate(candidate Candidate) (Priority, bool, 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/task_broker.go b/internal/controller/health/task_broker.go index 377edbb..b8f9627 100644 --- a/internal/controller/health/task_broker.go +++ b/internal/controller/health/task_broker.go @@ -6,7 +6,6 @@ import ( "crypto/sha256" "encoding/hex" "errors" - "fmt" "reflect" "sort" "strconv" @@ -30,79 +29,12 @@ var ( const defaultMaxTasksPerClaim = 128 -// TaskMaterial is the short-lived proxy connection material needed to execute -// one probe. It crosses only the authenticated Checker control stream and is -// deliberately redacted from formatted values and persistence APIs. -type TaskMaterial struct { - Protocol proxyDomain.Scheme - Host string - Port uint16 - SecretRef string - CredentialVersion string - Username string - Password string -} - -func (TaskMaterial) Format(state fmt.State, _ rune) { - _, _ = state.Write([]byte("health.TaskMaterial{Credentials:}")) -} - -// TaskMaterialResolver resolves a task's endpoint and credential material at -// claim time. A production implementation reads the Controller-local -// credential store; Checkers never read Redis or PostgreSQL. -type TaskMaterialResolver interface { - ResolveCheckTask(context.Context, Candidate) (TaskMaterial, error) -} - -type TaskMaterialResolverFunc func(context.Context, Candidate) (TaskMaterial, error) - -func (resolver TaskMaterialResolverFunc) ResolveCheckTask(ctx context.Context, candidate Candidate) (TaskMaterial, error) { - return resolver(ctx, candidate) -} - -// TaskClaim is one bounded pull request from a Checker process. MaxInFlight -// applies across all streams for the checker ID, so reconnecting cannot grow -// its local work window. -type TaskClaim struct { - CheckerID string - InstanceID string - MaxInFlight int - SupportedLevels []healthDomain.Level -} - -// LeasedTask is a Controller-assigned task. It holds material only in memory -// for the duration of a task lease and must not be logged. -type LeasedTask struct { - TaskID string - LeaseToken string - ProxyID string - Protocol proxyDomain.Scheme - Host string - Port uint16 - SecretRef string - CredentialVersion string - Username string - Password string - Level healthDomain.Level - RoutingName string - TargetURL string - Deadline time.Time - Attempts int -} - -func (LeasedTask) Format(state fmt.State, _ rune) { - _, _ = state.Write([]byte("health.LeasedTask{Credentials:}")) -} - -// TaskBroker is the shared task lease boundary. The Scheduler uses Offer, -// Checkers use Claim, and ReportObservations fences facts against the lease. -// The production Redis implementation will use this exact contract. -type TaskBroker interface { - TaskSink - Claim(context.Context, TaskClaim) ([]LeasedTask, error) - AuthorizeObservation(context.Context, string, string, healthDomain.Observation, time.Time) error - CompleteObservation(context.Context, string, string, healthDomain.Observation, time.Time) error -} +type TaskMaterial = healthDomain.TaskMaterial +type TaskMaterialResolver = healthDomain.TaskMaterialResolver +type TaskMaterialResolverFunc = healthDomain.TaskMaterialResolverFunc +type TaskClaim = healthDomain.TaskClaim +type LeasedTask = healthDomain.LeasedTask +type TaskBroker = healthDomain.TaskBroker type MemoryTaskBrokerOptions struct { LeaseTTL time.Duration @@ -453,7 +385,7 @@ func validateTaskMaterial(material TaskMaterial) error { } func deterministicTaskID(task PlannedTask) string { - payload := candidateIdentity(task.Candidate) + "\x00" + task.Deadline.UTC().Format(time.RFC3339Nano) + "\x00" + + payload := healthDomain.CandidateIdentity(task.Candidate) + "\x00" + task.Deadline.UTC().Format(time.RFC3339Nano) + "\x00" + strconv.Itoa(task.Attempts) digest := sha256.Sum256([]byte(payload)) return "check_" + hex.EncodeToString(digest[:]) diff --git a/internal/domain/health/task.go b/internal/domain/health/task.go new file mode 100644 index 0000000..27babc6 --- /dev/null +++ b/internal/domain/health/task.go @@ -0,0 +1,113 @@ +package health + +import ( + "context" + "fmt" + "time" + + proxyDomain "proxy-pool/internal/domain/proxy" +) + +// Candidate is one bounded due-index item. It identifies a check without +// carrying endpoint credentials or any persistence-specific representation. +type Candidate struct { + ProxyID string + State proxyDomain.State + Level Level + RoutingName string + TargetURL string + DueAt time.Time +} + +// CandidateIdentity is stable across Controller instances and is suitable for +// deterministic task identity and jitter derivation. +func CandidateIdentity(candidate Candidate) string { + return candidate.ProxyID + "\x00" + string(candidate.Level) + "\x00" + candidate.RoutingName + "\x00" + candidate.TargetURL +} + +type Priority uint8 + +const ( + PriorityFetched Priority = iota + 1 + PrioritySuspect + PriorityUnhealthy + PriorityAvailable +) + +// PlannedTask is transport-neutral work ready for a shared leased task store. +type PlannedTask struct { + Candidate Candidate + Priority Priority + Deadline time.Time + Attempts int +} + +// TaskMaterial is short-lived proxy connection material. It crosses only the +// authenticated Checker control stream and must never be logged or persisted +// by a task queue. +type TaskMaterial struct { + Protocol proxyDomain.Scheme + Host string + Port uint16 + SecretRef string + CredentialVersion string + Username string + Password string +} + +func (TaskMaterial) Format(state fmt.State, _ rune) { + _, _ = state.Write([]byte("health.TaskMaterial{Credentials:}")) +} + +// TaskMaterialResolver resolves endpoint and credential material only when a +// task lease is granted. Checkers do not directly access Redis or PostgreSQL. +type TaskMaterialResolver interface { + ResolveCheckTask(context.Context, Candidate) (TaskMaterial, error) +} + +type TaskMaterialResolverFunc func(context.Context, Candidate) (TaskMaterial, error) + +func (resolver TaskMaterialResolverFunc) ResolveCheckTask(ctx context.Context, candidate Candidate) (TaskMaterial, error) { + return resolver(ctx, candidate) +} + +// TaskClaim is one bounded pull request from a Checker process. +type TaskClaim struct { + CheckerID string + InstanceID string + MaxInFlight int + SupportedLevels []Level +} + +// LeasedTask is a Controller-assigned task. Its material is kept in memory +// only for the task lease duration and is redacted from formatted values. +type LeasedTask struct { + TaskID string + LeaseToken string + ProxyID string + Protocol proxyDomain.Scheme + Host string + Port uint16 + SecretRef string + CredentialVersion string + Username string + Password string + Level Level + RoutingName string + TargetURL string + Deadline time.Time + Attempts int +} + +func (LeasedTask) Format(state fmt.State, _ rune) { + _, _ = state.Write([]byte("health.LeasedTask{Credentials:}")) +} + +// TaskBroker is the shared task lease boundary. Schedulers offer bounded +// work, Checkers claim work, and observation commits are fenced by a lease. +type TaskBroker interface { + Offer(context.Context, []PlannedTask) (int, error) + Claim(context.Context, TaskClaim) ([]LeasedTask, error) + AuthorizeObservation(context.Context, string, string, Observation, time.Time) error + CompleteObservation(context.Context, string, string, Observation, time.Time) error +}