youfak/proxy-pool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add redis activity state inventory

Browse Source
This commit is contained in:
youfak 2026-07-30 11:38:07 +08:00
parent 96701e4241
commit 6a3660d639
18 changed files with 680 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
internal/adapters/redisactivity/adapter_test.go
View File

@ -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{

3
internal/adapters/redisactivity/extract.go
View File

@ -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

3
internal/adapters/redisactivity/health.go
View File

@ -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 {

6
internal/adapters/redisactivity/keys.go
View File

@ -19,6 +19,7 @@ type keyspace struct {
ownerExpiry string
epoch string
inventory string
stateInventory string
}
func newKeyspace(namespace string) keyspace {
@ -34,9 +35,14 @@ func newKeyspace(namespace string) keyspace {
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)
}

3
internal/adapters/redisactivity/maintenance.go
View File

@ -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

3
internal/adapters/redisactivity/ownership.go
View File

@ -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 {

20
internal/adapters/redisactivity/scripts.go
View File

@ -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) {

43
internal/adapters/redisactivity/scripts/extract.lua
View File

@ -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)

43
internal/adapters/redisactivity/scripts/health.lua
View File

@ -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))

32
internal/adapters/redisactivity/scripts/ownership.lua
View File

@ -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

117
internal/adapters/redisactivity/scripts/status.lua Normal file
View File

@ -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})

30
internal/adapters/redisactivity/scripts/sweep.lua
View File

@ -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

37
internal/adapters/redisactivity/scripts/upsert.lua
View File

@ -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

72
internal/adapters/redisactivity/status.go Normal file
View File

@ -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
}

61
internal/adapters/redisactivity/status_integration_test.go Normal file
View File

@ -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)
}
}

3
internal/adapters/redisactivity/upsert.go
View File

@ -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

90
internal/domain/activitypool/contracttest/contract.go
View File

@ -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})

84
internal/domain/activitypool/pool.go
View File

@ -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