diff --git a/internal/adapters/redisactivity/adapter.go b/internal/adapters/redisactivity/adapter.go index 0fdaf69..2eb3e51 100644 --- a/internal/adapters/redisactivity/adapter.go +++ b/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{ diff --git a/internal/adapters/redisactivity/health_tasks.go b/internal/adapters/redisactivity/health_tasks.go new file mode 100644 index 0000000..c7dc9e1 --- /dev/null +++ b/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 +} diff --git a/internal/adapters/redisactivity/health_tasks_integration_test.go b/internal/adapters/redisactivity/health_tasks_integration_test.go new file mode 100644 index 0000000..6f56b5e --- /dev/null +++ b/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) + } +} diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index 5702f0d..861730a 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/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 } diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 842686c..7305a6b 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/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) diff --git a/internal/adapters/redisactivity/scripts/health_tasks.lua b/internal/adapters/redisactivity/scripts/health_tasks.lua new file mode 100644 index 0000000..6f68b92 --- /dev/null +++ b/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'}) diff --git a/internal/adapters/redisactivity/scripts/upsert.lua b/internal/adapters/redisactivity/scripts/upsert.lua index 1b43dc0..a8ac248 100644 --- a/internal/adapters/redisactivity/scripts/upsert.lua +++ b/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 diff --git a/internal/adapters/redisactivity/upsert.go b/internal/adapters/redisactivity/upsert.go index 31cf9f8..e76e4b5 100644 --- a/internal/adapters/redisactivity/upsert.go +++ b/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 diff --git a/internal/controller/health/scheduler.go b/internal/controller/health/scheduler.go index 928698d..e3f327e 100644 --- a/internal/controller/health/scheduler.go +++ b/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 diff --git a/internal/controller/health/task_broker.go b/internal/controller/health/task_broker.go index b8f9627..30d5e07 100644 --- a/internal/controller/health/task_broker.go +++ b/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") - 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") + ErrInvalidTaskBroker = healthDomain.ErrInvalidTaskBroker + ErrInvalidTaskClaim = healthDomain.ErrInvalidTaskClaim + ErrInvalidLeasedTask = healthDomain.ErrInvalidLeasedTask + ErrTaskNotFound = healthDomain.ErrTaskNotFound + ErrTaskLeaseExpired = healthDomain.ErrTaskLeaseExpired + ErrTaskLeaseNotOwned = healthDomain.ErrTaskLeaseNotOwned + 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" + - strconv.Itoa(task.Attempts) - digest := sha256.Sum256([]byte(payload)) - return "check_" + hex.EncodeToString(digest[:]) + return healthDomain.TaskIDFor(task) } func leasedTask(taskID, leaseToken string, plan PlannedTask, material TaskMaterial) LeasedTask { diff --git a/internal/domain/health/health.go b/internal/domain/health/health.go index ee6b5ab..b2f42a6 100644 --- a/internal/domain/health/health.go +++ b/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 diff --git a/internal/domain/health/task.go b/internal/domain/health/task.go index 27babc6..f3d7739 100644 --- a/internal/domain/health/task.go +++ b/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.