diff --git a/internal/adapters/redisactivity/adapter_test.go b/internal/adapters/redisactivity/adapter_test.go index 64f78f3..17ab60d 100644 --- a/internal/adapters/redisactivity/adapter_test.go +++ b/internal/adapters/redisactivity/adapter_test.go @@ -10,6 +10,7 @@ import ( "github.com/redis/go-redis/v9" + "proxy-pool/internal/domain/activitypool" extractionDomain "proxy-pool/internal/domain/extraction" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/platform/credentials" @@ -92,6 +93,7 @@ func TestNewBuildsClusterSafeKeyspaceAndHashesDynamicTokens(t *testing.T) { adapter.keys.records, adapter.keys.unique, adapter.keys.idkeys, adapter.keys.expiry, adapter.keys.available, adapter.keys.owners, adapter.keys.ownerExpiry, adapter.keys.epoch, adapter.keys.inventory, + adapter.keys.stateInventory, } for _, key := range staticKeys { if strings.Count(key, "{activity}") != 1 || strings.Count(key, "{") != 1 || strings.Count(key, "}") != 1 { @@ -118,6 +120,38 @@ func TestNewBuildsClusterSafeKeyspaceAndHashesDynamicTokens(t *testing.T) { } } +func TestStateInventoryFieldsAreCollisionFreeAndStatusDoesNotScanRecords(t *testing.T) { + t.Parallel() + first := stateInventoryField("provider:a", "FETCHED") + second := stateInventoryField("provider", "a:FETCHED") + if first == second || first == "" || second == "" { + t.Fatalf("state inventory fields collide: %q and %q", first, second) + } + upper := strings.ToUpper(statusSource) + if strings.Contains(upper, "HGETALL") || strings.Contains(upper, "HSCAN") { + t.Fatal("status script scans Redis hashes") + } +} + +func TestReadStateInventoryRejectsInvalidCalls(t *testing.T) { + t.Parallel() + now := time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC) + var adapter *Adapter + if _, err := adapter.ReadStateInventory(context.Background(), []string{"provider-a"}, now); !errors.Is(err, activitypool.ErrInvalidInventory) { + t.Fatalf("nil adapter error = %v", err) + } + if _, err := adapter.ReadStateInventory(nil, []string{"provider-a"}, now); !errors.Is(err, activitypool.ErrInvalidInventory) { + t.Fatalf("nil context error = %v", err) + } + adapter = &Adapter{} + if _, err := adapter.ReadStateInventory(context.Background(), []string{""}, now); !errors.Is(err, activitypool.ErrInvalidInventory) { + t.Fatalf("empty upstream error = %v", err) + } + if _, err := adapter.ReadStateInventory(context.Background(), []string{"provider-a"}, time.Time{}); !errors.Is(err, activitypool.ErrInvalidInventory) { + t.Fatalf("zero time error = %v", err) + } +} + func TestProxyRecordCodecIsDeterministicStrictAndRedacted(t *testing.T) { t.Parallel() record := proxyRecord{ diff --git a/internal/adapters/redisactivity/extract.go b/internal/adapters/redisactivity/extract.go index 17cc155..8a5832c 100644 --- a/internal/adapters/redisactivity/extract.go +++ b/internal/adapters/redisactivity/extract.go @@ -86,7 +86,8 @@ func (a *Adapter) Extract(ctx context.Context, command extractionDomain.Command) } keys := []string{ a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available, - a.keys.inventory, a.keys.owners, a.keys.ownerExpiry, operationKey, idempotencyKey, + a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, + operationKey, idempotencyKey, } keys = append(keys, a.extractionDriverKeys(digestInput)...) idempotencyTTL := command.IdempotencyTTL diff --git a/internal/adapters/redisactivity/health.go b/internal/adapters/redisactivity/health.go index b9012ff..21f0be9 100644 --- a/internal/adapters/redisactivity/health.go +++ b/internal/adapters/redisactivity/health.go @@ -25,7 +25,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.owners, a.keys.ownerExpiry, a.keys.operation(operationID), + a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, + a.keys.operation(operationID), }, update.CheckedAt.UnixMilli(), string(update.NextState), int64(update.Latency), a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), update.ProxyID) if err != nil { diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index 1a19ce0..aded371 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -9,34 +9,40 @@ import ( const redisKeyPrefix = "pp:{activity}:" type keyspace struct { - prefix string - records string - unique string - idkeys string - expiry string - available string - owners string - ownerExpiry string - epoch string - inventory string + prefix string + records string + unique string + idkeys string + expiry string + available string + owners string + ownerExpiry string + epoch string + inventory string + stateInventory string } func newKeyspace(namespace string) keyspace { prefix := redisKeyPrefix + namespace return keyspace{ - prefix: prefix, - records: prefix + ":records", - unique: prefix + ":unique", - idkeys: prefix + ":idkeys", - expiry: prefix + ":expiry", - available: prefix + ":available", - owners: prefix + ":owners", - ownerExpiry: prefix + ":owner-expiry", - epoch: prefix + ":epoch", - inventory: prefix + ":inventory", + prefix: prefix, + records: prefix + ":records", + unique: prefix + ":unique", + idkeys: prefix + ":idkeys", + expiry: prefix + ":expiry", + available: prefix + ":available", + owners: prefix + ":owners", + ownerExpiry: prefix + ":owner-expiry", + epoch: prefix + ":epoch", + inventory: prefix + ":inventory", + stateInventory: prefix + ":state-inventory", } } +func stateInventoryField(upstreamID, state string) string { + return strconv.Itoa(len(upstreamID)) + ":" + upstreamID + ":" + state +} + func (keys keyspace) idempotency(clientID, idempotencyKey string) string { return keys.prefix + ":idem:" + digestParts(clientID, idempotencyKey) } diff --git a/internal/adapters/redisactivity/maintenance.go b/internal/adapters/redisactivity/maintenance.go index 3aa6572..c46260e 100644 --- a/internal/adapters/redisactivity/maintenance.go +++ b/internal/adapters/redisactivity/maintenance.go @@ -78,7 +78,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.owners, a.keys.ownerExpiry, a.keys.operation(operationID), + a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, + a.keys.operation(operationID), }, 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 9af9dfe..9cb350b 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -223,7 +223,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.owners, a.keys.ownerExpiry, a.keys.epoch, operationKey, + a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, + a.keys.epoch, operationKey, }, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit, nowMS, proxyID, workerID, epoch, value, active, reserved) if err != nil { diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 7067498..5f41a28 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -58,6 +58,22 @@ type maintenanceScriptReply struct { Count int `json:"count"` } +type statusScriptReply struct { + Status scriptStatus `json:"status"` + Inventories []statusScriptInventory `json:"inventories"` +} + +type statusScriptInventory struct { + UpstreamID string `json:"upstreamId"` + Fetched int64 `json:"fetched"` + Checking int64 `json:"checking"` + Available int64 `json:"available"` + Suspect int64 `json:"suspect"` + Draining int64 `json:"draining"` + Unhealthy int64 `json:"unhealthy"` + Extracted int64 `json:"extracted"` +} + //go:embed scripts/upsert.lua var upsertSource string @@ -73,12 +89,16 @@ var ownershipSource string //go:embed scripts/sweep.lua var sweepSource string +//go:embed scripts/status.lua +var statusSource string + var ( upsertScript = redis.NewScript(upsertSource) healthScript = redis.NewScript(healthSource) extractScript = redis.NewScript(extractSource) ownershipScript = redis.NewScript(ownershipSource) sweepScript = redis.NewScript(sweepSource) + statusScript = redis.NewScript(statusSource) ) func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) { diff --git a/internal/adapters/redisactivity/scripts/extract.lua b/internal/adapters/redisactivity/scripts/extract.lua index c3e4b6f..933d245 100644 --- a/internal/adapters/redisactivity/scripts/extract.lua +++ b/internal/adapters/redisactivity/scripts/extract.lua @@ -4,10 +4,11 @@ local idkeys_key = KEYS[3] local expiry_key = KEYS[4] local available_key = KEYS[5] local inventory_key = KEYS[6] -local owners_key = KEYS[7] -local owner_expiry_key = KEYS[8] -local operation_key = KEYS[9] -local idempotency_key = KEYS[10] +local state_inventory_key = KEYS[7] +local owners_key = KEYS[8] +local owner_expiry_key = KEYS[9] +local operation_key = KEYS[10] +local idempotency_key = KEYS[11] local now_ms = tonumber(ARGV[1]) local requested = tonumber(ARGV[2]) @@ -73,6 +74,33 @@ local function decrement_inventory(upstream) end end +local function state_field(upstream, state) + return string.len(upstream) .. ':' .. upstream .. ':' .. state +end + +local function is_counted(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' or state == 'UNHEALTHY' or state == 'EXTRACTED' +end + +local function increment_state(upstream, state) + if not is_counted(state) then + return + end + redis.call('HINCRBY', state_inventory_key, state_field(upstream, state), 1) +end + +local function decrement_state(upstream, state) + if type(upstream) ~= 'string' or upstream == '' or not is_counted(state) then + return + end + 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 remove_available(proxy_id, record) redis.call('ZREM', available_key, proxy_id) local index_keys = record and record.indexKeys @@ -95,6 +123,9 @@ local function remove_proxy(proxy_id) if decoded and type(record) == 'table' and is_managed(record.state) then decrement_inventory(record.sourceUpstream) end + if decoded and type(record) == 'table' then + decrement_state(record.sourceUpstream, record.state) + end else redis.call('ZREM', available_key, proxy_id) end @@ -205,7 +236,7 @@ end local driver_key = available_key local driver_size = redis.call('ZCARD', available_key) -for index = 11, #KEYS do +for index = 12, #KEYS do local size = redis.call('ZCARD', KEYS[index]) if size < driver_size then driver_key = KEYS[index] @@ -281,7 +312,9 @@ for index = 1, selected_count do if is_managed(record.state) then decrement_inventory(record.sourceUpstream) end + decrement_state(record.sourceUpstream, record.state) record.state = 'EXTRACTED' + increment_state(record.sourceUpstream, record.state) local encoded = cjson.encode(record) redis.call('HSET', records_key, selected.id, encoded) diff --git a/internal/adapters/redisactivity/scripts/health.lua b/internal/adapters/redisactivity/scripts/health.lua index 0d95661..8d2490e 100644 --- a/internal/adapters/redisactivity/scripts/health.lua +++ b/internal/adapters/redisactivity/scripts/health.lua @@ -4,9 +4,10 @@ local idkeys_key = KEYS[3] local expiry_key = KEYS[4] local available_key = KEYS[5] local inventory_key = KEYS[6] -local owners_key = KEYS[7] -local owner_expiry_key = KEYS[8] -local operation_key = KEYS[9] +local state_inventory_key = KEYS[7] +local owners_key = KEYS[8] +local owner_expiry_key = KEYS[9] +local operation_key = KEYS[10] local checked_at_ms = tonumber(ARGV[1]) local next_state = ARGV[2] @@ -35,6 +36,33 @@ local function decrement_inventory(upstream) end end +local function state_field(upstream, state) + return string.len(upstream) .. ':' .. upstream .. ':' .. state +end + +local function is_counted(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' or state == 'UNHEALTHY' or state == 'EXTRACTED' +end + +local function increment_state(upstream, state) + if not is_counted(state) then + return + end + redis.call('HINCRBY', state_inventory_key, state_field(upstream, state), 1) +end + +local function decrement_state(upstream, state) + if type(upstream) ~= 'string' or upstream == '' or not is_counted(state) then + return + end + 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 remove_available(id, record) redis.call('ZREM', available_key, id) for _, index_key in ipairs(record and record.indexKeys or {}) do @@ -51,6 +79,7 @@ local function remove_proxy(id) if is_managed(record.state) then decrement_inventory(record.sourceUpstream) end + decrement_state(record.sourceUpstream, record.state) else redis.call('ZREM', available_key, id) end @@ -132,7 +161,8 @@ if record.state ~= next_state and not (transitions[record.state] and transitions return finish({status = 'invalid'}) end -local was_managed = is_managed(record.state) +local previous_state = record.state +local was_managed = is_managed(previous_state) local will_be_managed = is_managed(next_state) remove_available(proxy_id, record) record.state = next_state @@ -146,6 +176,10 @@ if was_managed and not will_be_managed then elseif not was_managed and will_be_managed then redis.call('HINCRBY', inventory_key, record.sourceUpstream, 1) end +if previous_state ~= next_state then + decrement_state(record.sourceUpstream, previous_state) + increment_state(record.sourceUpstream, next_state) +end local encoded = cjson.encode(record) redis.call('HSET', records_key, proxy_id, encoded) @@ -163,6 +197,7 @@ touch(idkeys_key, tonumber(record.expiresAtMs)) touch(expiry_key, tonumber(record.expiresAtMs)) touch(available_key, tonumber(record.expiresAtMs)) touch(inventory_key, tonumber(record.expiresAtMs)) +touch(state_inventory_key, tonumber(record.expiresAtMs)) touch(owners_key, tonumber(record.expiresAtMs)) touch(owner_expiry_key, tonumber(record.expiresAtMs)) diff --git a/internal/adapters/redisactivity/scripts/ownership.lua b/internal/adapters/redisactivity/scripts/ownership.lua index 3506591..9312274 100644 --- a/internal/adapters/redisactivity/scripts/ownership.lua +++ b/internal/adapters/redisactivity/scripts/ownership.lua @@ -4,10 +4,11 @@ local idkeys_key = KEYS[3] local expiry_key = KEYS[4] local available_key = KEYS[5] local inventory_key = KEYS[6] -local owners_key = KEYS[7] -local owner_expiry_key = KEYS[8] -local epoch_key = KEYS[9] -local operation_key = KEYS[10] +local state_inventory_key = KEYS[7] +local owners_key = KEYS[8] +local owner_expiry_key = KEYS[9] +local epoch_key = KEYS[10] +local operation_key = KEYS[11] local operation = ARGV[1] local operation_ttl_ms = tonumber(ARGV[2]) @@ -51,6 +52,26 @@ local function decrement_inventory(upstream) end end +local function state_field(upstream, state) + return string.len(upstream) .. ':' .. upstream .. ':' .. state +end + +local function is_counted(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' or state == 'UNHEALTHY' or state == 'EXTRACTED' +end + +local function decrement_state(upstream, state) + if type(upstream) ~= 'string' or upstream == '' or not is_counted(state) then + return + end + local field = state_field(upstream, state) + local count = redis.call('HINCRBY', state_inventory_key, field, -1) + if count <= 0 then + redis.call('HDEL', state_inventory_key, field) + end +end + local function touch(key, expires_at_ms) if redis.call('EXISTS', key) == 0 then return @@ -100,6 +121,9 @@ local function remove_proxy(id) if decoded and type(record) == 'table' and is_managed(record.state) then decrement_inventory(record.sourceUpstream) end + if decoded and type(record) == 'table' then + decrement_state(record.sourceUpstream, record.state) + end else redis.call('ZREM', available_key, id) end diff --git a/internal/adapters/redisactivity/scripts/status.lua b/internal/adapters/redisactivity/scripts/status.lua new file mode 100644 index 0000000..89c62d0 --- /dev/null +++ b/internal/adapters/redisactivity/scripts/status.lua @@ -0,0 +1,117 @@ +local records_key = KEYS[1] +local unique_key = KEYS[2] +local idkeys_key = KEYS[3] +local expiry_key = KEYS[4] +local available_key = KEYS[5] +local inventory_key = KEYS[6] +local state_inventory_key = KEYS[7] +local owners_key = KEYS[8] +local owner_expiry_key = KEYS[9] + +local now_ms = tonumber(ARGV[1]) +local cleanup_limit = tonumber(ARGV[2]) +local upstream_ids = cjson.decode(ARGV[3]) + +local function is_managed(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' +end + +local function state_field(upstream, state) + return string.len(upstream) .. ':' .. upstream .. ':' .. state +end + +local function is_counted(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' or state == 'UNHEALTHY' or state == 'EXTRACTED' +end + +local function decrement_inventory(upstream) + if type(upstream) ~= 'string' or upstream == '' then + return + end + local count = redis.call('HINCRBY', inventory_key, upstream, -1) + if count < 0 then + redis.call('HSET', inventory_key, upstream, 0) + end +end + +local function decrement_state(upstream, state) + if type(upstream) ~= 'string' or upstream == '' or not is_counted(state) then + return + end + local field = state_field(upstream, state) + local count = redis.call('HINCRBY', state_inventory_key, field, -1) + if count <= 0 then + redis.call('HDEL', state_inventory_key, field) + end +end + +local function remove_available(proxy_id, record) + redis.call('ZREM', available_key, proxy_id) + local index_keys = record and record.indexKeys + if type(index_keys) == 'table' then + for _, index_key in ipairs(index_keys) do + if type(index_key) == 'string' and index_key ~= '' then + redis.call('ZREM', index_key, proxy_id) + end + end + end +end + +local function remove_proxy(proxy_id) + local raw = redis.call('HGET', records_key, proxy_id) + local record = nil + if raw then + local decoded + decoded, record = pcall(cjson.decode, raw) + remove_available(proxy_id, decoded and record or nil) + if decoded and type(record) == 'table' then + if is_managed(record.state) then + decrement_inventory(record.sourceUpstream) + end + decrement_state(record.sourceUpstream, record.state) + end + else + redis.call('ZREM', available_key, proxy_id) + end + local digest = redis.call('HGET', idkeys_key, proxy_id) + if digest and redis.call('HGET', unique_key, digest) == proxy_id then + redis.call('HDEL', unique_key, digest) + end + redis.call('HDEL', idkeys_key, proxy_id) + redis.call('HDEL', records_key, proxy_id) + redis.call('ZREM', expiry_key, proxy_id) + redis.call('HDEL', owners_key, proxy_id) + redis.call('ZREM', owner_expiry_key, proxy_id) +end + +local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now_ms, 'LIMIT', 0, cleanup_limit) +for _, proxy_id in ipairs(expired) do + remove_proxy(proxy_id) +end + +local oldest = redis.call('ZRANGE', expiry_key, 0, 0, 'WITHSCORES') +if #oldest == 2 and tonumber(oldest[2]) <= now_ms then + return cjson.encode({status = 'unavailable', inventories = cjson.decode('[]')}) +end + +local inventories = cjson.decode('[]') +local states = {'FETCHED', 'CHECKING', 'AVAILABLE', 'SUSPECT', 'DRAINING', 'UNHEALTHY', 'EXTRACTED'} +for _, upstream_id in ipairs(upstream_ids) do + if type(upstream_id) ~= 'string' or upstream_id == '' then + return cjson.encode({status = 'invalid', inventories = cjson.decode('[]')}) + end + local counts = {} + for _, state in ipairs(states) do + local count = tonumber(redis.call('HGET', state_inventory_key, state_field(upstream_id, state)) or '0') + if count < 0 then + return cjson.encode({status = 'unavailable', inventories = cjson.decode('[]')}) + end + counts[string.lower(state)] = count + end + counts.upstreamId = upstream_id + inventories[#inventories + 1] = counts +end + +return cjson.encode({status = 'ok', inventories = inventories}) diff --git a/internal/adapters/redisactivity/scripts/sweep.lua b/internal/adapters/redisactivity/scripts/sweep.lua index 342f8e5..280699c 100644 --- a/internal/adapters/redisactivity/scripts/sweep.lua +++ b/internal/adapters/redisactivity/scripts/sweep.lua @@ -4,9 +4,10 @@ local idkeys_key = KEYS[3] local expiry_key = KEYS[4] local available_key = KEYS[5] local inventory_key = KEYS[6] -local owners_key = KEYS[7] -local owner_expiry_key = KEYS[8] -local operation_key = KEYS[9] +local state_inventory_key = KEYS[7] +local owners_key = KEYS[8] +local owner_expiry_key = KEYS[9] +local operation_key = KEYS[10] local operation = ARGV[1] local now_ms = tonumber(ARGV[2]) @@ -40,6 +41,26 @@ local function decrement_inventory(upstream) end end +local function state_field(upstream, state) + return string.len(upstream) .. ':' .. upstream .. ':' .. state +end + +local function is_counted(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' or state == 'UNHEALTHY' or state == 'EXTRACTED' +end + +local function decrement_state(upstream, state) + if type(upstream) ~= 'string' or upstream == '' or not is_counted(state) then + return + end + local field = state_field(upstream, state) + local count = redis.call('HINCRBY', state_inventory_key, field, -1) + if count <= 0 then + redis.call('HDEL', state_inventory_key, field) + end +end + local function remove_available(proxy_id, record) redis.call('ZREM', available_key, proxy_id) local index_keys = record and record.indexKeys @@ -62,6 +83,9 @@ local function remove_proxy(proxy_id) if decoded and type(record) == 'table' and is_managed(record.state) then decrement_inventory(record.sourceUpstream) end + if decoded and type(record) == 'table' then + decrement_state(record.sourceUpstream, record.state) + end else redis.call('ZREM', available_key, proxy_id) end diff --git a/internal/adapters/redisactivity/scripts/upsert.lua b/internal/adapters/redisactivity/scripts/upsert.lua index 35cde0d..15b9bb8 100644 --- a/internal/adapters/redisactivity/scripts/upsert.lua +++ b/internal/adapters/redisactivity/scripts/upsert.lua @@ -4,9 +4,10 @@ local idkeys_key = KEYS[3] local expiry_key = KEYS[4] local available_key = KEYS[5] local inventory_key = KEYS[6] -local owners_key = KEYS[7] -local owner_expiry_key = KEYS[8] -local operation_key = KEYS[9] +local state_inventory_key = KEYS[7] +local owners_key = KEYS[8] +local owner_expiry_key = KEYS[9] +local operation_key = KEYS[10] local now_ms = tonumber(ARGV[1]) local cleanup_limit = tonumber(ARGV[2]) @@ -34,6 +35,33 @@ local function decrement_inventory(upstream) end end +local function state_field(upstream, state) + return string.len(upstream) .. ':' .. upstream .. ':' .. state +end + +local function is_counted(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' or state == 'UNHEALTHY' or state == 'EXTRACTED' +end + +local function increment_state(upstream, state) + if not is_counted(state) then + return + end + redis.call('HINCRBY', state_inventory_key, state_field(upstream, state), 1) +end + +local function decrement_state(upstream, state) + if type(upstream) ~= 'string' or upstream == '' or not is_counted(state) then + return + end + 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 remove_available(proxy_id, record) redis.call('ZREM', available_key, proxy_id) local indexes = record and record.indexKeys or {} @@ -51,6 +79,7 @@ local function remove_proxy(proxy_id) if is_managed(record.state) then decrement_inventory(record.sourceUpstream) end + decrement_state(record.sourceUpstream, record.state) else redis.call('ZREM', available_key, proxy_id) end @@ -166,6 +195,7 @@ for _, candidate in ipairs(candidates) do if is_managed(incoming.state) then redis.call('HINCRBY', inventory_key, candidate.upstream, 1) end + increment_state(candidate.upstream, incoming.state) add_available(candidate.proxyId, incoming) if tonumber(incoming.expiresAtMs) > max_expiry_ms then max_expiry_ms = tonumber(incoming.expiresAtMs) @@ -182,6 +212,7 @@ if max_expiry_ms > 0 then touch(expiry_key, max_expiry_ms) touch(available_key, max_expiry_ms) touch(inventory_key, max_expiry_ms) + touch(state_inventory_key, max_expiry_ms) touch(owners_key, max_expiry_ms) touch(owner_expiry_key, max_expiry_ms) end diff --git a/internal/adapters/redisactivity/status.go b/internal/adapters/redisactivity/status.go new file mode 100644 index 0000000..ecce12a --- /dev/null +++ b/internal/adapters/redisactivity/status.go @@ -0,0 +1,72 @@ +package redisactivity + +import ( + "context" + "encoding/json" + "time" + + "proxy-pool/internal/domain/activitypool" +) + +var _ activitypool.StateInventoryReader = (*Adapter)(nil) + +func (a *Adapter) ReadStateInventory( + ctx context.Context, + upstreamIDs []string, + now time.Time, +) ([]activitypool.StateInventory, error) { + if ctx == nil { + return nil, activitypool.ErrInvalidInventory + } + if err := ctx.Err(); err != nil { + return nil, err + } + if a == nil || now.IsZero() { + return nil, activitypool.ErrInvalidInventory + } + for _, upstreamID := range upstreamIDs { + if upstreamID == "" { + return nil, activitypool.ErrInvalidInventory + } + } + if len(upstreamIDs) == 0 { + return []activitypool.StateInventory{}, nil + } + payload, err := json.Marshal(upstreamIDs) + if err != nil { + return nil, err + } + 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, + }, now.UnixMilli(), a.options.CleanupLimit, string(payload)) + if err != nil { + return nil, err + } + var reply statusScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return nil, err + } + if reply.Status == scriptInvalid { + return nil, activitypool.ErrInvalidInventory + } + if reply.Status == scriptUnavailable { + return nil, invalidScriptReply("expired cleanup is backlogged") + } + if reply.Status != scriptOK || len(reply.Inventories) != len(upstreamIDs) { + return nil, invalidScriptReply("unexpected state inventory reply") + } + inventories := make([]activitypool.StateInventory, len(reply.Inventories)) + for index, item := range reply.Inventories { + if item.UpstreamID != upstreamIDs[index] || item.Fetched < 0 || item.Checking < 0 || + item.Available < 0 || item.Suspect < 0 || item.Draining < 0 || item.Unhealthy < 0 || item.Extracted < 0 { + return nil, invalidScriptReply("invalid state inventory counters") + } + inventories[index] = activitypool.StateInventory{ + UpstreamID: item.UpstreamID, Fetched: item.Fetched, Checking: item.Checking, + Available: item.Available, Suspect: item.Suspect, Draining: item.Draining, + Unhealthy: item.Unhealthy, Extracted: item.Extracted, + } + } + return inventories, nil +} diff --git a/internal/adapters/redisactivity/status_integration_test.go b/internal/adapters/redisactivity/status_integration_test.go new file mode 100644 index 0000000..f4cf057 --- /dev/null +++ b/internal/adapters/redisactivity/status_integration_test.go @@ -0,0 +1,61 @@ +//go:build integration + +package redisactivity + +import ( + "context" + "errors" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + extractionDomain "proxy-pool/internal/domain/extraction" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestReadStateInventoryFailsClosedWhileExpiredCleanupIsBacklogged(t *testing.T) { + fixture := newRedisTestFixture(t) + bounded, err := New(fixture.Client, Options{ + Namespace: fixture.Namespace, Credentials: fixture.Credentials, + OperationTTL: time.Minute, MaxCandidateScan: 32, CleanupLimit: 1, + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + now := time.Date(2026, 7, 30, 12, 0, 0, 0, time.UTC) + _, err = bounded.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Second, MaxSize: 10, + Proxies: []proxyDomain.Proxy{ + {ID: "expired-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, State: proxyDomain.StateFetched}, + {ID: "expired-b", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.11", Port: 8080, State: proxyDomain.StateFetched}, + }, + }) + if err != nil { + t.Fatalf("UpsertFetched() error = %v", err) + } + + if _, err = bounded.ReadStateInventory(context.Background(), []string{"provider-a"}, now.Add(2*time.Second)); !errors.Is(err, extractionDomain.ErrStoreUnavailable) { + t.Fatalf("ReadStateInventory(backlog) error = %v", err) + } + inventories, err := bounded.ReadStateInventory(context.Background(), []string{"provider-a"}, now.Add(2*time.Second)) + if err != nil || len(inventories) != 1 || inventories[0] != (activitypool.StateInventory{UpstreamID: "provider-a"}) { + t.Fatalf("ReadStateInventory(after cleanup) = %+v, %v", inventories, err) + } +} + +func TestReadStateInventoryFailsClosedOnNegativeCounters(t *testing.T) { + fixture := newRedisTestFixture(t) + if err := fixture.Client.HSet( + context.Background(), + fixture.Adapter.keys.stateInventory, + stateInventoryField("provider-a", string(proxyDomain.StateAvailable)), + -1, + ).Err(); err != nil { + t.Fatalf("seed invalid state counter: %v", err) + } + if _, err := fixture.Adapter.ReadStateInventory( + context.Background(), []string{"provider-a"}, time.Now().UTC(), + ); !errors.Is(err, extractionDomain.ErrStoreUnavailable) { + t.Fatalf("ReadStateInventory(negative counter) error = %v", err) + } +} diff --git a/internal/adapters/redisactivity/upsert.go b/internal/adapters/redisactivity/upsert.go index 72b6369..897f940 100644 --- a/internal/adapters/redisactivity/upsert.go +++ b/internal/adapters/redisactivity/upsert.go @@ -175,7 +175,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.owners, a.keys.ownerExpiry, a.keys.operation(operationID), + a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, + a.keys.operation(operationID), }, observedAt.UnixMilli(), a.options.CleanupLimit, maxSize, operationTTLMillis(a.options.OperationTTL), string(payload)) if err != nil { return upsertScriptReply{}, err diff --git a/internal/domain/activitypool/contracttest/contract.go b/internal/domain/activitypool/contracttest/contract.go index dd2b2ef..2951c72 100644 --- a/internal/domain/activitypool/contracttest/contract.go +++ b/internal/domain/activitypool/contracttest/contract.go @@ -18,6 +18,7 @@ type Store interface { activitypool.Upserter activitypool.HealthStore activitypool.InventoryReader + activitypool.StateInventoryReader activitypool.Maintainer extractionDomain.Store ownershipDomain.Repository @@ -48,6 +49,9 @@ func Run(t *testing.T, factory Factory) { t.Run("inventory and bounded maintenance", func(t *testing.T) { runMaintenanceContract(t, newStore(t, factory)) }) + t.Run("state inventory lifecycle", func(t *testing.T) { + runStateInventoryContract(t, newStore(t, factory)) + }) t.Run("concurrent exclusivity", func(t *testing.T) { runConcurrencyContract(t, factory) }) @@ -338,6 +342,88 @@ func runMaintenanceContract(t *testing.T, store Store) { assertInventory(t, store, "provider-a", now.Add(6*time.Second), 0) } +func runStateInventoryContract(t *testing.T, store Store) { + t.Helper() + now := contractNow() + upstreamA := "provider:a:FETCHED" + upstreamB := "provider:a" + states := []proxyDomain.State{ + proxyDomain.StateFetched, + proxyDomain.StateChecking, + proxyDomain.StateAvailable, + proxyDomain.StateSuspect, + proxyDomain.StateDraining, + proxyDomain.StateUnhealthy, + proxyDomain.StateExtracted, + } + for index, state := range states { + upsertOne(t, store, upstreamA, now, time.Minute, + contractProxy(fmt.Sprintf("state-%d", index), fmt.Sprintf("192.0.2.%d", index+30), state)) + } + upsertOne(t, store, upstreamB, now, time.Minute, + contractProxy("collision-control", "198.51.100.30", proxyDomain.StateFetched)) + + inventories, err := store.ReadStateInventory(context.Background(), []string{upstreamB, upstreamA}, now) + if err != nil || len(inventories) != 2 { + t.Fatalf("ReadStateInventory() = %+v, %v", inventories, err) + } + if inventories[0] != (activitypool.StateInventory{UpstreamID: upstreamB, Fetched: 1}) { + t.Fatalf("ReadStateInventory(collision control) = %+v", inventories[0]) + } + wantAll := activitypool.StateInventory{ + UpstreamID: upstreamA, Fetched: 1, Checking: 1, Available: 1, Suspect: 1, + Draining: 1, Unhealthy: 1, Extracted: 1, + } + if inventories[1] != wantAll { + t.Fatalf("ReadStateInventory(all states) = %+v, want %+v", inventories[1], wantAll) + } + + transition := activitypool.HealthUpdate{ + ProxyID: "state-0", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking, + } + if _, err := store.ApplyHealth(context.Background(), transition); err != nil { + t.Fatalf("ApplyHealth(state inventory transition): %v", err) + } + if _, err := store.ApplyHealth(context.Background(), transition); err != nil { + t.Fatalf("ApplyHealth(state inventory replay): %v", err) + } + afterTransition, err := store.ReadStateInventory(context.Background(), []string{upstreamA}, now.Add(time.Second)) + if err != nil || len(afterTransition) != 1 || afterTransition[0].Fetched != 0 || afterTransition[0].Checking != 2 { + t.Fatalf("ReadStateInventory(after transition) = %+v, %v", afterTransition, err) + } + + extractCommand := extractionDomain.Command{ + RequestID: "state-inventory-extract", ClientID: "client-a", Requested: 1, + IdempotencyKey: "state-inventory-extract", IdempotencyTTL: time.Minute, + Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), + Upstreams: []string{upstreamA}, + } + result, err := store.Extract(context.Background(), extractCommand) + if err != nil || result.Returned != 1 { + t.Fatalf("Extract(state inventory) = %+v, %v", result, err) + } + extractCommand.RequestID = "state-inventory-extract-replay" + if replayed, err := store.Extract(context.Background(), extractCommand); err != nil || replayed.Returned != 1 || + replayed.Items[0].ID != result.Items[0].ID { + t.Fatalf("Extract(state inventory replay) = %+v, %v", replayed, err) + } + afterExtract, err := store.ReadStateInventory(context.Background(), []string{upstreamA}, now.Add(2*time.Second)) + if err != nil || len(afterExtract) != 1 || afterExtract[0].Available != 0 || afterExtract[0].Extracted != 2 { + t.Fatalf("ReadStateInventory(after extract) = %+v, %v", afterExtract, err) + } + + afterExpiry, err := store.ReadStateInventory(context.Background(), []string{upstreamA, upstreamB}, now.Add(2*time.Minute)) + if err != nil || len(afterExpiry) != 2 || afterExpiry[0] != (activitypool.StateInventory{UpstreamID: upstreamA}) || + afterExpiry[1] != (activitypool.StateInventory{UpstreamID: upstreamB}) { + t.Fatalf("ReadStateInventory(after expiry) = %+v, %v", afterExpiry, err) + } + + empty, err := store.ReadStateInventory(context.Background(), nil, now) + if err != nil || len(empty) != 0 { + t.Fatalf("ReadStateInventory(empty) = %+v, %v", empty, err) + } +} + func runConcurrencyContract(t *testing.T, factory Factory) { t.Helper() now := contractNow() @@ -439,6 +525,10 @@ func runCancellationContract(t *testing.T, store Store) { return err }}, {name: "inventory", call: func() error { _, err := store.Inventory(ctx, "provider-a", now); return err }}, + {name: "state inventory", call: func() error { + _, err := store.ReadStateInventory(ctx, []string{"provider-a"}, now) + return err + }}, {name: "sweep", call: func() error { _, err := store.SweepExpired(ctx, now, 1); return err }}, {name: "extract", call: func() error { _, err := store.Extract(ctx, extractionDomain.Command{Requested: 1, Fulfillment: extractionDomain.Partial, Now: now}) diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index c8cd5ec..878b92d 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -60,6 +60,19 @@ type Inventory struct { Managed int } +// StateInventory is a low-cardinality operational view of one upstream. +// Expired and removed entries are intentionally excluded. +type StateInventory struct { + UpstreamID string + Fetched int64 + Checking int64 + Available int64 + Suspect int64 + Draining int64 + Unhealthy int64 + Extracted int64 +} + type HealthStore interface { ApplyHealth(context.Context, HealthUpdate) (Entry, error) } @@ -68,6 +81,10 @@ type InventoryReader interface { Inventory(context.Context, string, time.Time) (Inventory, error) } +type StateInventoryReader interface { + ReadStateInventory(context.Context, []string, time.Time) ([]StateInventory, error) +} + type Maintainer interface { SweepExpired(context.Context, time.Time, int) (int, error) } @@ -99,6 +116,7 @@ var ( _ Upserter = (*MemoryPool)(nil) _ HealthStore = (*MemoryPool)(nil) _ InventoryReader = (*MemoryPool)(nil) + _ StateInventoryReader = (*MemoryPool)(nil) _ Maintainer = (*MemoryPool)(nil) _ extractionDomain.Store = (*MemoryPool)(nil) _ ownershipDomain.Repository = (*MemoryPool)(nil) @@ -302,6 +320,50 @@ func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time. return result, nil } +func (p *MemoryPool) ReadStateInventory( + ctx context.Context, + upstreamIDs []string, + now time.Time, +) ([]StateInventory, error) { + if ctx == nil { + return nil, ErrInvalidInventory + } + if err := ctx.Err(); err != nil { + return nil, err + } + if p == nil || now.IsZero() { + return nil, ErrInvalidInventory + } + result := make([]StateInventory, len(upstreamIDs)) + positions := make(map[string][]int, len(upstreamIDs)) + for index, upstreamID := range upstreamIDs { + if upstreamID == "" { + return nil, ErrInvalidInventory + } + result[index].UpstreamID = upstreamID + positions[upstreamID] = append(positions[upstreamID], index) + } + if len(result) == 0 { + return result, nil + } + + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, err + } + for _, entry := range p.entries { + indexes := positions[entry.Proxy.SourceUpstream] + if len(indexes) == 0 || entry.Proxy.ExpiresAt == nil || !entry.Proxy.ExpiresAt.After(now) { + continue + } + for _, index := range indexes { + incrementStateInventory(&result[index], entry.State) + } + } + return result, nil +} + func (p *MemoryPool) SweepExpired(ctx context.Context, now time.Time, limit int) (int, error) { if ctx == nil { return 0, ErrInvalidMaintenance @@ -744,6 +806,28 @@ func managedActivityState(state proxyDomain.State) bool { } } +func incrementStateInventory(inventory *StateInventory, state proxyDomain.State) { + if inventory == nil { + return + } + switch state { + case proxyDomain.StateFetched: + inventory.Fetched++ + case proxyDomain.StateChecking: + inventory.Checking++ + case proxyDomain.StateAvailable: + inventory.Available++ + case proxyDomain.StateSuspect: + inventory.Suspect++ + case proxyDomain.StateDraining: + inventory.Draining++ + case proxyDomain.StateUnhealthy: + inventory.Unhealthy++ + case proxyDomain.StateExtracted: + inventory.Extracted++ + } +} + func cloneProxy(candidate proxyDomain.Proxy) proxyDomain.Proxy { if candidate.ExpiresAt != nil { value := *candidate.ExpiresAt