diff --git a/internal/adapters/redisactivity/extract.go b/internal/adapters/redisactivity/extract.go index 8a5832c..e0601f2 100644 --- a/internal/adapters/redisactivity/extract.go +++ b/internal/adapters/redisactivity/extract.go @@ -90,6 +90,8 @@ func (a *Adapter) Extract(ctx context.Context, command extractionDomain.Command) operationKey, idempotencyKey, } keys = append(keys, a.extractionDriverKeys(digestInput)...) + keys = append(keys, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases, + a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask) idempotencyTTL := command.IdempotencyTTL if idempotencyTTL == 0 { idempotencyTTL = defaultRedisIdempotencyTTL diff --git a/internal/adapters/redisactivity/health.go b/internal/adapters/redisactivity/health.go index 15f8e26..505b3bc 100644 --- a/internal/adapters/redisactivity/health.go +++ b/internal/adapters/redisactivity/health.go @@ -44,7 +44,8 @@ func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpd result, err := runScript(ctx, a.client, healthScript, []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, }, update.CheckedAt.UnixMilli(), string(update.NextState), int64(update.Latency), a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), update.ProxyID) if err != nil { @@ -109,7 +110,8 @@ func (a *Adapter) ApplyGlobalObservation( result, err := runScript(ctx, a.client, healthScript, []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, }, normalized.ObservedAt.UnixMilli(), "", int64(normalized.Latency), a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), normalized.ProxyID, "global", success, command.MaxConsecutiveFailures, normalized.TaskID, hex.EncodeToString(digest[:])) diff --git a/internal/adapters/redisactivity/health_tasks_integration_test.go b/internal/adapters/redisactivity/health_tasks_integration_test.go index 7310e3e..39007e1 100644 --- a/internal/adapters/redisactivity/health_tasks_integration_test.go +++ b/internal/adapters/redisactivity/health_tasks_integration_test.go @@ -4,6 +4,7 @@ package redisactivity import ( "context" + "errors" "testing" "time" @@ -72,3 +73,53 @@ func TestRedisHealthTasksLeaseAndRescheduleBasicChecks(t *testing.T) { t.Fatalf("DueCandidates(next) = (%+v, %v)", next, err) } } + +func TestRedisHealthTaskSweepReleasesLeaseAndUpstreamCapacity(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.Second, MaxSize: 1, + Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")}, + }); err != nil { + t.Fatalf("UpsertFetched(): %v", err) + } + candidates, err := fixture.Adapter.DueCandidatesForUpstream(ctx, "provider-a", now, 1) + if err != nil || len(candidates) != 1 { + t.Fatalf("DueCandidatesForUpstream() = (%+v, %v)", candidates, err) + } + planner, err := controllerHealth.NewPlanner(controllerHealth.SchedulePolicy{ + Interval: 30 * time.Second, MaxInFlight: 1, Timeout: 10 * time.Second, MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + plans, err := planner.Plan(now, 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 { + t.Fatalf("Claim() = (%+v, %v)", claimed, err) + } + expiredAt := now.Add(2 * time.Second) + if removed, err := fixture.Adapter.SweepExpired(ctx, expiredAt, 1); err != nil || removed != 1 { + t.Fatalf("SweepExpired() = (%d, %v)", removed, err) + } + if inFlight, err := fixture.Adapter.InFlightForUpstream(ctx, "provider-a", expiredAt); err != nil || inFlight != 0 { + t.Fatalf("InFlightForUpstream() = (%d, %v)", inFlight, err) + } + observation := healthDomain.Observation{ + TaskID: claimed[0].TaskID, ProxyID: claimed[0].ProxyID, Level: healthDomain.LevelBasic, + Success: true, Latency: time.Millisecond, ObservedAt: now, + } + if err := fixture.Adapter.AuthorizeObservation(ctx, "checker-a", claimed[0].LeaseToken, observation, expiredAt); !errors.Is(err, healthDomain.ErrTaskNotFound) { + t.Fatalf("AuthorizeObservation(after sweep) error = %v, want ErrTaskNotFound", err) + } +} diff --git a/internal/adapters/redisactivity/maintenance.go b/internal/adapters/redisactivity/maintenance.go index c46260e..bc76160 100644 --- a/internal/adapters/redisactivity/maintenance.go +++ b/internal/adapters/redisactivity/maintenance.go @@ -79,7 +79,8 @@ func (a *Adapter) runMaintenance( result, err := runScript(ctx, a.client, sweepScript, []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, }, operation, now.UnixMilli(), limit, upstreamID, operationTTLMillis(a.options.OperationTTL)) if err != nil { return maintenanceScriptReply{}, err diff --git a/internal/adapters/redisactivity/ownership.go b/internal/adapters/redisactivity/ownership.go index 3bb6389..a985a38 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -224,7 +224,8 @@ func (a *Adapter) runOwnership( result, err := runScript(ctx, a.client, ownershipScript, []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.epoch, operationKey, + a.keys.epoch, operationKey, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases, + a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, }, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit, nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID)) if err != nil { diff --git a/internal/adapters/redisactivity/scripts/extract.lua b/internal/adapters/redisactivity/scripts/extract.lua index b4b4423..f60eec0 100644 --- a/internal/adapters/redisactivity/scripts/extract.lua +++ b/internal/adapters/redisactivity/scripts/extract.lua @@ -9,6 +9,12 @@ local owners_key = KEYS[8] local owner_expiry_key = KEYS[9] local operation_key = KEYS[10] local idempotency_key = KEYS[11] +local health_due_key = KEYS[#KEYS - 5] +local health_queued_key = KEYS[#KEYS - 4] +local health_leases_key = KEYS[#KEYS - 3] +local health_tasks_key = KEYS[#KEYS - 2] +local health_task_expiry_key = KEYS[#KEYS - 1] +local health_ref_task_key = KEYS[#KEYS] local now_ms = tonumber(ARGV[1]) local requested = tonumber(ARGV[2]) @@ -131,6 +137,31 @@ local function remove_worker_owned(proxy_id) end end +local function remove_health_task(proxy_id) + redis.call('ZREM', health_due_key, proxy_id) + local task_id = redis.call('HGET', health_ref_task_key, proxy_id) + if not task_id then + return + end + local raw = redis.call('HGET', health_tasks_key, task_id) + if raw then + local decoded, task = pcall(cjson.decode, raw) + if decoded and type(task) == 'table' then + if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then + redis.call('ZREM', task.checkerLeaseKey, task_id) + end + if type(task.upstreamTasksKey) == 'string' and task.upstreamTasksKey ~= '' then + redis.call('ZREM', task.upstreamTasksKey, task_id) + end + end + end + 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 + local function remove_proxy(proxy_id) local raw = redis.call('HGET', records_key, proxy_id) local record = nil @@ -158,6 +189,7 @@ 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) + remove_health_task(proxy_id) end local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now_ms, 'LIMIT', 0, cleanup_limit) @@ -256,7 +288,7 @@ end local driver_key = available_key local driver_size = redis.call('ZCARD', available_key) -for index = 12, #KEYS do +for index = 12, #KEYS - 6 do local size = redis.call('ZCARD', KEYS[index]) if size < driver_size then driver_key = KEYS[index] @@ -337,6 +369,7 @@ for index = 1, selected_count do increment_state(record.sourceUpstream, record.state) local encoded = cjson.encode(record) redis.call('HSET', records_key, selected.id, encoded) + remove_health_task(selected.id) local tags = record.tags or {} local item = { diff --git a/internal/adapters/redisactivity/scripts/health.lua b/internal/adapters/redisactivity/scripts/health.lua index 6c876ca..50ba872 100644 --- a/internal/adapters/redisactivity/scripts/health.lua +++ b/internal/adapters/redisactivity/scripts/health.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 checked_at_ms = tonumber(ARGV[1]) local next_state = ARGV[2] @@ -93,6 +99,31 @@ local function remove_worker_owned(id) end end +local function remove_health_task(proxy_id) + redis.call('ZREM', health_due_key, proxy_id) + local task_id = redis.call('HGET', health_ref_task_key, proxy_id) + if not task_id then + return + end + local raw = redis.call('HGET', health_tasks_key, task_id) + if raw then + local decoded, task = pcall(cjson.decode, raw) + if decoded and type(task) == 'table' then + if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then + redis.call('ZREM', task.checkerLeaseKey, task_id) + end + if type(task.upstreamTasksKey) == 'string' and task.upstreamTasksKey ~= '' then + redis.call('ZREM', task.upstreamTasksKey, task_id) + end + end + end + 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 + local function sync_worker_owned(id, record) local raw = redis.call('HGET', owners_key, id) if not raw then @@ -140,6 +171,7 @@ local function remove_proxy(id) remove_worker_owned(id) redis.call('HDEL', owners_key, id) redis.call('ZREM', owner_expiry_key, id) + remove_health_task(id) end local function cleanup_expired() diff --git a/internal/adapters/redisactivity/scripts/ownership.lua b/internal/adapters/redisactivity/scripts/ownership.lua index f5a279d..bdeb4e1 100644 --- a/internal/adapters/redisactivity/scripts/ownership.lua +++ b/internal/adapters/redisactivity/scripts/ownership.lua @@ -9,6 +9,12 @@ local owners_key = KEYS[8] local owner_expiry_key = KEYS[9] local epoch_key = KEYS[10] local operation_key = KEYS[11] +local health_due_key = KEYS[12] +local health_queued_key = KEYS[13] +local health_leases_key = KEYS[14] +local health_tasks_key = KEYS[15] +local health_task_expiry_key = KEYS[16] +local health_ref_task_key = KEYS[17] local operation = ARGV[1] local operation_ttl_ms = tonumber(ARGV[2]) @@ -118,6 +124,31 @@ local function add_available(id, record, at_ms) end end +local function remove_health_task(proxy_id) + redis.call('ZREM', health_due_key, proxy_id) + local task_id = redis.call('HGET', health_ref_task_key, proxy_id) + if not task_id then + return + end + local raw = redis.call('HGET', health_tasks_key, task_id) + if raw then + local decoded, task = pcall(cjson.decode, raw) + if decoded and type(task) == 'table' then + if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then + redis.call('ZREM', task.checkerLeaseKey, task_id) + end + if type(task.upstreamTasksKey) == 'string' and task.upstreamTasksKey ~= '' then + redis.call('ZREM', task.upstreamTasksKey, task_id) + end + end + end + 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 + local function remove_proxy(id) local raw = redis.call('HGET', records_key, id) local record = nil @@ -144,6 +175,7 @@ local function remove_proxy(id) redis.call('ZREM', expiry_key, id) redis.call('HDEL', owners_key, id) redis.call('ZREM', owner_expiry_key, id) + remove_health_task(id) end local function cleanup_hard_expired(at_ms) diff --git a/internal/adapters/redisactivity/scripts/status.lua b/internal/adapters/redisactivity/scripts/status.lua index 89c62d0..8c668e6 100644 --- a/internal/adapters/redisactivity/scripts/status.lua +++ b/internal/adapters/redisactivity/scripts/status.lua @@ -7,6 +7,12 @@ local inventory_key = KEYS[6] local state_inventory_key = KEYS[7] local owners_key = KEYS[8] local owner_expiry_key = KEYS[9] +local health_due_key = KEYS[10] +local health_queued_key = KEYS[11] +local health_leases_key = KEYS[12] +local health_tasks_key = KEYS[13] +local health_task_expiry_key = KEYS[14] +local health_ref_task_key = KEYS[15] local now_ms = tonumber(ARGV[1]) local cleanup_limit = tonumber(ARGV[2]) @@ -59,6 +65,31 @@ local function remove_available(proxy_id, record) end end +local function remove_health_task(proxy_id) + redis.call('ZREM', health_due_key, proxy_id) + local task_id = redis.call('HGET', health_ref_task_key, proxy_id) + if not task_id then + return + end + local raw = redis.call('HGET', health_tasks_key, task_id) + if raw then + local decoded, task = pcall(cjson.decode, raw) + if decoded and type(task) == 'table' then + if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then + redis.call('ZREM', task.checkerLeaseKey, task_id) + end + if type(task.upstreamTasksKey) == 'string' and task.upstreamTasksKey ~= '' then + redis.call('ZREM', task.upstreamTasksKey, task_id) + end + end + end + 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 + local function remove_proxy(proxy_id) local raw = redis.call('HGET', records_key, proxy_id) local record = nil @@ -84,6 +115,7 @@ local function remove_proxy(proxy_id) redis.call('ZREM', expiry_key, proxy_id) redis.call('HDEL', owners_key, proxy_id) redis.call('ZREM', owner_expiry_key, proxy_id) + remove_health_task(proxy_id) end local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now_ms, 'LIMIT', 0, cleanup_limit) diff --git a/internal/adapters/redisactivity/scripts/sweep.lua b/internal/adapters/redisactivity/scripts/sweep.lua index 8a23a0c..01417f8 100644 --- a/internal/adapters/redisactivity/scripts/sweep.lua +++ b/internal/adapters/redisactivity/scripts/sweep.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 operation = ARGV[1] local now_ms = tonumber(ARGV[2]) @@ -91,6 +97,31 @@ local function remove_worker_owned(proxy_id) end end +local function remove_health_task(proxy_id) + redis.call('ZREM', health_due_key, proxy_id) + local task_id = redis.call('HGET', health_ref_task_key, proxy_id) + if not task_id then + return + end + local raw = redis.call('HGET', health_tasks_key, task_id) + if raw then + local decoded, task = pcall(cjson.decode, raw) + if decoded and type(task) == 'table' then + if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then + redis.call('ZREM', task.checkerLeaseKey, task_id) + end + if type(task.upstreamTasksKey) == 'string' and task.upstreamTasksKey ~= '' then + redis.call('ZREM', task.upstreamTasksKey, task_id) + end + end + end + 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 + local function remove_proxy(proxy_id) local raw = redis.call('HGET', records_key, proxy_id) local record = nil @@ -118,6 +149,7 @@ 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) + remove_health_task(proxy_id) end local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now_ms, 'LIMIT', 0, limit) diff --git a/internal/adapters/redisactivity/scripts/upsert.lua b/internal/adapters/redisactivity/scripts/upsert.lua index a8ac248..bb83cbf 100644 --- a/internal/adapters/redisactivity/scripts/upsert.lua +++ b/internal/adapters/redisactivity/scripts/upsert.lua @@ -144,6 +144,18 @@ local function remove_proxy(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 + local task_raw = redis.call('HGET', health_tasks_key, task_id) + if task_raw then + local decoded, task = pcall(cjson.decode, task_raw) + if decoded and type(task) == 'table' then + if type(task.checkerLeaseKey) == 'string' and task.checkerLeaseKey ~= '' then + redis.call('ZREM', task.checkerLeaseKey, task_id) + end + if type(task.upstreamTasksKey) == 'string' and task.upstreamTasksKey ~= '' then + redis.call('ZREM', task.upstreamTasksKey, task_id) + end + end + end 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) diff --git a/internal/adapters/redisactivity/status.go b/internal/adapters/redisactivity/status.go index ecce12a..31b95e6 100644 --- a/internal/adapters/redisactivity/status.go +++ b/internal/adapters/redisactivity/status.go @@ -39,6 +39,8 @@ func (a *Adapter) ReadStateInventory( result, err := runScript(ctx, a.client, statusScript, []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.healthDue, a.keys.healthQueued, a.keys.healthLeases, a.keys.healthTasks, + a.keys.healthTaskExpiry, a.keys.healthRefTask, }, now.UnixMilli(), a.options.CleanupLimit, string(payload)) if err != nil { return nil, err