diff --git a/internal/adapters/redisactivity/codec.go b/internal/adapters/redisactivity/codec.go index 2c199d7..4f226ff 100644 --- a/internal/adapters/redisactivity/codec.go +++ b/internal/adapters/redisactivity/codec.go @@ -18,24 +18,26 @@ var ErrInvalidRecord = errors.New("invalid redis activity record") const recordVersion = 1 type proxyRecord struct { - Version int `json:"version"` - ID string `json:"id"` - Scheme string `json:"scheme"` - Host string `json:"host"` - Port int64 `json:"port"` - Username string `json:"username,omitempty"` - Password string `json:"password,omitempty"` - SourceUpstream string `json:"sourceUpstream"` - CreatedAtMS int64 `json:"createdAtMs"` - ExpiresAtMS int64 `json:"expiresAtMs"` - UsableUntilMS int64 `json:"usableUntilMs"` - LastCheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"` - LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"` - LatencyNS int64 `json:"latencyNs"` - MaxConcurrency int64 `json:"maxConcurrency"` - State string `json:"state"` - Tags map[string]string `json:"tags,omitempty"` - OwnerWorkerID string `json:"ownerWorkerId,omitempty"` + Version int `json:"version"` + ID string `json:"id"` + Scheme string `json:"scheme"` + Host string `json:"host"` + Port int64 `json:"port"` + Username string `json:"username,omitempty"` + Password string `json:"password,omitempty"` + CredentialVersion string `json:"credentialVersion,omitempty"` + SourceUpstream string `json:"sourceUpstream"` + CreatedAtMS int64 `json:"createdAtMs"` + ExpiresAtMS int64 `json:"expiresAtMs"` + UsableUntilMS int64 `json:"usableUntilMs"` + LastCheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"` + LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"` + LatencyNS int64 `json:"latencyNs"` + MaxConcurrency int64 `json:"maxConcurrency"` + State string `json:"state"` + Tags map[string]string `json:"tags,omitempty"` + OwnerWorkerID string `json:"ownerWorkerId,omitempty"` + IndexKeys []string `json:"indexKeys,omitempty"` } type ownershipRecord struct { @@ -204,6 +206,11 @@ func validateProxyRecord(record proxyRecord) error { !validScheme(record.Scheme) || !validProxyState(record.State) { return ErrInvalidRecord } + for _, key := range record.IndexKeys { + if key == "" || !strings.Contains(key, "{activity}") { + return ErrInvalidRecord + } + } return nil } diff --git a/internal/adapters/redisactivity/health.go b/internal/adapters/redisactivity/health.go new file mode 100644 index 0000000..b9012ff --- /dev/null +++ b/internal/adapters/redisactivity/health.go @@ -0,0 +1,57 @@ +package redisactivity + +import ( + "context" + + "proxy-pool/internal/domain/activitypool" +) + +var _ activitypool.HealthStore = (*Adapter)(nil) + +func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpdate) (activitypool.Entry, error) { + if ctx == nil { + return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate + } + if err := ctx.Err(); err != nil { + return activitypool.Entry{}, err + } + if a == nil || update.ProxyID == "" || update.CheckedAt.IsZero() || update.Latency < 0 || + !validProxyState(string(update.NextState)) { + return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate + } + operationID, err := newOperationID() + if err != nil { + return activitypool.Entry{}, err + } + 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), + }, update.CheckedAt.UnixMilli(), string(update.NextState), int64(update.Latency), + a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), update.ProxyID) + if err != nil { + return activitypool.Entry{}, err + } + var reply healthScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return activitypool.Entry{}, err + } + switch reply.Status { + case scriptNotFound: + return activitypool.Entry{}, activitypool.ErrActivityNotFound + case scriptStale: + return activitypool.Entry{}, activitypool.ErrStaleHealthUpdate + case scriptInvalid: + return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate + case scriptOK: + if reply.Record == "" { + return activitypool.Entry{}, invalidScriptReply("health reply omitted record") + } + record, err := decodeProxyRecord(reply.Record) + if err != nil { + return activitypool.Entry{}, invalidScriptReply("health reply contained an invalid record") + } + return proxyRecordEntry(record), nil + default: + return activitypool.Entry{}, invalidScriptReply("unexpected health status") + } +} diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index d3c9d2c..5042017 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -2,8 +2,12 @@ package redisactivity import ( "context" + "crypto/rand" + _ "embed" + "encoding/hex" "errors" "fmt" + "time" "github.com/redis/go-redis/v9" @@ -17,6 +21,7 @@ const ( scriptInvalid scriptStatus = "invalid" scriptNotFound scriptStatus = "not_found" scriptConflict scriptStatus = "conflict" + scriptStale scriptStatus = "stale" scriptUnavailable scriptStatus = "unavailable" scriptInsufficient scriptStatus = "insufficient" ) @@ -49,6 +54,17 @@ type maintenanceScriptReply struct { Count int `json:"count"` } +//go:embed scripts/upsert.lua +var upsertSource string + +//go:embed scripts/health.lua +var healthSource string + +var ( + upsertScript = redis.NewScript(upsertSource) + healthScript = redis.NewScript(healthSource) +) + func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) { result, err := script.Run(ctx, client, keys, args...).Result() if err != nil { @@ -62,3 +78,35 @@ func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, } return result, nil } + +func newOperationID() (string, error) { + var value [16]byte + if _, err := rand.Read(value[:]); err != nil { + return "", fmt.Errorf("create redis activity operation ID: %w", err) + } + return hex.EncodeToString(value[:]), nil +} + +func operationTTLMillis(ttl time.Duration) int64 { + milliseconds := ttl / time.Millisecond + if ttl%time.Millisecond != 0 { + milliseconds++ + } + return int64(milliseconds) +} + +func decodeScriptResult(result any, destination any) error { + var payload string + switch value := result.(type) { + case string: + payload = value + case []byte: + payload = string(value) + default: + return errors.Join(extractionDomain.ErrStoreUnavailable, errors.New("invalid Redis script reply type")) + } + if err := decodeJSON(payload, destination); err != nil { + return errors.Join(extractionDomain.ErrStoreUnavailable, fmt.Errorf("decode Redis script reply: %w", err)) + } + return nil +} diff --git a/internal/adapters/redisactivity/scripts/health.lua b/internal/adapters/redisactivity/scripts/health.lua new file mode 100644 index 0000000..0d95661 --- /dev/null +++ b/internal/adapters/redisactivity/scripts/health.lua @@ -0,0 +1,169 @@ +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 owners_key = KEYS[7] +local owner_expiry_key = KEYS[8] +local operation_key = KEYS[9] + +local checked_at_ms = tonumber(ARGV[1]) +local next_state = ARGV[2] +local latency_ns = tonumber(ARGV[3]) +local cleanup_limit = tonumber(ARGV[4]) +local operation_ttl_ms = tonumber(ARGV[5]) +local proxy_id = ARGV[6] + +local committed = redis.call('GET', operation_key) +if committed then + return committed +end + +local function is_managed(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' +end + +local function decrement_inventory(upstream) + if not upstream or upstream == '' then + return + end + local value = redis.call('HINCRBY', inventory_key, upstream, -1) + if value < 0 then + redis.call('HSET', inventory_key, upstream, 0) + 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 + redis.call('ZREM', index_key, id) + end +end + +local function remove_proxy(id) + local raw = redis.call('HGET', records_key, id) + local record = nil + if raw then + record = cjson.decode(raw) + remove_available(id, record) + if is_managed(record.state) then + decrement_inventory(record.sourceUpstream) + end + else + redis.call('ZREM', available_key, id) + end + local digest = redis.call('HGET', idkeys_key, id) + if digest and redis.call('HGET', unique_key, digest) == id then + redis.call('HDEL', unique_key, digest) + end + redis.call('HDEL', idkeys_key, id) + redis.call('HDEL', records_key, id) + redis.call('ZREM', expiry_key, id) + redis.call('HDEL', owners_key, id) + redis.call('ZREM', owner_expiry_key, id) +end + +local function cleanup_expired() + local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', checked_at_ms, 'LIMIT', 0, cleanup_limit) + for _, id in ipairs(expired) do + remove_proxy(id) + end +end + +local function touch(key, expires_at_ms) + if redis.call('EXISTS', key) == 0 then + return + end + local current = redis.call('PEXPIRETIME', key) + if current < expires_at_ms then + redis.call('PEXPIREAT', key, expires_at_ms) + end +end + +local function finish(reply, hard_expiry_ms) + local encoded = cjson.encode(reply) + redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms) + if hard_expiry_ms then + local operation_expiry_ms = redis.call('PEXPIRETIME', operation_key) + if operation_expiry_ms > hard_expiry_ms then + redis.call('PEXPIREAT', operation_key, hard_expiry_ms) + end + end + return encoded +end + +local transitions = { + FETCHED = {CHECKING = true, EXPIRED = true, REMOVED = true}, + CHECKING = {AVAILABLE = true, UNHEALTHY = true, EXPIRED = true, REMOVED = true}, + AVAILABLE = {SUSPECT = true, DRAINING = true, EXTRACTED = true, EXPIRED = true}, + SUSPECT = {AVAILABLE = true, UNHEALTHY = true, DRAINING = true, EXPIRED = true}, + DRAINING = {EXPIRED = true, UNHEALTHY = true, REMOVED = true}, + UNHEALTHY = {CHECKING = true, REMOVED = true, EXPIRED = true}, + EXTRACTED = {EXPIRED = true, REMOVED = true}, + EXPIRED = {REMOVED = true}, + REMOVED = {}, +} + +cleanup_expired() + +local raw = redis.call('HGET', records_key, proxy_id) +if not raw then + return finish({status = 'not_found'}) +end +local record = cjson.decode(raw) +if tonumber(record.expiresAtMs) <= checked_at_ms then + remove_proxy(proxy_id) + return finish({status = 'not_found'}) +end + +local last_checked_at_ms = tonumber(record.lastCheckedAtMs or '0') +if checked_at_ms < last_checked_at_ms then + return finish({status = 'stale'}) +end +if checked_at_ms == last_checked_at_ms then + if record.state ~= next_state then + return finish({status = 'stale'}) + end + return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs)) +end +if record.state ~= next_state and not (transitions[record.state] and transitions[record.state][next_state]) then + return finish({status = 'invalid'}) +end + +local was_managed = is_managed(record.state) +local will_be_managed = is_managed(next_state) +remove_available(proxy_id, record) +record.state = next_state +record.lastCheckedAtMs = checked_at_ms +record.latencyNs = latency_ns +if next_state == 'AVAILABLE' then + record.lastSuccessAtMs = checked_at_ms +end +if was_managed and not will_be_managed then + decrement_inventory(record.sourceUpstream) +elseif not was_managed and will_be_managed then + redis.call('HINCRBY', inventory_key, record.sourceUpstream, 1) +end + +local encoded = cjson.encode(record) +redis.call('HSET', records_key, proxy_id, encoded) +local owned = (record.ownerWorkerId and record.ownerWorkerId ~= '') or redis.call('HEXISTS', owners_key, proxy_id) == 1 +if next_state == 'AVAILABLE' and not owned and tonumber(record.usableUntilMs) > checked_at_ms then + redis.call('ZADD', available_key, record.usableUntilMs, proxy_id) + for _, index_key in ipairs(record.indexKeys or {}) do + redis.call('ZADD', index_key, record.usableUntilMs, proxy_id) + touch(index_key, tonumber(record.expiresAtMs)) + end +end +touch(records_key, tonumber(record.expiresAtMs)) +touch(unique_key, tonumber(record.expiresAtMs)) +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(owners_key, tonumber(record.expiresAtMs)) +touch(owner_expiry_key, tonumber(record.expiresAtMs)) + +return finish({status = 'ok', record = encoded}, tonumber(record.expiresAtMs)) diff --git a/internal/adapters/redisactivity/scripts/upsert.lua b/internal/adapters/redisactivity/scripts/upsert.lua new file mode 100644 index 0000000..35cde0d --- /dev/null +++ b/internal/adapters/redisactivity/scripts/upsert.lua @@ -0,0 +1,192 @@ +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 owners_key = KEYS[7] +local owner_expiry_key = KEYS[8] +local operation_key = KEYS[9] + +local now_ms = tonumber(ARGV[1]) +local cleanup_limit = tonumber(ARGV[2]) +local max_size = tonumber(ARGV[3]) +local operation_ttl_ms = tonumber(ARGV[4]) +local candidates = cjson.decode(ARGV[5]) + +local committed = redis.call('GET', operation_key) +if committed then + return committed +end + +local function is_managed(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' +end + +local function decrement_inventory(upstream) + if not upstream or upstream == '' then + return + end + local value = redis.call('HINCRBY', inventory_key, upstream, -1) + if value < 0 then + redis.call('HSET', inventory_key, upstream, 0) + end +end + +local function remove_available(proxy_id, record) + redis.call('ZREM', available_key, proxy_id) + local indexes = record and record.indexKeys or {} + for _, index_key in ipairs(indexes) do + redis.call('ZREM', index_key, proxy_id) + end +end + +local function remove_proxy(proxy_id) + local raw = redis.call('HGET', records_key, proxy_id) + local record = nil + if raw then + record = cjson.decode(raw) + remove_available(proxy_id, record) + if is_managed(record.state) then + decrement_inventory(record.sourceUpstream) + 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 function cleanup_expired() + 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 +end + +local function touch(key, expires_at_ms) + if redis.call('EXISTS', key) == 0 then + return + end + local current = redis.call('PEXPIRETIME', key) + if current < expires_at_ms then + redis.call('PEXPIREAT', key, expires_at_ms) + end +end + +local function add_available(proxy_id, record) + if record.state ~= 'AVAILABLE' or (record.ownerWorkerId and record.ownerWorkerId ~= '') or + tonumber(record.usableUntilMs) <= now_ms then + return + end + redis.call('ZADD', available_key, record.usableUntilMs, proxy_id) + for _, index_key in ipairs(record.indexKeys or {}) do + redis.call('ZADD', index_key, record.usableUntilMs, proxy_id) + touch(index_key, tonumber(record.expiresAtMs)) + end +end + +local function finish(reply) + local encoded = cjson.encode(reply) + redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms) + return encoded +end + +cleanup_expired() + +for _, candidate in ipairs(candidates) do + local mapped = redis.call('HGET', idkeys_key, candidate.proxyId) + if mapped and mapped ~= candidate.uniqueDigest then + return finish({status = 'invalid', accepted = 0, inserted = 0, refreshed = 0, dropped = 0}) + end +end + +local accepted = #candidates +local inserted = 0 +local refreshed = 0 +local dropped = 0 +local max_expiry_ms = 0 + +for _, candidate in ipairs(candidates) do + local incumbent_id = redis.call('HGET', unique_key, candidate.uniqueDigest) + local current_raw = incumbent_id and redis.call('HGET', records_key, incumbent_id) or nil + if incumbent_id and not current_raw then + remove_proxy(incumbent_id) + redis.call('HDEL', unique_key, candidate.uniqueDigest) + incumbent_id = nil + end + + if current_raw then + local current = cjson.decode(current_raw) + if tonumber(current.expiresAtMs) <= now_ms then + remove_proxy(incumbent_id) + incumbent_id = nil + current_raw = nil + elseif current.state == 'EXTRACTED' or current.sourceUpstream ~= candidate.upstream then + refreshed = refreshed + 1 + else + local incoming = cjson.decode(candidate.record) + remove_available(incumbent_id, current) + incoming.id = current.id + incoming.createdAtMs = current.createdAtMs + incoming.state = current.state + incoming.lastCheckedAtMs = current.lastCheckedAtMs + incoming.lastSuccessAtMs = current.lastSuccessAtMs + incoming.latencyNs = current.latencyNs + incoming.ownerWorkerId = current.ownerWorkerId + local encoded = cjson.encode(incoming) + redis.call('HSET', records_key, incumbent_id, encoded) + redis.call('ZADD', expiry_key, incoming.expiresAtMs, incumbent_id) + add_available(incumbent_id, incoming) + if tonumber(incoming.expiresAtMs) > max_expiry_ms then + max_expiry_ms = tonumber(incoming.expiresAtMs) + end + refreshed = refreshed + 1 + end + end + + if not incumbent_id then + local current_size = tonumber(redis.call('HGET', inventory_key, candidate.upstream) or '0') + if current_size >= max_size then + dropped = dropped + 1 + else + local incoming = cjson.decode(candidate.record) + redis.call('HSET', records_key, candidate.proxyId, candidate.record) + redis.call('HSET', unique_key, candidate.uniqueDigest, candidate.proxyId) + redis.call('HSET', idkeys_key, candidate.proxyId, candidate.uniqueDigest) + redis.call('ZADD', expiry_key, incoming.expiresAtMs, candidate.proxyId) + if is_managed(incoming.state) then + redis.call('HINCRBY', inventory_key, candidate.upstream, 1) + end + add_available(candidate.proxyId, incoming) + if tonumber(incoming.expiresAtMs) > max_expiry_ms then + max_expiry_ms = tonumber(incoming.expiresAtMs) + end + inserted = inserted + 1 + end + end +end + +if max_expiry_ms > 0 then + touch(records_key, max_expiry_ms) + touch(unique_key, max_expiry_ms) + touch(idkeys_key, max_expiry_ms) + touch(expiry_key, max_expiry_ms) + touch(available_key, max_expiry_ms) + touch(inventory_key, max_expiry_ms) + touch(owners_key, max_expiry_ms) + touch(owner_expiry_key, max_expiry_ms) +end + +return finish({ + status = 'ok', accepted = accepted, inserted = inserted, + refreshed = refreshed, dropped = dropped, +}) diff --git a/internal/adapters/redisactivity/upsert.go b/internal/adapters/redisactivity/upsert.go new file mode 100644 index 0000000..72b6369 --- /dev/null +++ b/internal/adapters/redisactivity/upsert.go @@ -0,0 +1,258 @@ +package redisactivity + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "time" + + "proxy-pool/internal/domain/activitypool" + extractionDomain "proxy-pool/internal/domain/extraction" + proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/platform/credentials" +) + +const maxUpsertScriptBatch = 256 + +type upsertCandidate struct { + ProxyID string `json:"proxyId"` + UniqueDigest string `json:"uniqueDigest"` + Upstream string `json:"upstream"` + Record string `json:"record"` +} + +var _ activitypool.Upserter = (*Adapter)(nil) + +func (a *Adapter) UpsertFetched(ctx context.Context, upstreamID string, batch activitypool.FetchedBatch) (activitypool.UpsertResult, error) { + var result activitypool.UpsertResult + if ctx == nil { + return result, activitypool.ErrInvalidBatch + } + if err := ctx.Err(); err != nil { + return result, err + } + if a == nil || upstreamID == "" || batch.ObservedAt.IsZero() || batch.ConfiguredTTL < 0 || + batch.AllocationSafetyMargin < 0 || batch.MaxSize <= 0 || + (batch.ConfiguredTTL > 0 && batch.AllocationSafetyMargin >= batch.ConfiguredTTL) { + return result, activitypool.ErrInvalidBatch + } + for _, candidate := range batch.Proxies { + if candidate.SourceUpstream != "" && candidate.SourceUpstream != upstreamID { + return result, activitypool.ErrInvalidBatch + } + } + + prepared := make([]upsertCandidate, 0, len(batch.Proxies)) + seenIDs := make(map[string]string, len(batch.Proxies)) + for _, candidate := range batch.Proxies { + if err := ctx.Err(); err != nil { + return activitypool.UpsertResult{}, err + } + item, accepted, err := a.prepareUpsertCandidate(ctx, upstreamID, batch, candidate) + if err != nil { + return activitypool.UpsertResult{}, err + } + if !accepted { + result.Dropped++ + continue + } + if digest, exists := seenIDs[item.ProxyID]; exists && digest != item.UniqueDigest { + return activitypool.UpsertResult{}, activitypool.ErrInvalidBatch + } + seenIDs[item.ProxyID] = item.UniqueDigest + prepared = append(prepared, item) + } + result.Accepted = len(prepared) + + for start := 0; start < len(prepared); start += maxUpsertScriptBatch { + end := min(start+maxUpsertScriptBatch, len(prepared)) + reply, err := a.upsertChunk(ctx, batch.ObservedAt, batch.MaxSize, prepared[start:end]) + if err != nil { + return activitypool.UpsertResult{}, err + } + if reply.Accepted != end-start || reply.Inserted < 0 || reply.Refreshed < 0 || reply.Dropped < 0 || + reply.Inserted+reply.Refreshed+reply.Dropped != reply.Accepted { + return activitypool.UpsertResult{}, invalidScriptReply("invalid upsert counters") + } + result.Inserted += reply.Inserted + result.Refreshed += reply.Refreshed + result.Dropped += reply.Dropped + } + return result, nil +} + +func (a *Adapter) prepareUpsertCandidate( + ctx context.Context, + upstreamID string, + batch activitypool.FetchedBatch, + candidate proxyDomain.Proxy, +) (upsertCandidate, bool, error) { + if !validCandidateIdentity(candidate) { + return upsertCandidate{}, false, nil + } + expiresAt := proxyDomain.EffectiveExpiry(batch.ObservedAt, candidate.ExpiresAt, 0, batch.ConfiguredTTL) + if expiresAt == nil { + return upsertCandidate{}, false, nil + } + usableUntil := expiresAt.Add(-batch.AllocationSafetyMargin) + if !usableUntil.After(batch.ObservedAt) || usableUntil.UnixMilli() <= batch.ObservedAt.UnixMilli() { + return upsertCandidate{}, false, nil + } + if candidate.MaxConcurrency < 0 || (candidate.State != "" && !validProxyState(string(candidate.State))) { + return upsertCandidate{}, false, activitypool.ErrInvalidBatch + } + if (candidate.SecretRef == "") != (candidate.CredentialVersion == "") { + return upsertCandidate{}, false, activitypool.ErrInvalidBatch + } + + password := "" + if candidate.SecretRef != "" { + value, err := a.credentials.Resolve(ctx, credentials.Reference{ + SecretRef: candidate.SecretRef, CredentialVersion: candidate.CredentialVersion, + }) + if err != nil { + return upsertCandidate{}, false, fmt.Errorf("resolve proxy credential: %w", err) + } + candidate.Username = value.Username + password = value.Password + } + candidate.SourceUpstream = upstreamID + candidate.ExpiresAt = expiresAt + candidate.UsableUntil = &usableUntil + if candidate.CreatedAt.IsZero() { + candidate.CreatedAt = batch.ObservedAt.UTC() + } + if candidate.State == "" { + candidate.State = proxyDomain.StateFetched + } + uniqueKey := candidate.UniqueKey() + if candidate.ID == "" { + candidate.ID = stableAdapterProxyID(uniqueKey) + } + record := proxyRecord{ + Version: recordVersion, ID: candidate.ID, Scheme: string(candidate.Scheme), Host: candidate.Host, + Port: int64(candidate.Port), Username: candidate.Username, Password: password, + CredentialVersion: candidate.CredentialVersion, SourceUpstream: upstreamID, + CreatedAtMS: candidate.CreatedAt.UnixMilli(), ExpiresAtMS: expiresAt.UnixMilli(), + UsableUntilMS: usableUntil.UnixMilli(), LatencyNS: int64(candidate.Latency), + MaxConcurrency: candidate.MaxConcurrency, State: string(candidate.State), + Tags: cloneTags(candidate.Tags), IndexKeys: a.availableIndexKeys(candidate), + } + if candidate.LastCheckedAt != nil { + record.LastCheckedAtMS = candidate.LastCheckedAt.UnixMilli() + } + if candidate.LastSuccessAt != nil { + record.LastSuccessAtMS = candidate.LastSuccessAt.UnixMilli() + } + encoded, err := encodeProxyRecord(record) + if err != nil { + if errors.Is(err, ErrInvalidRecord) { + return upsertCandidate{}, false, activitypool.ErrInvalidBatch + } + return upsertCandidate{}, false, err + } + return upsertCandidate{ + ProxyID: candidate.ID, UniqueDigest: digestToken(uniqueKey), Upstream: upstreamID, Record: encoded, + }, true, nil +} + +func (a *Adapter) upsertChunk( + ctx context.Context, + observedAt time.Time, + maxSize int, + candidates []upsertCandidate, +) (upsertScriptReply, error) { + operationID, err := newOperationID() + if err != nil { + return upsertScriptReply{}, err + } + payload, err := json.Marshal(candidates) + if err != nil { + return upsertScriptReply{}, fmt.Errorf("encode Redis upsert candidates: %w", err) + } + 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), + }, observedAt.UnixMilli(), a.options.CleanupLimit, maxSize, operationTTLMillis(a.options.OperationTTL), string(payload)) + if err != nil { + return upsertScriptReply{}, err + } + var reply upsertScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return upsertScriptReply{}, err + } + switch reply.Status { + case scriptOK: + return reply, nil + case scriptInvalid: + return upsertScriptReply{}, activitypool.ErrInvalidBatch + default: + return upsertScriptReply{}, invalidScriptReply("unexpected upsert status") + } +} + +func (a *Adapter) availableIndexKeys(candidate proxyDomain.Proxy) []string { + keys := []string{a.keys.protocol(string(candidate.Scheme))} + if region := candidate.Tags["region"]; region != "" { + keys = append(keys, a.keys.region(region)) + } + if carrier := candidate.Tags["carrier"]; carrier != "" { + keys = append(keys, a.keys.carrier(carrier)) + } + return append(keys, a.keys.upstream(candidate.SourceUpstream)) +} + +func proxyRecordEntry(record proxyRecord) activitypool.Entry { + createdAt := time.UnixMilli(record.CreatedAtMS).UTC() + expiresAt := time.UnixMilli(record.ExpiresAtMS).UTC() + usableUntil := time.UnixMilli(record.UsableUntilMS).UTC() + proxy := proxyDomain.Proxy{ + ID: record.ID, Scheme: proxyDomain.Scheme(record.Scheme), Host: record.Host, Port: uint16(record.Port), + Username: record.Username, CredentialVersion: record.CredentialVersion, + SourceUpstream: record.SourceUpstream, CreatedAt: createdAt, ExpiresAt: &expiresAt, + UsableUntil: &usableUntil, Latency: time.Duration(record.LatencyNS), + MaxConcurrency: record.MaxConcurrency, State: proxyDomain.State(record.State), Tags: cloneTags(record.Tags), + } + if record.LastCheckedAtMS > 0 { + value := time.UnixMilli(record.LastCheckedAtMS).UTC() + proxy.LastCheckedAt = &value + } + if record.LastSuccessAtMS > 0 { + value := time.UnixMilli(record.LastSuccessAtMS).UTC() + proxy.LastSuccessAt = &value + } + return activitypool.Entry{ + Proxy: proxy, UsableUntil: usableUntil, + OwnerWorkerID: record.OwnerWorkerID, State: proxyDomain.State(record.State), + } +} + +func validCandidateIdentity(candidate proxyDomain.Proxy) bool { + if candidate.Host == "" || candidate.Port == 0 { + return false + } + return validScheme(string(candidate.Scheme)) +} + +func stableAdapterProxyID(uniqueKey string) string { + digest := sha256.Sum256([]byte(uniqueKey)) + return "px_" + hex.EncodeToString(digest[:12]) +} + +func cloneTags(tags map[string]string) map[string]string { + if tags == nil { + return nil + } + cloned := make(map[string]string, len(tags)) + for key, value := range tags { + cloned[key] = value + } + return cloned +} + +func invalidScriptReply(reason string) error { + return errors.Join(extractionDomain.ErrStoreUnavailable, errors.New(reason)) +} diff --git a/internal/adapters/redisactivity/upsert_integration_test.go b/internal/adapters/redisactivity/upsert_integration_test.go new file mode 100644 index 0000000..cd9f788 --- /dev/null +++ b/internal/adapters/redisactivity/upsert_integration_test.go @@ -0,0 +1,230 @@ +//go:build integration + +package redisactivity + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/platform/credentials" +) + +func TestRedisUpsertEnforcesMaxSizeWithoutLeavingRejectedMappings(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + first := testProxy("proxy-a", "192.0.2.10") + second := testProxy("proxy-b", "192.0.2.11") + + result, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: 30 * time.Second, MaxSize: 1, + Proxies: []proxyDomain.Proxy{first, second}, + }) + if err != nil || result.Accepted != 2 || result.Inserted != 1 || result.Dropped != 1 { + t.Fatalf("UpsertFetched() = %+v, %v", result, err) + } + + retry, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now.Add(time.Second), ConfiguredTTL: 30 * time.Second, MaxSize: 2, + Proxies: []proxyDomain.Proxy{second}, + }) + if err != nil || retry.Inserted != 1 || retry.Refreshed != 0 { + t.Fatalf("UpsertFetched(capacity retry) = %+v, %v", retry, err) + } +} + +func TestRedisUpsertChunksLargeProviderResponsesWithGlobalCapacity(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + proxies := make([]proxyDomain.Proxy, 300) + for index := range proxies { + proxies[index] = testProxy("proxy-"+fmt.Sprint(index), fmt.Sprintf("192.0.2.%d", index%250+1)) + proxies[index].Port = uint16(10_000 + index) + } + + result, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 275, Proxies: proxies, + }) + if err != nil || result.Accepted != 300 || result.Inserted != 275 || result.Dropped != 25 { + t.Fatalf("UpsertFetched(large batch) = %+v, %v", result, err) + } +} + +func TestRedisUpsertPreservesIncumbentLifecycleAndRuntimeHealth(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + candidate := testProxy("proxy-a", "192.0.2.10") + + first, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: 30 * time.Second, + AllocationSafetyMargin: 3 * time.Second, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }) + if err != nil || first.Inserted != 1 { + t.Fatalf("first UpsertFetched() = %+v, %v", first, err) + } + if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(checking): %v", err) + } + healthyAt := now.Add(2 * time.Second) + if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: healthyAt, + NextState: proxyDomain.StateAvailable, Latency: 25 * time.Millisecond, + }); err != nil { + t.Fatalf("ApplyHealth(available): %v", err) + } + + refreshed, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now.Add(3 * time.Second), ConfiguredTTL: time.Minute, + AllocationSafetyMargin: 5 * time.Second, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }) + if err != nil || refreshed.Refreshed != 1 || refreshed.Inserted != 0 { + t.Fatalf("same-provider refresh = %+v, %v", refreshed, err) + } + entry, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: healthyAt, NextState: proxyDomain.StateAvailable, + }) + if err != nil || entry.State != proxyDomain.StateAvailable || entry.Proxy.Latency != 25*time.Millisecond || + entry.Proxy.SourceUpstream != "provider-a" || entry.Proxy.ExpiresAt == nil || + !entry.Proxy.ExpiresAt.Equal(now.Add(63*time.Second)) { + t.Fatalf("refreshed entry = %+v, %v", entry, err) + } + + duplicate, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-b", activitypool.FetchedBatch{ + ObservedAt: now.Add(4 * time.Second), ConfiguredTTL: 5 * time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }) + if err != nil || duplicate.Refreshed != 1 || duplicate.Inserted != 0 { + t.Fatalf("cross-provider duplicate = %+v, %v", duplicate, err) + } + entry, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: healthyAt, NextState: proxyDomain.StateAvailable, + }) + if err != nil || entry.Proxy.SourceUpstream != "provider-a" || + entry.Proxy.ExpiresAt == nil || !entry.Proxy.ExpiresAt.Equal(now.Add(63*time.Second)) { + t.Fatalf("incumbent entry = %+v, %v", entry, err) + } + + replacementAt := now.Add(64 * time.Second) + replaced, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-b", activitypool.FetchedBatch{ + ObservedAt: replacementAt, ConfiguredTTL: time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }) + if err != nil || replaced.Inserted != 1 || replaced.Refreshed != 0 { + t.Fatalf("expired incumbent replacement = %+v, %v", replaced, err) + } + entry, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: replacementAt.Add(time.Second), NextState: proxyDomain.StateChecking, + }) + if err != nil || entry.Proxy.SourceUpstream != "provider-b" { + t.Fatalf("replacement entry = %+v, %v", entry, err) + } +} + +func TestRedisUpsertResolvesCredentialsBeforeCommit(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + candidate := testProxy("proxy-a", "192.0.2.10") + candidate.Username = "user" + candidate.SecretRef = "cred_missing" + candidate.CredentialVersion = "v1" + + _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }) + if !errors.Is(err, credentials.ErrCredentialMissing) { + t.Fatalf("UpsertFetched(missing credential) error = %v", err) + } + + candidate.SecretRef = "" + candidate.CredentialVersion = "" + retry, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }) + if err != nil || retry.Inserted != 1 || retry.Refreshed != 0 { + t.Fatalf("UpsertFetched(after resolution failure) = %+v, %v", retry, err) + } + + reference, err := fixture.Credentials.Put(context.Background(), "proxy-b", credentials.Value{ + Username: "user", Password: "password", + }) + if err != nil { + t.Fatalf("Credentials.Put(): %v", err) + } + withCredential := testProxy("proxy-b", "192.0.2.11") + withCredential.Username = "user" + withCredential.SecretRef = reference.SecretRef + withCredential.CredentialVersion = reference.CredentialVersion + stored, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{withCredential}, + }) + if err != nil || stored.Inserted != 1 { + t.Fatalf("UpsertFetched(resolved credential) = %+v, %v", stored, err) + } +} + +func TestRedisHealthTransitionsAreMonotonicAndIdempotent(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")}, + }); err != nil { + t.Fatalf("UpsertFetched(): %v", err) + } + if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(checking): %v", err) + } + checkedAt := now.Add(2 * time.Second) + available, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: checkedAt, + NextState: proxyDomain.StateAvailable, Latency: 30 * time.Millisecond, + }) + if err != nil || available.State != proxyDomain.StateAvailable || + available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(checkedAt) { + t.Fatalf("ApplyHealth(available) = %+v, %v", available, err) + } + + replayed, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: checkedAt, + NextState: proxyDomain.StateAvailable, Latency: time.Second, + }) + if err != nil || replayed.Proxy.Latency != 30*time.Millisecond { + t.Fatalf("ApplyHealth(replay) = %+v, %v", replayed, err) + } + _, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: checkedAt, NextState: proxyDomain.StateSuspect, + }) + if !errors.Is(err, activitypool.ErrStaleHealthUpdate) { + t.Fatalf("ApplyHealth(conflicting replay) error = %v", err) + } + _, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateSuspect, + }) + if !errors.Is(err, activitypool.ErrStaleHealthUpdate) { + t.Fatalf("ApplyHealth(stale) error = %v", err) + } +} + +func redisTestNow() time.Time { + return time.Now().UTC().Add(time.Hour).Truncate(time.Millisecond) +} + +func testProxy(id, host string) proxyDomain.Proxy { + return proxyDomain.Proxy{ + ID: id, Scheme: proxyDomain.SchemeHTTP, Host: host, Port: 8080, + State: proxyDomain.StateFetched, Tags: map[string]string{"region": "cn", "carrier": "ct"}, + } +}