feat: add redis basic health task broker

internal/adapters/redisactivity/adapter.go

@@ -26,6 +26,8 @@ type Options struct {
 	MaxRuntimeCounters int
 	MaxInventoryScan   int
 	CleanupLimit       int
+	MaxCheckTasks      int
+	CheckLeaseTTL      time.Duration
 }
 
 type Adapter struct {
@@ -44,10 +46,17 @@ func New(client redis.Scripter, options Options) (*Adapter, error) {
 	if options.MaxInventoryScan == 0 {
 		options.MaxInventoryScan = options.MaxCandidateScan
 	}
+	if options.MaxCheckTasks == 0 {
+		options.MaxCheckTasks = 128
+	}
+	if options.CheckLeaseTTL == 0 {
+		options.CheckLeaseTTL = 30 * time.Second
+	}
 	if nilInterface(client) || nilInterface(options.Credentials) ||
 		!namespacePattern.MatchString(options.Namespace) || options.OperationTTL <= 0 ||
 		options.MaxCandidateScan <= 0 || options.MaxRuntimeCounters <= 0 ||
-		options.MaxInventoryScan <= 0 || options.CleanupLimit <= 0 {
+		options.MaxInventoryScan <= 0 || options.CleanupLimit <= 0 ||
+		options.MaxCheckTasks <= 0 || options.CheckLeaseTTL <= 0 {
 		return nil, ErrInvalidOptions
 	}
 	adapter := &Adapter{

internal/adapters/redisactivity/health_tasks.go

@@ -0,0 +1,363 @@
+package redisactivity
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	healthDomain "proxy-pool/internal/domain/health"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+const healthTaskRecordVersion = 1
+
+var _ healthDomain.TaskBroker = (*Adapter)(nil)
+
+type healthTaskRecord struct {
+	Version          int    `json:"version"`
+	TaskID           string `json:"taskId"`
+	ProxyID          string `json:"proxyId"`
+	Level            string `json:"level"`
+	Priority         int    `json:"priority"`
+	DeadlineMS       int64  `json:"deadlineMs"`
+	NextDueMS        int64  `json:"nextDueMs"`
+	Attempts         int    `json:"attempts"`
+	State            string `json:"state"`
+	RoutingName      string `json:"routingName,omitempty"`
+	TargetURL        string `json:"targetUrl,omitempty"`
+	LeaseCheckerID   string `json:"leaseCheckerId,omitempty"`
+	LeaseInstanceID  string `json:"leaseInstanceId,omitempty"`
+	LeaseToken       string `json:"leaseToken,omitempty"`
+	LeaseExpiresAtMS int64  `json:"leaseExpiresAtMs,omitempty"`
+	CheckerLeaseKey  string `json:"checkerLeaseKey,omitempty"`
+}
+
+type healthTaskRequest struct {
+	Limit       int                  `json:"limit"`
+	Tasks       []healthTaskRecord   `json:"tasks,omitempty"`
+	CheckerID   string               `json:"checkerId,omitempty"`
+	InstanceID  string               `json:"instanceId,omitempty"`
+	MaxInFlight int                  `json:"maxInFlight,omitempty"`
+	LeaseTTLMS  int64                `json:"leaseTTLMS,omitempty"`
+	Levels      []healthDomain.Level `json:"levels,omitempty"`
+	Tokens      []string             `json:"tokens,omitempty"`
+	Fact        *healthTaskFact      `json:"fact,omitempty"`
+}
+
+type healthTaskFact struct {
+	TaskID      string `json:"taskId"`
+	ProxyID     string `json:"proxyId"`
+	Level       string `json:"level"`
+	RoutingName string `json:"routingName,omitempty"`
+	TargetURL   string `json:"targetUrl,omitempty"`
+	CheckerID   string `json:"checkerId"`
+	LeaseToken  string `json:"leaseToken"`
+}
+
+// InFlight returns the total queued and leased BASIC tasks after bounded
+// maintenance. The Scheduler uses it only outside the Gateway hot path.
+func (a *Adapter) InFlight(ctx context.Context, now time.Time) (int, error) {
+	if a == nil || ctx == nil || now.IsZero() {
+		return 0, healthDomain.ErrInvalidTaskBroker
+	}
+	reply, err := a.runHealthTaskScript(ctx, "inflight", now, healthTaskRequest{Limit: a.options.MaxCheckTasks}, "runtime")
+	if err != nil {
+		return 0, err
+	}
+	if reply.Status != scriptOK || reply.Count < 0 {
+		return 0, invalidScriptReply("invalid health task inflight reply")
+	}
+	return reply.Count, nil
+}
+
+// DueCandidates returns a bounded BASIC due batch. Stale references are
+// discarded inside the Lua script before they can reach the Controller.
+func (a *Adapter) DueCandidates(ctx context.Context, now time.Time, limit int) ([]healthDomain.Candidate, error) {
+	if a == nil || ctx == nil || now.IsZero() || limit <= 0 || limit > a.options.MaxCheckTasks {
+		return nil, healthDomain.ErrInvalidTaskBroker
+	}
+	reply, err := a.runHealthTaskScript(ctx, "due", now, healthTaskRequest{Limit: limit}, "runtime")
+	if err != nil {
+		return nil, err
+	}
+	if reply.Status != scriptOK || len(reply.Candidates) > limit {
+		return nil, invalidScriptReply("invalid health task due reply")
+	}
+	result := make([]healthDomain.Candidate, 0, len(reply.Candidates))
+	for _, candidate := range reply.Candidates {
+		state := proxyDomain.State(candidate.State)
+		switch state {
+		case proxyDomain.StateFetched, proxyDomain.StateChecking, proxyDomain.StateAvailable, proxyDomain.StateSuspect, proxyDomain.StateUnhealthy:
+		default:
+			return nil, invalidScriptReply("health task due reply contains invalid state")
+		}
+		if !validHealthTaskIdentifier(candidate.ProxyID) || candidate.DueAtMS <= 0 {
+			return nil, invalidScriptReply("health task due reply contains invalid candidate")
+		}
+		result = append(result, healthDomain.Candidate{
+			ProxyID: candidate.ProxyID, State: state, Level: healthDomain.LevelBasic,
+			DueAt: time.UnixMilli(candidate.DueAtMS).UTC(),
+		})
+	}
+	return result, nil
+}
+
+// Offer atomically turns due BASIC plans into shared queued tasks. A plan that
+// loses the race leaves no duplicate task and no in-memory queue behind.
+func (a *Adapter) Offer(ctx context.Context, plans []healthDomain.PlannedTask) (int, error) {
+	if a == nil || ctx == nil {
+		return 0, healthDomain.ErrInvalidTaskBroker
+	}
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
+	if len(plans) == 0 {
+		return 0, nil
+	}
+	if len(plans) > a.options.MaxCheckTasks {
+		return 0, healthDomain.ErrInvalidLeasedTask
+	}
+	now := time.Now().UTC()
+	tasks := make([]healthTaskRecord, len(plans))
+	for index, plan := range plans {
+		if err := validateRedisPlannedTask(plan, now); err != nil {
+			return 0, err
+		}
+		tasks[index] = healthTaskRecord{
+			Version: healthTaskRecordVersion, TaskID: healthDomain.TaskIDFor(plan), ProxyID: plan.Candidate.ProxyID,
+			Level: string(plan.Candidate.Level), Priority: int(plan.Priority), DeadlineMS: plan.Deadline.UnixMilli(),
+			NextDueMS: plan.NextDue.UnixMilli(), Attempts: plan.Attempts, State: "QUEUED",
+		}
+	}
+	reply, err := a.runHealthTaskScript(ctx, "offer", now, healthTaskRequest{Limit: a.options.MaxCheckTasks, Tasks: tasks}, "runtime")
+	if err != nil {
+		return 0, err
+	}
+	if reply.Status != scriptOK || reply.Count < 0 || reply.Count > len(plans) {
+		return 0, invalidScriptReply("invalid health task offer reply")
+	}
+	return reply.Count, nil
+}
+
+// Claim leases a bounded task batch. Proxy credentials are read only from the
+// live record at lease time and are never copied into Redis task state.
+func (a *Adapter) Claim(ctx context.Context, claim healthDomain.TaskClaim) ([]healthDomain.LeasedTask, error) {
+	if a == nil || ctx == nil {
+		return nil, healthDomain.ErrInvalidTaskBroker
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if err := validateRedisTaskClaim(claim, a.options.MaxCheckTasks); err != nil {
+		return nil, err
+	}
+	tokens := make([]string, claim.MaxInFlight)
+	for index := range tokens {
+		token, err := newRedisLeaseToken()
+		if err != nil {
+			return nil, err
+		}
+		tokens[index] = token
+	}
+	now := time.Now().UTC()
+	reply, err := a.runHealthTaskScript(ctx, "claim", now, healthTaskRequest{
+		Limit: a.options.MaxCheckTasks, CheckerID: claim.CheckerID, InstanceID: claim.InstanceID,
+		MaxInFlight: claim.MaxInFlight, LeaseTTLMS: a.options.CheckLeaseTTL.Milliseconds(),
+		Levels: claim.SupportedLevels, Tokens: tokens,
+	}, claim.CheckerID)
+	if err != nil {
+		return nil, err
+	}
+	if reply.Status != scriptOK || len(reply.Tasks) > claim.MaxInFlight {
+		return nil, invalidScriptReply("invalid health task claim reply")
+	}
+	result := make([]healthDomain.LeasedTask, 0, len(reply.Tasks))
+	for _, item := range reply.Tasks {
+		task, err := decodeHealthTaskRecord(item.Task)
+		if err != nil {
+			return nil, invalidScriptReply("health task claim reply contains invalid task")
+		}
+		record, err := decodeProxyRecord(item.Record)
+		if err != nil {
+			return nil, invalidScriptReply("health task claim reply contains invalid record")
+		}
+		if task.State != "LEASED" || task.LeaseCheckerID != claim.CheckerID || task.LeaseToken == "" ||
+			task.Level != string(healthDomain.LevelBasic) || task.ProxyID != record.ID || task.DeadlineMS <= now.UnixMilli() {
+			return nil, invalidScriptReply("health task claim reply violates lease contract")
+		}
+		result = append(result, healthDomain.LeasedTask{
+			TaskID: task.TaskID, LeaseToken: task.LeaseToken, ProxyID: task.ProxyID,
+			Protocol: proxyDomain.Scheme(record.Scheme), Host: record.Host, Port: uint16(record.Port),
+			Username: record.Username, Password: record.Password, Level: healthDomain.Level(task.Level),
+			Deadline: time.UnixMilli(task.DeadlineMS).UTC(), Attempts: task.Attempts,
+		})
+	}
+	return result, nil
+}
+
+func (a *Adapter) AuthorizeObservation(
+	ctx context.Context,
+	checkerID string,
+	leaseToken string,
+	observation healthDomain.Observation,
+	now time.Time,
+) error {
+	return a.authorizeHealthTask(ctx, "authorize", checkerID, leaseToken, observation, now)
+}
+
+func (a *Adapter) CompleteObservation(
+	ctx context.Context,
+	checkerID string,
+	leaseToken string,
+	observation healthDomain.Observation,
+	now time.Time,
+) error {
+	return a.authorizeHealthTask(ctx, "complete", checkerID, leaseToken, observation, now)
+}
+
+func (a *Adapter) authorizeHealthTask(
+	ctx context.Context,
+	operation string,
+	checkerID string,
+	leaseToken string,
+	observation healthDomain.Observation,
+	now time.Time,
+) error {
+	if a == nil || ctx == nil || !validHealthTaskIdentifier(checkerID) || !validHealthTaskIdentifier(leaseToken) || now.IsZero() {
+		return healthDomain.ErrInvalidTaskBroker
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	normalized, err := healthDomain.NormalizeObservation(observation)
+	if err != nil {
+		return err
+	}
+	reply, err := a.runHealthTaskScript(ctx, operation, now, healthTaskRequest{
+		Limit: a.options.MaxCheckTasks,
+		Fact: &healthTaskFact{TaskID: normalized.TaskID, ProxyID: normalized.ProxyID, Level: string(normalized.Level),
+			RoutingName: normalized.RoutingName, TargetURL: normalized.TargetURL, CheckerID: checkerID, LeaseToken: leaseToken},
+	}, checkerID)
+	if err != nil {
+		return err
+	}
+	switch reply.Status {
+	case scriptOK:
+		return nil
+	case scriptNotFound:
+		return healthDomain.ErrTaskNotFound
+	case scriptInvalid:
+		return healthDomain.ErrInvalidLeasedTask
+	case scriptStatus("lease_expired"):
+		return healthDomain.ErrTaskLeaseExpired
+	case scriptStatus("not_owned"):
+		return healthDomain.ErrTaskLeaseNotOwned
+	case scriptStatus("observation"):
+		return healthDomain.ErrTaskObservation
+	default:
+		return invalidScriptReply("unexpected health task authorization status")
+	}
+}
+
+func (a *Adapter) runHealthTaskScript(
+	ctx context.Context,
+	operation string,
+	now time.Time,
+	request healthTaskRequest,
+	checkerID string,
+) (healthTaskScriptReply, error) {
+	if a == nil || ctx == nil || now.IsZero() || request.Limit <= 0 || strings.TrimSpace(operation) != operation || operation == "" {
+		return healthTaskScriptReply{}, healthDomain.ErrInvalidTaskBroker
+	}
+	payload, err := json.Marshal(request)
+	if err != nil {
+		return healthTaskScriptReply{}, fmt.Errorf("encode health task request: %w", err)
+	}
+	result, err := runScript(ctx, a.client, healthTasksScript, []string{
+		a.keys.records, a.keys.expiry, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
+		a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.checkerLeases(checkerID),
+		a.keys.stateInventory,
+	}, operation, now.UTC().UnixMilli(), string(payload))
+	if err != nil {
+		return healthTaskScriptReply{}, err
+	}
+	var reply healthTaskScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return healthTaskScriptReply{}, err
+	}
+	return reply, nil
+}
+
+func validateRedisPlannedTask(task healthDomain.PlannedTask, now time.Time) error {
+	if !validHealthTaskIdentifier(task.Candidate.ProxyID) || task.Candidate.Level != healthDomain.LevelBasic ||
+		task.Candidate.RoutingName != "" || task.Candidate.TargetURL != "" || task.Candidate.DueAt.IsZero() ||
+		task.Deadline.IsZero() || !task.Deadline.After(now) || task.NextDue.IsZero() || !task.NextDue.After(now) || task.Attempts <= 0 {
+		return healthDomain.ErrInvalidLeasedTask
+	}
+	switch task.Candidate.State {
+	case proxyDomain.StateFetched, proxyDomain.StateAvailable, proxyDomain.StateSuspect, proxyDomain.StateUnhealthy:
+		return nil
+	default:
+		return healthDomain.ErrInvalidLeasedTask
+	}
+}
+
+func validateRedisTaskClaim(claim healthDomain.TaskClaim, maximum int) error {
+	if !validHealthTaskIdentifier(claim.CheckerID) || !validHealthTaskIdentifier(claim.InstanceID) || claim.MaxInFlight <= 0 ||
+		claim.MaxInFlight > maximum || len(claim.SupportedLevels) == 0 {
+		return healthDomain.ErrInvalidTaskClaim
+	}
+	levels := make(map[healthDomain.Level]struct{}, len(claim.SupportedLevels))
+	for _, level := range claim.SupportedLevels {
+		switch level {
+		case healthDomain.LevelBasic, healthDomain.LevelEgress, healthDomain.LevelTarget:
+		default:
+			return healthDomain.ErrInvalidTaskClaim
+		}
+		if _, exists := levels[level]; exists {
+			return healthDomain.ErrInvalidTaskClaim
+		}
+		levels[level] = struct{}{}
+	}
+	return nil
+}
+
+func decodeHealthTaskRecord(payload string) (healthTaskRecord, error) {
+	var record healthTaskRecord
+	if err := decodeJSON(payload, &record); err != nil {
+		return healthTaskRecord{}, err
+	}
+	if record.Version != healthTaskRecordVersion || !validHealthTaskIdentifier(record.TaskID) ||
+		!validHealthTaskIdentifier(record.ProxyID) || record.Level != string(healthDomain.LevelBasic) ||
+		record.Priority < int(healthDomain.PriorityFetched) || record.Priority > int(healthDomain.PriorityAvailable) ||
+		record.DeadlineMS <= 0 || record.NextDueMS <= 0 || record.Attempts <= 0 ||
+		(record.State != "QUEUED" && record.State != "LEASED" && record.State != "DONE") {
+		return healthTaskRecord{}, errors.New("invalid health task record")
+	}
+	return record, nil
+}
+
+func newRedisLeaseToken() (string, error) {
+	var entropy [24]byte
+	if _, err := rand.Read(entropy[:]); err != nil {
+		return "", err
+	}
+	return "lease_" + hex.EncodeToString(entropy[:]), nil
+}
+
+func validHealthTaskIdentifier(value string) bool {
+	if value == "" || len(value) > 256 || strings.TrimSpace(value) != value {
+		return false
+	}
+	for _, character := range value {
+		if character <= ' ' || character == '\x7f' {
+			return false
+		}
+	}
+	return true
+}

internal/adapters/redisactivity/health_tasks_integration_test.go

@@ -0,0 +1,74 @@
+//go:build integration
+
+package redisactivity
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	controllerHealth "proxy-pool/internal/controller/health"
+	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+func TestRedisHealthTasksLeaseAndRescheduleBasicChecks(t *testing.T) {
+	fixture := newRedisTestFixture(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	if _, err := fixture.Adapter.UpsertFetched(ctx, "provider-a", activitypool.FetchedBatch{
+		ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
+		Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
+	}); err != nil {
+		t.Fatalf("UpsertFetched(): %v", err)
+	}
+
+	candidates, err := fixture.Adapter.DueCandidates(ctx, time.Now().UTC(), 1)
+	if err != nil || len(candidates) != 1 || candidates[0].ProxyID != "proxy-a" || candidates[0].State != proxyDomain.StateFetched {
+		t.Fatalf("DueCandidates() = (%+v, %v)", candidates, err)
+	}
+	planner, err := controllerHealth.NewPlanner(controllerHealth.SchedulePolicy{
+		Interval: 30 * time.Second, MaxInFlight: 1, Timeout: 5 * time.Second, MaxAttempts: 1,
+	})
+	if err != nil {
+		t.Fatalf("newHealthTaskPlanner(): %v", err)
+	}
+	plannedAt := time.Now().UTC()
+	plans, err := planner.Plan(plannedAt, 0, 1, candidates)
+	if err != nil || len(plans) != 1 {
+		t.Fatalf("Plan() = (%+v, %v)", plans, err)
+	}
+	if offered, err := fixture.Adapter.Offer(ctx, plans); err != nil || offered != 1 {
+		t.Fatalf("Offer() = (%d, %v)", offered, err)
+	}
+
+	claimed, err := fixture.Adapter.Claim(ctx, healthDomain.TaskClaim{
+		CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 1,
+		SupportedLevels: []healthDomain.Level{healthDomain.LevelBasic},
+	})
+	if err != nil || len(claimed) != 1 || claimed[0].ProxyID != "proxy-a" || claimed[0].LeaseToken == "" {
+		t.Fatalf("Claim() = (%+v, %v)", claimed, err)
+	}
+	observation := healthDomain.Observation{
+		TaskID: claimed[0].TaskID, ProxyID: claimed[0].ProxyID, Level: healthDomain.LevelBasic,
+		Success: true, Latency: time.Millisecond, ObservedAt: time.Now().UTC(),
+	}
+	if err := fixture.Adapter.AuthorizeObservation(ctx, "checker-a", claimed[0].LeaseToken, observation, time.Now().UTC()); err != nil {
+		t.Fatalf("AuthorizeObservation(): %v", err)
+	}
+	updated, err := fixture.Adapter.ApplyGlobalObservation(ctx, activitypool.GlobalHealthCommand{
+		Observation: observation, MaxConsecutiveFailures: 2,
+	})
+	if err != nil || updated.State != proxyDomain.StateAvailable {
+		t.Fatalf("ApplyGlobalObservation() = (%+v, %v)", updated, err)
+	}
+	if err := fixture.Adapter.CompleteObservation(ctx, "checker-a", claimed[0].LeaseToken, observation, time.Now().UTC()); err != nil {
+		t.Fatalf("CompleteObservation(): %v", err)
+	}
+
+	next, err := fixture.Adapter.DueCandidates(ctx, plans[0].NextDue, 1)
+	if err != nil || len(next) != 1 || next[0].ProxyID != "proxy-a" || next[0].State != proxyDomain.StateAvailable {
+		t.Fatalf("DueCandidates(next) = (%+v, %v)", next, err)
+	}
+}

internal/adapters/redisactivity/keys.go

@@ -27,6 +27,12 @@ type keyspace struct {
 	workerRuntime        string
 	workerRuntimeExpiry  string
 	workerOutcomes       string
+	healthDue            string
+	healthQueued         string
+	healthLeases         string
+	healthTasks          string
+	healthTaskExpiry     string
+	healthRefTask        string
 }
 
 func newKeyspace(namespace string) keyspace {
@@ -50,9 +56,19 @@ func newKeyspace(namespace string) keyspace {
 		workerRuntime:        prefix + ":worker-runtime",
 		workerRuntimeExpiry:  prefix + ":worker-runtime-expiry",
 		workerOutcomes:       prefix + ":worker-outcomes",
+		healthDue:            prefix + ":health-due",
+		healthQueued:         prefix + ":health-queued",
+		healthLeases:         prefix + ":health-leases",
+		healthTasks:          prefix + ":health-tasks",
+		healthTaskExpiry:     prefix + ":health-task-expiry",
+		healthRefTask:        prefix + ":health-ref-task",
 	}
 }
 
+func (keys keyspace) checkerLeases(checkerID string) string {
+	return keys.prefix + ":health-checker-leases:" + digestToken(checkerID)
+}
+
 func stateInventoryField(upstreamID, state string) string {
 	return strconv.Itoa(len(upstreamID)) + ":" + upstreamID + ":" + state
 }

internal/adapters/redisactivity/scripts.go

@@ -44,6 +44,24 @@ type healthScriptReply struct {
 	Record string       `json:"record,omitempty"`
 }
 
+type healthTaskScriptReply struct {
+	Status     scriptStatus              `json:"status"`
+	Count      int                       `json:"count"`
+	Candidates []healthTaskCandidateWire `json:"candidates"`
+	Tasks      []healthTaskClaimWire     `json:"tasks"`
+}
+
+type healthTaskCandidateWire struct {
+	ProxyID string `json:"proxyId"`
+	State   string `json:"state"`
+	DueAtMS int64  `json:"dueAtMs"`
+}
+
+type healthTaskClaimWire struct {
+	Task   string `json:"task"`
+	Record string `json:"record"`
+}
+
 type targetHealthScriptReply struct {
 	Status scriptStatus `json:"status"`
 	Target string       `json:"target,omitempty"`
@@ -115,6 +133,9 @@ var upsertSource string
 //go:embed scripts/health.lua
 var healthSource string
 
+//go:embed scripts/health_tasks.lua
+var healthTasksSource string
+
 //go:embed scripts/target_health.lua
 var targetHealthSource string
 
@@ -145,6 +166,7 @@ var workerSnapshotSource string
 var (
 	upsertScript         = redis.NewScript(upsertSource)
 	healthScript         = redis.NewScript(healthSource)
+	healthTasksScript    = redis.NewScript(healthTasksSource)
 	targetHealthScript   = redis.NewScript(targetHealthSource)
 	upstreamLookupScript = redis.NewScript(upstreamLookupSource)
 	extractScript        = redis.NewScript(extractSource)

internal/adapters/redisactivity/scripts/health_tasks.lua

@@ -0,0 +1,329 @@
+local records_key = KEYS[1]
+local expiry_key = KEYS[2]
+local due_key = KEYS[3]
+local queued_key = KEYS[4]
+local leases_key = KEYS[5]
+local tasks_key = KEYS[6]
+local task_expiry_key = KEYS[7]
+local ref_task_key = KEYS[8]
+local checker_leases_key = KEYS[9]
+local state_inventory_key = KEYS[10]
+
+local operation = ARGV[1]
+local now_ms = tonumber(ARGV[2])
+local decoded, payload = pcall(cjson.decode, ARGV[3])
+if not decoded or type(payload) ~= 'table' or not now_ms then
+    return cjson.encode({status = 'invalid'})
+end
+
+local limit = tonumber(payload.limit or 0)
+if not limit or limit <= 0 then
+    return cjson.encode({status = 'invalid'})
+end
+
+local function finish(reply)
+    return cjson.encode(reply)
+end
+
+local function live_record(proxy_id)
+    local raw = redis.call('HGET', records_key, proxy_id)
+    if not raw then
+        return nil, nil
+    end
+    local valid, record = pcall(cjson.decode, raw)
+    if not valid or type(record) ~= 'table' or tonumber(record.expiresAtMs) <= now_ms or record.state == 'EXTRACTED' then
+        return nil, nil
+    end
+    return raw, record
+end
+
+local function touch(key, expires_at_ms)
+    if redis.call('EXISTS', key) == 0 then
+        return
+    end
+    local current = redis.call('PEXPIRETIME', key)
+    if current < expires_at_ms then
+        redis.call('PEXPIREAT', key, expires_at_ms)
+    end
+end
+
+local function remove_task(task_id, requeue)
+    local raw = redis.call('HGET', tasks_key, task_id)
+    local task = nil
+    if raw then
+        local valid
+        valid, task = pcall(cjson.decode, raw)
+        if not valid or type(task) ~= 'table' then
+            task = nil
+        end
+    end
+    redis.call('ZREM', queued_key, task_id)
+    redis.call('ZREM', leases_key, task_id)
+    redis.call('ZREM', task_expiry_key, task_id)
+    redis.call('HDEL', tasks_key, task_id)
+    if task then
+        if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then
+            redis.call('ZREM', task.checkerLeaseKey, task_id)
+        end
+        if type(task.proxyId) == 'string' and redis.call('HGET', ref_task_key, task.proxyId) == task_id then
+            redis.call('HDEL', ref_task_key, task.proxyId)
+        end
+        if requeue and type(task.proxyId) == 'string' then
+            local _, record = live_record(task.proxyId)
+            if record then
+                redis.call('ZADD', due_key, now_ms, task.proxyId)
+                touch(due_key, tonumber(record.expiresAtMs))
+            end
+        end
+    end
+end
+
+local function reap()
+    local expired_leases = redis.call('ZRANGEBYSCORE', leases_key, '-inf', now_ms, 'LIMIT', 0, limit)
+    for _, task_id in ipairs(expired_leases) do
+        local raw = redis.call('HGET', tasks_key, task_id)
+        if not raw then
+            redis.call('ZREM', leases_key, task_id)
+        else
+            local valid, task = pcall(cjson.decode, raw)
+            if not valid or type(task) ~= 'table' then
+                remove_task(task_id, true)
+            elseif task.state == 'LEASED' and tonumber(task.leaseExpiresAtMs or 0) <= now_ms then
+                redis.call('ZREM', leases_key, task_id)
+                if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then
+                    redis.call('ZREM', task.checkerLeaseKey, task_id)
+                end
+                task.state = 'QUEUED'
+                task.leaseCheckerId = ''
+                task.leaseToken = ''
+                task.leaseExpiresAtMs = 0
+                task.checkerLeaseKey = ''
+                redis.call('HSET', tasks_key, task_id, cjson.encode(task))
+                redis.call('ZADD', queued_key, tonumber(task.priority), task_id)
+            else
+                redis.call('ZREM', leases_key, task_id)
+            end
+        end
+    end
+    local expired_tasks = redis.call('ZRANGEBYSCORE', task_expiry_key, '-inf', now_ms, 'LIMIT', 0, limit)
+    for _, task_id in ipairs(expired_tasks) do
+        local raw = redis.call('HGET', tasks_key, task_id)
+        local requeue = true
+        if raw then
+            local valid, task = pcall(cjson.decode, raw)
+            if valid and type(task) == 'table' and task.state == 'DONE' then
+                requeue = false
+            end
+        end
+        remove_task(task_id, requeue)
+    end
+end
+
+local function valid_state(state)
+    return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or
+        state == 'SUSPECT' or state == 'UNHEALTHY'
+end
+
+local function state_field(upstream, state)
+    return string.len(upstream) .. ':' .. upstream .. ':' .. state
+end
+
+local function decrement_state(upstream, state)
+    local field = state_field(upstream, state)
+    local value = redis.call('HINCRBY', state_inventory_key, field, -1)
+    if value <= 0 then
+        redis.call('HDEL', state_inventory_key, field)
+    end
+end
+
+local function increment_state(upstream, state)
+    redis.call('HINCRBY', state_inventory_key, state_field(upstream, state), 1)
+end
+
+local function matches_task(task, fact)
+    return task and task.taskId == fact.taskId and task.proxyId == fact.proxyId and task.level == fact.level and
+        (task.routingName or '') == (fact.routingName or '') and (task.targetUrl or '') == (fact.targetUrl or '')
+end
+
+local function authorize(fact)
+    if type(fact) ~= 'table' or type(fact.taskId) ~= 'string' or type(fact.proxyId) ~= 'string' or
+        type(fact.level) ~= 'string' or type(fact.checkerId) ~= 'string' or type(fact.leaseToken) ~= 'string' then
+        return nil, 'invalid'
+    end
+    local raw = redis.call('HGET', tasks_key, fact.taskId)
+    if not raw then
+        return nil, 'not_found'
+    end
+    local valid, task = pcall(cjson.decode, raw)
+    if not valid or type(task) ~= 'table' then
+        return nil, 'invalid'
+    end
+    if not matches_task(task, fact) then
+        return nil, 'observation'
+    end
+    if task.state == 'DONE' then
+        if task.leaseCheckerId == fact.checkerId and task.leaseToken == fact.leaseToken then
+            return task, 'ok'
+        end
+        return nil, 'not_owned'
+    end
+    if task.state ~= 'LEASED' or tonumber(task.leaseExpiresAtMs or 0) <= now_ms then
+        return nil, 'lease_expired'
+    end
+    if task.leaseCheckerId ~= fact.checkerId or task.leaseToken ~= fact.leaseToken then
+        return nil, 'not_owned'
+    end
+    return task, 'ok'
+end
+
+reap()
+
+if operation == 'inflight' then
+    return finish({status = 'ok', count = redis.call('ZCARD', queued_key) + redis.call('ZCARD', leases_key)})
+end
+
+if operation == 'due' then
+    local result = {}
+    local ids = redis.call('ZRANGEBYSCORE', due_key, '-inf', now_ms, 'LIMIT', 0, limit)
+    for _, proxy_id in ipairs(ids) do
+        local _, record = live_record(proxy_id)
+        if not record or not valid_state(record.state) then
+            redis.call('ZREM', due_key, proxy_id)
+        else
+            result[#result + 1] = {proxyId = proxy_id, state = record.state, dueAtMs = now_ms}
+        end
+    end
+    return finish({status = 'ok', candidates = result})
+end
+
+if operation == 'offer' then
+    if type(payload.tasks) ~= 'table' then
+        return finish({status = 'invalid'})
+    end
+    local offered = 0
+    for _, task in ipairs(payload.tasks) do
+        if type(task) ~= 'table' or task.version ~= 1 or task.level ~= 'BASIC' or task.state ~= 'QUEUED' or
+            type(task.taskId) ~= 'string' or type(task.proxyId) ~= 'string' or tonumber(task.deadlineMs or 0) <= now_ms or
+            tonumber(task.nextDueMs or 0) <= now_ms or tonumber(task.attempts or 0) <= 0 then
+            return finish({status = 'invalid'})
+        end
+        local score = redis.call('ZSCORE', due_key, task.proxyId)
+        local current = redis.call('HGET', ref_task_key, task.proxyId)
+        local raw, record = live_record(task.proxyId)
+        if score and tonumber(score) <= now_ms and not current and raw and valid_state(record.state) then
+            if record.state == 'FETCHED' or record.state == 'UNHEALTHY' then
+                decrement_state(record.sourceUpstream, record.state)
+                increment_state(record.sourceUpstream, 'CHECKING')
+                record.state = 'CHECKING'
+                raw = cjson.encode(record)
+                redis.call('HSET', records_key, task.proxyId, raw)
+            end
+            redis.call('ZREM', due_key, task.proxyId)
+            redis.call('HSET', tasks_key, task.taskId, cjson.encode(task))
+            redis.call('HSET', ref_task_key, task.proxyId, task.taskId)
+            redis.call('ZADD', queued_key, tonumber(task.priority), task.taskId)
+            redis.call('ZADD', task_expiry_key, tonumber(task.deadlineMs), task.taskId)
+            local expires_at_ms = tonumber(record.expiresAtMs)
+            touch(queued_key, expires_at_ms)
+            touch(tasks_key, expires_at_ms)
+            touch(task_expiry_key, expires_at_ms)
+            touch(ref_task_key, expires_at_ms)
+            offered = offered + 1
+        end
+    end
+    return finish({status = 'ok', count = offered})
+end
+
+if operation == 'claim' then
+    if type(payload.checkerId) ~= 'string' or type(payload.instanceId) ~= 'string' or
+        type(payload.levels) ~= 'table' or type(payload.tokens) ~= 'table' then
+        return finish({status = 'invalid'})
+    end
+    local max_in_flight = tonumber(payload.maxInFlight or 0)
+    local lease_ttl_ms = tonumber(payload.leaseTTLMS or 0)
+    if not max_in_flight or max_in_flight <= 0 or not lease_ttl_ms or lease_ttl_ms <= 0 then
+        return finish({status = 'invalid'})
+    end
+    redis.call('ZREMRANGEBYSCORE', checker_leases_key, '-inf', now_ms)
+    local capacity = max_in_flight - redis.call('ZCARD', checker_leases_key)
+    if capacity <= 0 then
+        return finish({status = 'ok', tasks = {}})
+    end
+    local supported = {}
+    for _, level in ipairs(payload.levels) do
+        supported[level] = true
+    end
+    local candidates = redis.call('ZRANGE', queued_key, 0, limit - 1)
+    local result = {}
+    for _, task_id in ipairs(candidates) do
+        if #result >= capacity or #result >= #payload.tokens then
+            break
+        end
+        local raw = redis.call('HGET', tasks_key, task_id)
+        local valid, task = pcall(cjson.decode, raw or '')
+        if not valid or type(task) ~= 'table' or task.state ~= 'QUEUED' or tonumber(task.deadlineMs or 0) <= now_ms then
+            remove_task(task_id, true)
+        elseif not supported[task.level] then
+            -- Preserve unsupported work for a Checker that advertises this level.
+        else
+            local record_raw, record = live_record(task.proxyId)
+            if not record_raw or not record then
+                remove_task(task_id, false)
+            else
+                local lease_expires_at_ms = now_ms + lease_ttl_ms
+                if tonumber(task.deadlineMs) < lease_expires_at_ms then
+                    lease_expires_at_ms = tonumber(task.deadlineMs)
+                end
+                local token = payload.tokens[#result + 1]
+                if type(token) ~= 'string' or token == '' then
+                    return finish({status = 'invalid'})
+                end
+                task.state = 'LEASED'
+                task.leaseCheckerId = payload.checkerId
+                task.leaseInstanceId = payload.instanceId
+                task.leaseToken = token
+                task.leaseExpiresAtMs = lease_expires_at_ms
+                task.checkerLeaseKey = checker_leases_key
+                local encoded = cjson.encode(task)
+                redis.call('HSET', tasks_key, task_id, encoded)
+                redis.call('ZREM', queued_key, task_id)
+                redis.call('ZADD', leases_key, lease_expires_at_ms, task_id)
+                redis.call('ZADD', checker_leases_key, lease_expires_at_ms, task_id)
+                touch(checker_leases_key, tonumber(record.expiresAtMs))
+                result[#result + 1] = {task = encoded, record = record_raw}
+            end
+        end
+    end
+    return finish({status = 'ok', tasks = result})
+end
+
+if operation == 'authorize' or operation == 'complete' then
+    local task, status = authorize(payload.fact)
+    if status ~= 'ok' then
+        return finish({status = status})
+    end
+    if operation == 'complete' and task.state ~= 'DONE' then
+        redis.call('ZREM', queued_key, task.taskId)
+        redis.call('ZREM', leases_key, task.taskId)
+        if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then
+            redis.call('ZREM', task.checkerLeaseKey, task.taskId)
+        end
+        task.state = 'DONE'
+        redis.call('HSET', tasks_key, task.taskId, cjson.encode(task))
+        if redis.call('HGET', ref_task_key, task.proxyId) == task.taskId then
+            redis.call('HDEL', ref_task_key, task.proxyId)
+        end
+        local _, record = live_record(task.proxyId)
+        if record then
+            local next_due_ms = tonumber(task.nextDueMs or 0)
+            if next_due_ms <= now_ms then
+                next_due_ms = now_ms
+            end
+            redis.call('ZADD', due_key, next_due_ms, task.proxyId)
+            touch(due_key, tonumber(record.expiresAtMs))
+        end
+    end
+    return finish({status = 'ok'})
+end
+
+return finish({status = 'invalid'})

internal/adapters/redisactivity/scripts/upsert.lua

@@ -8,6 +8,12 @@ local state_inventory_key = KEYS[7]
 local owners_key = KEYS[8]
 local owner_expiry_key = KEYS[9]
 local operation_key = KEYS[10]
+local health_due_key = KEYS[11]
+local health_queued_key = KEYS[12]
+local health_leases_key = KEYS[13]
+local health_tasks_key = KEYS[14]
+local health_task_expiry_key = KEYS[15]
+local health_ref_task_key = KEYS[16]
 
 local now_ms = tonumber(ARGV[1])
 local cleanup_limit = tonumber(ARGV[2])
@@ -135,6 +141,15 @@ local function remove_proxy(proxy_id)
     remove_worker_owned(proxy_id)
     redis.call('HDEL', owners_key, proxy_id)
     redis.call('ZREM', owner_expiry_key, proxy_id)
+    redis.call('ZREM', health_due_key, proxy_id)
+    local task_id = redis.call('HGET', health_ref_task_key, proxy_id)
+    if task_id then
+        redis.call('ZREM', health_queued_key, task_id)
+        redis.call('ZREM', health_leases_key, task_id)
+        redis.call('ZREM', health_task_expiry_key, task_id)
+        redis.call('HDEL', health_tasks_key, task_id)
+        redis.call('HDEL', health_ref_task_key, proxy_id)
+    end
 end
 
 local function cleanup_expired()
@@ -256,6 +271,7 @@ for _, candidate in ipairs(candidates) do
             redis.call('HSET', unique_key, candidate.uniqueDigest, candidate.proxyId)
             redis.call('HSET', idkeys_key, candidate.proxyId, candidate.uniqueDigest)
             redis.call('ZADD', expiry_key, incoming.expiresAtMs, candidate.proxyId)
+			redis.call('ZADD', health_due_key, now_ms, candidate.proxyId)
             sync_owned(candidate.proxyId, incoming)
             sync_worker_owned(candidate.proxyId, incoming)
             if is_managed(incoming.state) then
@@ -279,6 +295,7 @@ if max_expiry_ms > 0 then
     touch(available_key, max_expiry_ms)
     touch(inventory_key, max_expiry_ms)
     touch(state_inventory_key, max_expiry_ms)
+		touch(health_due_key, max_expiry_ms)
     touch(owners_key, max_expiry_ms)
     touch(owner_expiry_key, max_expiry_ms)
 end

internal/adapters/redisactivity/upsert.go

@@ -204,7 +204,8 @@ func (a *Adapter) upsertChunk(
 	result, err := runScript(ctx, a.client, upsertScript, []string{
 		a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
 		a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
-		a.keys.operation(operationID),
+		a.keys.operation(operationID), a.keys.healthDue, a.keys.healthQueued,
+		a.keys.healthLeases, a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask,
 	}, observedAt.UnixMilli(), a.options.CleanupLimit, maxSize, operationTTLMillis(a.options.OperationTTL), string(payload))
 	if err != nil {
 		return upsertScriptReply{}, err

internal/controller/health/scheduler.go

@@ -120,8 +120,13 @@ func (planner *Planner) Plan(now time.Time, inFlight, maxTasks int, candidates [
 	deadline := now.UTC().Add(planner.policy.Timeout)
 	result := make([]PlannedTask, len(eligible))
 	for index, item := range eligible {
+		nextDue, err := planner.NextDue(now.UTC(), healthDomain.CandidateIdentity(item.candidate))
+		if err != nil {
+			return nil, err
+		}
 		result[index] = PlannedTask{
-			Candidate: item.candidate, Priority: item.priority, Deadline: deadline, Attempts: planner.policy.MaxAttempts,
+			Candidate: item.candidate, Priority: item.priority, Deadline: deadline, NextDue: nextDue,
+			Attempts: planner.policy.MaxAttempts,
 		}
 	}
 	return result, nil

internal/controller/health/task_broker.go

@@ -3,12 +3,9 @@ package health
 import (
 	"context"
 	"crypto/rand"
-	"crypto/sha256"
 	"encoding/hex"
-	"errors"
 	"reflect"
 	"sort"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -18,13 +15,13 @@ import (
 )
 
 var (
-	ErrInvalidTaskBroker = errors.New("invalid health task broker")
+	ErrInvalidTaskBroker = healthDomain.ErrInvalidTaskBroker
-	ErrInvalidTaskClaim  = errors.New("invalid health task claim")
+	ErrInvalidTaskClaim  = healthDomain.ErrInvalidTaskClaim
-	ErrInvalidLeasedTask = errors.New("invalid leased health task")
+	ErrInvalidLeasedTask = healthDomain.ErrInvalidLeasedTask
-	ErrTaskNotFound      = errors.New("health task not found")
+	ErrTaskNotFound      = healthDomain.ErrTaskNotFound
-	ErrTaskLeaseExpired  = errors.New("health task lease expired")
+	ErrTaskLeaseExpired  = healthDomain.ErrTaskLeaseExpired
-	ErrTaskLeaseNotOwned = errors.New("health task lease is not owned by checker")
+	ErrTaskLeaseNotOwned = healthDomain.ErrTaskLeaseNotOwned
-	ErrTaskObservation   = errors.New("health observation does not match task")
+	ErrTaskObservation   = healthDomain.ErrTaskObservation
 )
 
 const defaultMaxTasksPerClaim = 128
@@ -385,10 +382,7 @@ func validateTaskMaterial(material TaskMaterial) error {
 }
 
 func deterministicTaskID(task PlannedTask) string {
-	payload := healthDomain.CandidateIdentity(task.Candidate) + "\x00" + task.Deadline.UTC().Format(time.RFC3339Nano) + "\x00" +
+	return healthDomain.TaskIDFor(task)
-		strconv.Itoa(task.Attempts)
-	digest := sha256.Sum256([]byte(payload))
-	return "check_" + hex.EncodeToString(digest[:])
 }
 
 func leasedTask(taskID, leaseToken string, plan PlannedTask, material TaskMaterial) LeasedTask {

internal/domain/health/health.go

@@ -24,6 +24,13 @@ var (
 	ErrConflictingObservation  = errors.New("conflicting health observation replay")
 	ErrInvalidFailureThreshold = errors.New("invalid health failure threshold")
 	ErrInvalidCheckPreparation = errors.New("invalid health check preparation")
+	ErrInvalidTaskBroker       = errors.New("invalid health task broker")
+	ErrInvalidTaskClaim        = errors.New("invalid health task claim")
+	ErrInvalidLeasedTask       = errors.New("invalid leased health task")
+	ErrTaskNotFound            = errors.New("health task not found")
+	ErrTaskLeaseExpired        = errors.New("health task lease expired")
+	ErrTaskLeaseNotOwned       = errors.New("health task lease is not owned by checker")
+	ErrTaskObservation         = errors.New("health observation does not match task")
 )
 
 type Level string

internal/domain/health/task.go

@@ -2,7 +2,10 @@ package health
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
+	"strconv"
 	"time"
 
 	proxyDomain "proxy-pool/internal/domain/proxy"
@@ -39,9 +42,20 @@ type PlannedTask struct {
 	Candidate Candidate
 	Priority  Priority
 	Deadline  time.Time
+	NextDue   time.Time
 	Attempts  int
 }
 
+// TaskIDFor derives an idempotent task identity from all execution-relevant
+// plan fields. Multiple Controller replicas can therefore offer the same plan
+// without creating duplicate check work.
+func TaskIDFor(task PlannedTask) string {
+	payload := CandidateIdentity(task.Candidate) + "\x00" + task.Deadline.UTC().Format(time.RFC3339Nano) + "\x00" +
+		task.NextDue.UTC().Format(time.RFC3339Nano) + "\x00" + strconv.Itoa(task.Attempts)
+	digest := sha256.Sum256([]byte(payload))
+	return "check_" + hex.EncodeToString(digest[:])
+}
+
 // 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.