diff --git a/configs/proxy-pool.yaml b/configs/proxy-pool.yaml index 19ff874..5f35e1b 100644 --- a/configs/proxy-pool.yaml +++ b/configs/proxy-pool.yaml @@ -5,6 +5,7 @@ security: defaults: fetch: + estimatedIPsPerCall: 100 requestInterval: 1s timeout: 5s maxAttempts: 3 @@ -139,10 +140,15 @@ upstreams: shrinkDelay: 30s capacity: maxConcurrencyPerProxy: 20 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 8000 + targetAvailableSlots: 12000 lifecycle: ttl: 5m allocationSafetyMargin: 30s fetch: + estimatedIPsPerCall: 100 requestInterval: 1s timeout: 5s maxAttempts: 3 @@ -192,10 +198,15 @@ upstreams: shrinkDelay: 30s capacity: maxConcurrencyPerProxy: 10 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 8000 + targetAvailableSlots: 12000 lifecycle: ttl: 2m allocationSafetyMargin: 20s fetch: + estimatedIPsPerCall: 100 requestInterval: 2s timeout: 5s maxAttempts: 3 diff --git a/deploy/config/local.yaml b/deploy/config/local.yaml index 307f0f6..17079df 100644 --- a/deploy/config/local.yaml +++ b/deploy/config/local.yaml @@ -111,10 +111,15 @@ upstreams: shrinkDelay: 30s capacity: maxConcurrencyPerProxy: 20 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 8000 + targetAvailableSlots: 12000 lifecycle: ttl: 5m allocationSafetyMargin: 20s fetch: + estimatedIPsPerCall: 100 requestInterval: 1s timeout: 3s maxAttempts: 3 @@ -156,10 +161,15 @@ upstreams: shrinkDelay: 30s capacity: maxConcurrencyPerProxy: 20 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 8000 + targetAvailableSlots: 12000 lifecycle: ttl: 5m allocationSafetyMargin: 20s fetch: + estimatedIPsPerCall: 100 requestInterval: 1s timeout: 3s maxAttempts: 3 diff --git a/deploy/kubernetes/base/configmap.yaml b/deploy/kubernetes/base/configmap.yaml index 52f1329..62116b9 100644 --- a/deploy/kubernetes/base/configmap.yaml +++ b/deploy/kubernetes/base/configmap.yaml @@ -110,10 +110,15 @@ data: shrinkDelay: 30s capacity: maxConcurrencyPerProxy: 20 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 8000 + targetAvailableSlots: 12000 lifecycle: ttl: 5m allocationSafetyMargin: 20s fetch: + estimatedIPsPerCall: 100 requestInterval: 1s timeout: 3s maxAttempts: 3 @@ -152,10 +157,15 @@ data: shrinkDelay: 30s capacity: maxConcurrencyPerProxy: 20 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 8000 + targetAvailableSlots: 12000 lifecycle: ttl: 5m allocationSafetyMargin: 20s fetch: + estimatedIPsPerCall: 100 requestInterval: 1s timeout: 3s maxAttempts: 3 diff --git a/examples/config/01-local-all.yaml b/examples/config/01-local-all.yaml index aa8cc9a..09d5ac1 100644 --- a/examples/config/01-local-all.yaml +++ b/examples/config/01-local-all.yaml @@ -48,6 +48,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 100} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 10s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 1000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 1000} check: {interval: 30s, jitter: 20, maxInFlight: 50, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/02-gateway-only.yaml b/examples/config/02-gateway-only.yaml index bb87463..87a5c27 100644 --- a/examples/config/02-gateway-only.yaml +++ b/examples/config/02-gateway-only.yaml @@ -28,6 +28,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 2000} capacity: {maxConcurrencyPerProxy: 20} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 5m, allocationSafetyMargin: 30s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/03-extract-only.yaml b/examples/config/03-extract-only.yaml index bdd13d8..7d320f4 100644 --- a/examples/config/03-extract-only.yaml +++ b/examples/config/03-extract-only.yaml @@ -30,6 +30,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 500} capacity: {maxConcurrencyPerProxy: 1} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 10s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} check: {interval: 15s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/04-public-gateway-basic-auth.yaml b/examples/config/04-public-gateway-basic-auth.yaml index e13093f..c2fda96 100644 --- a/examples/config/04-public-gateway-basic-auth.yaml +++ b/examples/config/04-public-gateway-basic-auth.yaml @@ -32,6 +32,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 3000} capacity: {maxConcurrencyPerProxy: 20} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 5m, allocationSafetyMargin: 30s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 200, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/05-public-extract-api-key.yaml b/examples/config/05-public-extract-api-key.yaml index b9eaa4f..be93f61 100644 --- a/examples/config/05-public-extract-api-key.yaml +++ b/examples/config/05-public-extract-api-key.yaml @@ -33,6 +33,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 1} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 3m, allocationSafetyMargin: 30s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 100000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 100000} check: {interval: 15s, jitter: 20, maxInFlight: 200, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/06-internal-cidr-no-auth.yaml b/examples/config/06-internal-cidr-no-auth.yaml index b2910f9..25b3bdb 100644 --- a/examples/config/06-internal-cidr-no-auth.yaml +++ b/examples/config/06-internal-cidr-no-auth.yaml @@ -40,6 +40,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 20s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/07-auth-any.yaml b/examples/config/07-auth-any.yaml index 41bb4e3..ba31331 100644 --- a/examples/config/07-auth-any.yaml +++ b/examples/config/07-auth-any.yaml @@ -32,6 +32,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 500} capacity: {maxConcurrencyPerProxy: 1} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} check: {interval: 15s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/08-sequential-failover.yaml b/examples/config/08-sequential-failover.yaml index b139554..2bf037e 100644 --- a/examples/config/08-sequential-failover.yaml +++ b/examples/config/08-sequential-failover.yaml @@ -22,8 +22,9 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 500} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} provider-b: enabled: true @@ -33,6 +34,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 500} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 3m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 2s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 2s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/09-weighted-routing.yaml b/examples/config/09-weighted-routing.yaml index 2159d6a..a60d601 100644 --- a/examples/config/09-weighted-routing.yaml +++ b/examples/config/09-weighted-routing.yaml @@ -21,8 +21,9 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} provider-b: enabled: true @@ -32,6 +33,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/10-round-robin-routing.yaml b/examples/config/10-round-robin-routing.yaml index 727dbe1..a7af03b 100644 --- a/examples/config/10-round-robin-routing.yaml +++ b/examples/config/10-round-robin-routing.yaml @@ -19,8 +19,9 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} provider-b: enabled: true exposure: [gateway] @@ -29,5 +30,6 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} diff --git a/examples/config/11-random-routing.yaml b/examples/config/11-random-routing.yaml index 27ff8d9..fe49687 100644 --- a/examples/config/11-random-routing.yaml +++ b/examples/config/11-random-routing.yaml @@ -19,8 +19,9 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} provider-b: enabled: true exposure: [gateway] @@ -29,5 +30,6 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} diff --git a/examples/config/12-least-connections-routing.yaml b/examples/config/12-least-connections-routing.yaml index 3a2fd7d..543b712 100644 --- a/examples/config/12-least-connections-routing.yaml +++ b/examples/config/12-least-connections-routing.yaml @@ -23,6 +23,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 10000} capacity: {maxConcurrencyPerProxy: 50} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 5m, allocationSafetyMargin: 30s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 500, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/13-extract-all-or-nothing.yaml b/examples/config/13-extract-all-or-nothing.yaml index 1a75dd5..3d4c704 100644 --- a/examples/config/13-extract-all-or-nothing.yaml +++ b/examples/config/13-extract-all-or-nothing.yaml @@ -29,6 +29,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 500} capacity: {maxConcurrencyPerProxy: 1} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 50000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 50000} check: {interval: 15s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/14-gateway-reserve.yaml b/examples/config/14-gateway-reserve.yaml index c381c35..6ba8c48 100644 --- a/examples/config/14-gateway-reserve.yaml +++ b/examples/config/14-gateway-reserve.yaml @@ -37,6 +37,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 2000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 3m, allocationSafetyMargin: 20s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 15s, jitter: 20, maxInFlight: 200, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/15-strict-ttl-health.yaml b/examples/config/15-strict-ttl-health.yaml index b0f823e..0900a7b 100644 --- a/examples/config/15-strict-ttl-health.yaml +++ b/examples/config/15-strict-ttl-health.yaml @@ -29,6 +29,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 1} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 5m, allocationSafetyMargin: 60s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 100000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 100000} check: {interval: 5s, jitter: 20, maxInFlight: 500, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/16-provider-basic-auth.yaml b/examples/config/16-provider-basic-auth.yaml index 44815ff..32c9ebc 100644 --- a/examples/config/16-provider-basic-auth.yaml +++ b/examples/config/16-provider-basic-auth.yaml @@ -26,6 +26,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/17-provider-api-key.yaml b/examples/config/17-provider-api-key.yaml index b853e0a..1068d86 100644 --- a/examples/config/17-provider-api-key.yaml +++ b/examples/config/17-provider-api-key.yaml @@ -27,6 +27,7 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 500} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 10000} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/18-provider-post-json.yaml b/examples/config/18-provider-post-json.yaml index 0427f48..57dd209 100644 --- a/examples/config/18-provider-post-json.yaml +++ b/examples/config/18-provider-post-json.yaml @@ -27,8 +27,10 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 3m, allocationSafetyMargin: 20s} fetch: + estimatedIPsPerCall: 10 requestInterval: 2s timeout: 5s maxAttempts: 3 diff --git a/examples/config/19-socks5-upstream.yaml b/examples/config/19-socks5-upstream.yaml index c3c4895..bc3bdb7 100644 --- a/examples/config/19-socks5-upstream.yaml +++ b/examples/config/19-socks5-upstream.yaml @@ -22,6 +22,7 @@ upstreams: password: "${SOCKS_PASSWORD}" pool: {maxSize: 1000} capacity: {maxConcurrencyPerProxy: 20} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 10m, allocationSafetyMargin: 60s} - fetch: {requestInterval: 5s, timeout: 5s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 10, requestInterval: 5s, timeout: 5s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 3s, maxAttempts: 2, maxConsecutiveFailures: 3} diff --git a/examples/config/20-fetch-billing-quota.yaml b/examples/config/20-fetch-billing-quota.yaml index b730204..d32b79c 100644 --- a/examples/config/20-fetch-billing-quota.yaml +++ b/examples/config/20-fetch-billing-quota.yaml @@ -24,8 +24,10 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 100} capacity: {maxConcurrencyPerProxy: 1} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 25, targetAvailableSlots: 50} lifecycle: {ttl: 2m, allocationSafetyMargin: 15s} fetch: + estimatedIPsPerCall: 10 requestInterval: 1s timeout: 3s maxAttempts: 3 diff --git a/internal/adapters/redisprovider/adapter.go b/internal/adapters/redisprovider/adapter.go new file mode 100644 index 0000000..bfafab6 --- /dev/null +++ b/internal/adapters/redisprovider/adapter.go @@ -0,0 +1,355 @@ +package redisprovider + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "math" + "reflect" + "regexp" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" + + controllerProvider "proxy-pool/internal/controller/provider" +) + +var namespacePattern = regexp.MustCompile(`^[A-Za-z0-9._-]+$`) + +type Options struct { + Namespace string + HolderID string + LeaseTTL time.Duration + RenewEvery time.Duration + RetryInterval time.Duration + PermitGrace time.Duration +} + +type Adapter struct { + client redis.Scripter + options Options + keys keyBuilder +} + +var _ controllerProvider.Coordinator = (*Adapter)(nil) + +func New(client redis.Scripter, options Options) (*Adapter, error) { + options.Namespace = strings.TrimSpace(options.Namespace) + options.HolderID = strings.TrimSpace(options.HolderID) + if nilInterface(client) || !namespacePattern.MatchString(options.Namespace) || options.HolderID == "" || + options.LeaseTTL <= 0 || options.RenewEvery <= 0 || options.RenewEvery > options.LeaseTTL/3 || + options.RetryInterval <= 0 || options.PermitGrace < 0 { + return nil, controllerProvider.ErrInvalidCoordination + } + return &Adapter{client: client, options: options, keys: keyBuilder{namespace: options.Namespace}}, nil +} + +func (adapter *Adapter) RunLeader( + ctx context.Context, + upstreamID string, + limits controllerProvider.CoordinationLimits, + work func(context.Context, controllerProvider.LeaderSession) error, +) error { + if ctx == nil || adapter == nil || work == nil || strings.TrimSpace(upstreamID) != upstreamID || upstreamID == "" || + limits.RequestInterval < 0 || limits.MaxInFlight <= 0 || limits.MaxAttemptDuration <= 0 || + limits.MaxAttemptDuration > time.Duration(math.MaxInt64)-adapter.options.PermitGrace { + return controllerProvider.ErrInvalidCoordination + } + if err := ctx.Err(); err != nil { + return err + } + keys, err := adapter.keys.forUpstream(upstreamID) + if err != nil { + return err + } + token, err := randomToken() + if err != nil { + return errors.Join(controllerProvider.ErrCoordinationUnavailable, err) + } + + for ctx.Err() == nil { + generationCandidate, tokenErr := randomToken() + if tokenErr != nil { + return errors.Join(controllerProvider.ErrCoordinationUnavailable, tokenErr) + } + reply, acquireErr := runScript(ctx, adapter.client, keys, + "acquire_leader", generationCandidate, adapter.options.HolderID, token, + durationMillis(adapter.options.LeaseTTL), + ) + if acquireErr != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if err := wait(ctx, adapter.options.RetryInterval); err != nil { + return err + } + continue + } + switch reply.Status { + case "busy": + delay := adapter.options.RetryInterval + if reply.WaitMS > 0 && time.Duration(reply.WaitMS)*time.Millisecond < delay { + delay = time.Duration(reply.WaitMS) * time.Millisecond + } + if err := wait(ctx, delay); err != nil { + return err + } + continue + case "ok": + if reply.Generation == "" || reply.Epoch == 0 { + if err := wait(ctx, adapter.options.RetryInterval); err != nil { + return err + } + continue + } + default: + if err := wait(ctx, adapter.options.RetryInterval); err != nil { + return err + } + continue + } + + session := &leaderSession{ + adapter: adapter, keys: keys, upstreamID: upstreamID, limits: limits, + generation: reply.Generation, holderID: adapter.options.HolderID, + token: token, epoch: reply.Epoch, + } + lost, runErr := adapter.runLeaderTerm(ctx, session, work) + if runErr != nil { + return runErr + } + if !lost { + return nil + } + if err := wait(ctx, adapter.options.RetryInterval); err != nil { + return err + } + } + return ctx.Err() +} + +func (adapter *Adapter) runLeaderTerm( + ctx context.Context, + session *leaderSession, + work func(context.Context, controllerProvider.LeaderSession) error, +) (bool, error) { + leaderCtx, cancel := context.WithCancel(ctx) + defer cancel() + session.ctx = leaderCtx + workDone := make(chan error, 1) + go func() { workDone <- work(leaderCtx, session) }() + + ticker := time.NewTicker(adapter.options.RenewEvery) + defer ticker.Stop() + deadline := time.NewTimer(adapter.options.LeaseTTL - adapter.options.RenewEvery) + defer deadline.Stop() + for { + select { + case <-ctx.Done(): + cancel() + <-workDone + adapter.releaseLeader(session) + return false, ctx.Err() + case workErr := <-workDone: + cancel() + adapter.releaseLeader(session) + return false, leaderWorkResult(ctx, workErr) + case <-deadline.C: + cancel() + <-workDone + return true, nil + case <-ticker.C: + renewCtx, renewCancel := context.WithTimeout(leaderCtx, adapter.options.RenewEvery) + reply, err := runScript(renewCtx, adapter.client, session.keys, + "renew_leader", session.generation, session.holderID, session.token, + session.epoch, durationMillis(adapter.options.LeaseTTL), + ) + renewCancel() + if err != nil || reply.Status != "ok" { + cancel() + <-workDone + return true, nil + } + resetTimer(deadline, adapter.options.LeaseTTL-adapter.options.RenewEvery) + } + } +} + +func leaderWorkResult(ctx context.Context, workErr error) error { + if err := ctx.Err(); err != nil { + return err + } + if workErr != nil { + return workErr + } + return controllerProvider.ErrLeaderWorkStopped +} + +func (adapter *Adapter) releaseLeader(session *leaderSession) { + ctx, cancel := context.WithTimeout(context.Background(), adapter.options.RenewEvery) + defer cancel() + _, _ = runScript(ctx, adapter.client, session.keys, + "release_leader", session.generation, session.holderID, session.token, session.epoch, + ) +} + +type leaderSession struct { + adapter *Adapter + keys upstreamKeys + upstreamID string + limits controllerProvider.CoordinationLimits + ctx context.Context + generation string + holderID string + token string + epoch uint64 +} + +var _ controllerProvider.LeaderSession = (*leaderSession)(nil) + +func (session *leaderSession) Fence() controllerProvider.Fence { + if session == nil { + return controllerProvider.Fence{} + } + return controllerProvider.Fence{Generation: session.generation, Epoch: session.epoch} +} + +func (session *leaderSession) AcquireFetch(ctx context.Context) (controllerProvider.RequestPermit, error) { + if ctx == nil || session == nil || session.adapter == nil || session.ctx == nil { + return nil, controllerProvider.ErrInvalidCoordination + } + operationCtx, cancel := context.WithCancel(ctx) + stop := context.AfterFunc(session.ctx, cancel) + defer func() { + stop() + cancel() + }() + permitToken, err := randomToken() + if err != nil { + return nil, errors.Join(controllerProvider.ErrCoordinationUnavailable, err) + } + permitTTL := session.limits.MaxAttemptDuration + session.adapter.options.PermitGrace + for operationCtx.Err() == nil { + reply, scriptErr := runScript(operationCtx, session.adapter.client, session.keys, + "acquire_fetch", session.generation, session.holderID, session.token, session.epoch, + permitToken, durationMillis(session.limits.RequestInterval), session.limits.MaxInFlight, + durationMillis(permitTTL), + ) + if scriptErr != nil { + if session.ctx.Err() != nil { + return nil, controllerProvider.ErrLeadershipLost + } + if operationCtx.Err() != nil { + return nil, operationCtx.Err() + } + if err := wait(operationCtx, session.adapter.options.RetryInterval); err != nil { + return nil, err + } + continue + } + switch reply.Status { + case "ok": + return &requestPermit{adapter: session.adapter, keys: session.keys, token: permitToken}, nil + case "stale": + return nil, controllerProvider.ErrLeadershipLost + case "rate_limited", "at_capacity": + delay := time.Duration(reply.WaitMS) * time.Millisecond + if delay <= 0 { + delay = session.adapter.options.RetryInterval + } + if err := wait(operationCtx, delay); err != nil { + if session.ctx.Err() != nil { + return nil, controllerProvider.ErrLeadershipLost + } + return nil, err + } + default: + return nil, controllerProvider.ErrCoordinationUnavailable + } + } + if session.ctx.Err() != nil { + return nil, controllerProvider.ErrLeadershipLost + } + return nil, operationCtx.Err() +} + +type requestPermit struct { + adapter *Adapter + keys upstreamKeys + token string + mu sync.Mutex + done bool +} + +func (permit *requestPermit) Release(ctx context.Context) error { + if ctx == nil || permit == nil || permit.adapter == nil || permit.token == "" { + return controllerProvider.ErrInvalidCoordination + } + permit.mu.Lock() + defer permit.mu.Unlock() + if permit.done { + return nil + } + reply, err := runScript(ctx, permit.adapter.client, permit.keys, "release_fetch", permit.token) + if err != nil { + return err + } + if reply.Status != "ok" { + return controllerProvider.ErrCoordinationUnavailable + } + permit.done = true + return nil +} + +func randomToken() (string, error) { + var token [16]byte + if _, err := rand.Read(token[:]); err != nil { + return "", err + } + return hex.EncodeToString(token[:]), nil +} + +func durationMillis(value time.Duration) int64 { + milliseconds := value / time.Millisecond + if value%time.Millisecond != 0 { + milliseconds++ + } + return int64(milliseconds) +} + +func wait(ctx context.Context, duration time.Duration) error { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func resetTimer(timer *time.Timer, duration time.Duration) { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(duration) +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/adapters/redisprovider/adapter_test.go b/internal/adapters/redisprovider/adapter_test.go new file mode 100644 index 0000000..278b52e --- /dev/null +++ b/internal/adapters/redisprovider/adapter_test.go @@ -0,0 +1,129 @@ +package redisprovider + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/redis/go-redis/v9" + + controllerProvider "proxy-pool/internal/controller/provider" +) + +func TestNewRejectsInvalidDependenciesAndOptions(t *testing.T) { + t.Parallel() + client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"}) + t.Cleanup(func() { _ = client.Close() }) + valid := Options{ + Namespace: "controller", HolderID: "controller-a", LeaseTTL: 3 * time.Second, + RenewEvery: time.Second, RetryInterval: 50 * time.Millisecond, PermitGrace: time.Second, + } + var typedNil *redis.Client + tests := []struct { + name string + client redis.Scripter + options Options + }{ + {name: "nil client", options: valid}, + {name: "typed nil client", client: typedNil, options: valid}, + {name: "empty namespace", client: client, options: withNamespace(valid, "")}, + {name: "unsafe namespace", client: client, options: withNamespace(valid, "bad:value")}, + {name: "empty holder", client: client, options: withHolder(valid, "")}, + {name: "zero lease", client: client, options: withLeaseTTL(valid, 0)}, + {name: "renew exceeds third", client: client, options: withRenewEvery(valid, 2*time.Second)}, + {name: "zero retry", client: client, options: withRetryInterval(valid, 0)}, + {name: "negative grace", client: client, options: withPermitGrace(valid, -1)}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + adapter, err := New(test.client, test.options) + if err == nil || adapter != nil { + t.Fatalf("New() = (%v, %v), want nil adapter and error", adapter, err) + } + }) + } +} + +func TestNewBuildsPerUpstreamClusterSafeKeys(t *testing.T) { + t.Parallel() + client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"}) + t.Cleanup(func() { _ = client.Close() }) + adapter, err := New(client, Options{ + Namespace: " controller ", HolderID: "controller-a", LeaseTTL: 3 * time.Second, + RenewEvery: time.Second, RetryInterval: 50 * time.Millisecond, PermitGrace: time.Second, + }) + if err != nil { + t.Fatalf("New(): %v", err) + } + keys, err := adapter.keys.forUpstream("provider:{unsafe}") + if err != nil { + t.Fatalf("forUpstream(): %v", err) + } + all := keys.all() + for _, key := range all { + if strings.Contains(key, "provider:{unsafe}") || strings.Count(key, "{") != 1 || + strings.Count(key, "}") != 1 || !strings.Contains(key, "{provider:") { + t.Fatalf("unsafe provider coordination key %q", key) + } + } + if strings.Split(all[0], "}")[0] != strings.Split(all[len(all)-1], "}")[0] { + t.Fatalf("keys do not share one upstream hash tag: %v", all) + } + if _, err := adapter.keys.forUpstream(""); !errors.Is(err, controllerProvider.ErrInvalidCoordination) { + t.Fatalf("forUpstream(empty) error = %v", err) + } +} + +func TestRunLeaderRejectsInvalidCalls(t *testing.T) { + t.Parallel() + var adapter *Adapter + limits := controllerProvider.CoordinationLimits{ + RequestInterval: time.Second, MaxInFlight: 1, MaxAttemptDuration: time.Second, + } + work := func(context.Context, controllerProvider.LeaderSession) error { return nil } + if err := adapter.RunLeader(context.Background(), "provider-a", limits, work); !errors.Is(err, controllerProvider.ErrInvalidCoordination) { + t.Fatalf("nil adapter error = %v", err) + } +} + +func TestLeaderWorkResultPrefersParentCancellation(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := leaderWorkResult(ctx, nil); !errors.Is(err, context.Canceled) { + t.Fatalf("leaderWorkResult() error = %v, want context cancellation", err) + } +} + +func withNamespace(options Options, value string) Options { + options.Namespace = value + return options +} + +func withHolder(options Options, value string) Options { + options.HolderID = value + return options +} + +func withLeaseTTL(options Options, value time.Duration) Options { + options.LeaseTTL = value + return options +} + +func withRenewEvery(options Options, value time.Duration) Options { + options.RenewEvery = value + return options +} + +func withRetryInterval(options Options, value time.Duration) Options { + options.RetryInterval = value + return options +} + +func withPermitGrace(options Options, value time.Duration) Options { + options.PermitGrace = value + return options +} diff --git a/internal/adapters/redisprovider/coordination_integration_test.go b/internal/adapters/redisprovider/coordination_integration_test.go new file mode 100644 index 0000000..f041571 --- /dev/null +++ b/internal/adapters/redisprovider/coordination_integration_test.go @@ -0,0 +1,297 @@ +//go:build integration + +package redisprovider + +import ( + "context" + "fmt" + "os" + "sync/atomic" + "testing" + "time" + + "github.com/redis/go-redis/v9" + + controllerProvider "proxy-pool/internal/controller/provider" +) + +var integrationNamespaceSequence atomic.Uint64 + +func TestRedisCoordinatorElectsOneLeaderAndFencesFailover(t *testing.T) { + fixture := newRedisFixture(t) + first := fixture.coordinator(t, "controller-a") + second := fixture.coordinator(t, "controller-b") + limits := controllerProvider.CoordinationLimits{ + RequestInterval: 50 * time.Millisecond, MaxInFlight: 1, MaxAttemptDuration: 300 * time.Millisecond, + } + + ctxA, cancelA := context.WithCancel(context.Background()) + ctxB, cancelB := context.WithCancel(context.Background()) + defer cancelA() + defer cancelB() + started := make(chan leadershipFixture, 4) + var active atomic.Int64 + var maximum atomic.Int64 + work := func(holder string) func(context.Context, controllerProvider.LeaderSession) error { + return func(ctx context.Context, session controllerProvider.LeaderSession) error { + current := active.Add(1) + for { + observed := maximum.Load() + if current <= observed || maximum.CompareAndSwap(observed, current) { + break + } + } + started <- leadershipFixture{holder: holder, fence: session.Fence()} + <-ctx.Done() + active.Add(-1) + return nil + } + } + doneA := make(chan error, 1) + doneB := make(chan error, 1) + go func() { doneA <- first.RunLeader(ctxA, "provider-a", limits, work("controller-a")) }() + go func() { doneB <- second.RunLeader(ctxB, "provider-a", limits, work("controller-b")) }() + + initial := receiveLeadership(t, started) + time.Sleep(200 * time.Millisecond) + select { + case duplicate := <-started: + t.Fatalf("simultaneous leaders started: first=%+v duplicate=%+v", initial, duplicate) + default: + } + if got := maximum.Load(); got != 1 { + t.Fatalf("maximum simultaneous leaders = %d, want 1", got) + } + if initial.holder == "controller-a" { + cancelA() + } else { + cancelB() + } + replacement := receiveLeadership(t, started) + if replacement.holder == initial.holder { + t.Fatalf("replacement holder = %q, want the other controller", replacement.holder) + } + if replacement.fence.Generation != initial.fence.Generation || replacement.fence.Epoch <= initial.fence.Epoch { + t.Fatalf("replacement fence = %+v, initial = %+v", replacement.fence, initial.fence) + } + if got := maximum.Load(); got != 1 { + t.Fatalf("maximum simultaneous leaders after failover = %d, want 1", got) + } + + cancelA() + cancelB() + waitRunner(t, doneA) + waitRunner(t, doneB) +} + +func TestRedisLeaderSessionEnforcesGlobalIntervalAndInFlightLimit(t *testing.T) { + fixture := newRedisFixture(t) + coordinator := fixture.coordinator(t, "controller-a") + limits := controllerProvider.CoordinationLimits{ + RequestInterval: 250 * time.Millisecond, MaxInFlight: 1, MaxAttemptDuration: time.Second, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + sessions := make(chan controllerProvider.LeaderSession, 1) + done := make(chan error, 1) + go func() { + done <- coordinator.RunLeader(ctx, "provider-a", limits, + func(workCtx context.Context, session controllerProvider.LeaderSession) error { + sessions <- session + <-workCtx.Done() + return nil + }) + }() + session := receiveSession(t, sessions) + first, err := session.AcquireFetch(context.Background()) + if err != nil { + t.Fatalf("first AcquireFetch(): %v", err) + } + startedAt := time.Now() + secondResult := make(chan permitResultFixture, 1) + go func() { + permit, acquireErr := session.AcquireFetch(context.Background()) + secondResult <- permitResultFixture{permit: permit, err: acquireErr} + }() + select { + case result := <-secondResult: + t.Fatalf("second AcquireFetch() returned before release: %+v", result) + case <-time.After(100 * time.Millisecond): + } + if err := first.Release(context.Background()); err != nil { + t.Fatalf("first Release(): %v", err) + } + result := receivePermit(t, secondResult) + if result.err != nil || result.permit == nil { + t.Fatalf("second AcquireFetch() = (%v, %v)", result.permit, result.err) + } + if elapsed := time.Since(startedAt); elapsed < 200*time.Millisecond { + t.Fatalf("global request interval = %s, want at least 200ms", elapsed) + } + if err := result.permit.Release(context.Background()); err != nil { + t.Fatalf("second Release(): %v", err) + } + if err := result.permit.Release(context.Background()); err != nil { + t.Fatalf("idempotent second Release(): %v", err) + } + cancel() + waitRunner(t, done) +} + +func TestRedisCoordinatorRebuildsWithNewGenerationAfterStateLoss(t *testing.T) { + fixture := newRedisFixture(t) + coordinator := fixture.coordinator(t, "controller-a") + limits := controllerProvider.CoordinationLimits{ + RequestInterval: 50 * time.Millisecond, MaxInFlight: 1, MaxAttemptDuration: time.Second, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + started := make(chan controllerProvider.Fence, 4) + done := make(chan error, 1) + go func() { + done <- coordinator.RunLeader(ctx, "provider-a", limits, + func(workCtx context.Context, session controllerProvider.LeaderSession) error { + started <- session.Fence() + <-workCtx.Done() + return nil + }) + }() + first := receiveFence(t, started) + fixture.deleteKeys(t) + second := receiveFence(t, started) + if second.Generation == first.Generation { + t.Fatalf("generation after Redis state loss = %q, want a new generation", second.Generation) + } + cancel() + waitRunner(t, done) +} + +type redisFixture struct { + client *redis.Client + namespace string +} + +func newRedisFixture(t *testing.T) redisFixture { + t.Helper() + redisURL := os.Getenv("PROXY_POOL_TEST_REDIS_URL") + if redisURL == "" { + t.Skip("PROXY_POOL_TEST_REDIS_URL is not set") + } + options, err := redis.ParseURL(redisURL) + if err != nil { + t.Fatalf("parse PROXY_POOL_TEST_REDIS_URL: %v", err) + } + client := redis.NewClient(options) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := client.Ping(ctx).Err(); err != nil { + _ = client.Close() + t.Fatalf("ping Redis: %v", err) + } + namespace := fmt.Sprintf("provider-it-%d-%d-%d", os.Getpid(), time.Now().UnixNano(), integrationNamespaceSequence.Add(1)) + t.Cleanup(func() { + redisFixture{client: client, namespace: namespace}.deleteKeys(t) + _ = client.Close() + }) + return redisFixture{client: client, namespace: namespace} +} + +func (fixture redisFixture) deleteKeys(t *testing.T) { + t.Helper() + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cleanupCancel() + var cursor uint64 + for { + keys, next, err := fixture.client.Scan(cleanupCtx, cursor, "pp:"+fixture.namespace+":*", 128).Result() + if err != nil { + t.Fatalf("scan Redis provider keys: %v", err) + } + if len(keys) > 0 { + if err := fixture.client.Unlink(cleanupCtx, keys...).Err(); err != nil { + t.Fatalf("remove Redis provider keys: %v", err) + } + } + cursor = next + if cursor == 0 { + return + } + } +} + +func (fixture redisFixture) coordinator(t *testing.T, holder string) *Adapter { + t.Helper() + adapter, err := New(fixture.client, Options{ + Namespace: fixture.namespace, HolderID: holder, LeaseTTL: 600 * time.Millisecond, + RenewEvery: 150 * time.Millisecond, RetryInterval: 20 * time.Millisecond, PermitGrace: 100 * time.Millisecond, + }) + if err != nil { + t.Fatalf("New(%s): %v", holder, err) + } + return adapter +} + +func receiveLeadership(t *testing.T, values <-chan leadershipFixture) leadershipFixture { + t.Helper() + select { + case value := <-values: + return value + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for leadership") + return leadershipFixture{} + } +} + +type leadershipFixture struct { + holder string + fence controllerProvider.Fence +} + +func receiveSession(t *testing.T, values <-chan controllerProvider.LeaderSession) controllerProvider.LeaderSession { + t.Helper() + select { + case value := <-values: + return value + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for leader session") + return nil + } +} + +func receiveFence(t *testing.T, values <-chan controllerProvider.Fence) controllerProvider.Fence { + t.Helper() + select { + case value := <-values: + return value + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for provider fence") + return controllerProvider.Fence{} + } +} + +func receivePermit(t *testing.T, values <-chan permitResultFixture) permitResultFixture { + t.Helper() + select { + case value := <-values: + return value + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for request permit") + return permitResultFixture{} + } +} + +type permitResultFixture struct { + permit controllerProvider.RequestPermit + err error +} + +func waitRunner(t *testing.T, done <-chan error) { + t.Helper() + select { + case err := <-done: + if err != nil && err != context.Canceled { + t.Fatalf("RunLeader() error = %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for RunLeader shutdown") + } +} diff --git a/internal/adapters/redisprovider/keys.go b/internal/adapters/redisprovider/keys.go new file mode 100644 index 0000000..e9a7da9 --- /dev/null +++ b/internal/adapters/redisprovider/keys.go @@ -0,0 +1,51 @@ +package redisprovider + +import ( + "crypto/sha256" + "encoding/hex" + "strconv" + "strings" + + controllerProvider "proxy-pool/internal/controller/provider" +) + +type keyBuilder struct { + namespace string +} + +type upstreamKeys struct { + generation string + epoch string + leader string + next string + inflight string +} + +func (builder keyBuilder) forUpstream(upstreamID string) (upstreamKeys, error) { + if strings.TrimSpace(upstreamID) != upstreamID || upstreamID == "" { + return upstreamKeys{}, controllerProvider.ErrInvalidCoordination + } + digest := digestParts(upstreamID) + prefix := "pp:" + builder.namespace + ":{provider:" + digest + "}" + return upstreamKeys{ + generation: prefix + ":generation", + epoch: prefix + ":epoch", + leader: prefix + ":leader", + next: prefix + ":next-request", + inflight: prefix + ":inflight", + }, nil +} + +func (keys upstreamKeys) all() []string { + return []string{keys.generation, keys.epoch, keys.leader, keys.next, keys.inflight} +} + +func digestParts(values ...string) string { + digest := sha256.New() + for _, value := range values { + _, _ = digest.Write([]byte(strconv.Itoa(len(value)))) + _, _ = digest.Write([]byte{':'}) + _, _ = digest.Write([]byte(value)) + } + return hex.EncodeToString(digest.Sum(nil)) +} diff --git a/internal/adapters/redisprovider/scripts.go b/internal/adapters/redisprovider/scripts.go new file mode 100644 index 0000000..10c5008 --- /dev/null +++ b/internal/adapters/redisprovider/scripts.go @@ -0,0 +1,61 @@ +package redisprovider + +import ( + "bytes" + "context" + _ "embed" + "encoding/json" + "errors" + "fmt" + "io" + + "github.com/redis/go-redis/v9" + + controllerProvider "proxy-pool/internal/controller/provider" +) + +type scriptReply struct { + Status string `json:"status"` + Generation string `json:"generation,omitempty"` + Epoch uint64 `json:"epoch,string,omitempty"` + WaitMS int64 `json:"waitMs,omitempty"` +} + +//go:embed scripts/provider.lua +var providerSource string + +var providerScript = redis.NewScript(providerSource) + +func runScript(ctx context.Context, client redis.Scripter, keys upstreamKeys, args ...any) (scriptReply, error) { + var reply scriptReply + result, err := providerScript.Run(ctx, client, keys.all(), args...).Result() + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return reply, err + } + return reply, errors.Join(controllerProvider.ErrCoordinationUnavailable, + fmt.Errorf("run Redis provider script: %w", err)) + } + var payload []byte + switch value := result.(type) { + case string: + payload = []byte(value) + case []byte: + payload = value + default: + return reply, errors.Join(controllerProvider.ErrCoordinationUnavailable, + fmt.Errorf("decode Redis provider script: unexpected reply type %T", result)) + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&reply); err != nil || reply.Status == "" { + return scriptReply{}, errors.Join(controllerProvider.ErrCoordinationUnavailable, + fmt.Errorf("decode Redis provider script reply: %w", err)) + } + var trailing any + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { + return scriptReply{}, errors.Join(controllerProvider.ErrCoordinationUnavailable, + errors.New("decode Redis provider script reply: trailing value")) + } + return reply, nil +} diff --git a/internal/adapters/redisprovider/scripts/provider.lua b/internal/adapters/redisprovider/scripts/provider.lua new file mode 100644 index 0000000..cb6813f --- /dev/null +++ b/internal/adapters/redisprovider/scripts/provider.lua @@ -0,0 +1,158 @@ +local operation = ARGV[1] + +local function now_ms() + local value = redis.call('TIME') + return tonumber(value[1]) * 1000 + math.floor(tonumber(value[2]) / 1000) +end + +local function reply(status, generation, epoch, wait_ms) + return cjson.encode({ + status = status, + generation = generation, + epoch = tostring(epoch or '0'), + waitMs = wait_ms + }) +end + +local function read_leader() + local encoded = redis.call('GET', KEYS[3]) + if not encoded then + return nil, nil + end + local ok, value = pcall(cjson.decode, encoded) + if not ok or type(value) ~= 'table' or type(value.generation) ~= 'string' or + type(value.holderId) ~= 'string' or type(value.token) ~= 'string' or + type(value.epoch) ~= 'string' then + return nil, 'invalid' + end + return value, nil +end + +local function same_leader(value, generation, holder_id, token, epoch) + return value and value.generation == generation and value.holderId == holder_id and + value.token == token and value.epoch == tostring(epoch) +end + +if operation == 'acquire_leader' then + local generation_candidate = ARGV[2] + local holder_id = ARGV[3] + local token = ARGV[4] + local lease_ttl = tonumber(ARGV[5]) + if not lease_ttl or lease_ttl <= 0 then + return reply('invalid', '', 0, 0) + end + redis.call('SET', KEYS[1], generation_candidate, 'NX') + local generation = redis.call('GET', KEYS[1]) + local current, current_error = read_leader() + if current_error then + return reply('unavailable', generation, 0, 0) + end + if current then + if current.generation == generation and current.holderId == holder_id and current.token == token then + current.expiresAtMs = now_ms() + lease_ttl + redis.call('SET', KEYS[3], cjson.encode(current), 'PX', lease_ttl) + return reply('ok', generation, current.epoch, 0) + end + local remaining = redis.call('PTTL', KEYS[3]) + return reply('busy', generation, 0, math.max(remaining, 1)) + end + redis.call('INCR', KEYS[2]) + local epoch = redis.call('GET', KEYS[2]) + local leader = { + version = 1, + generation = generation, + holderId = holder_id, + token = token, + epoch = epoch, + expiresAtMs = now_ms() + lease_ttl + } + redis.call('SET', KEYS[3], cjson.encode(leader), 'PX', lease_ttl) + return reply('ok', generation, epoch, 0) +end + +if operation == 'renew_leader' then + local generation = ARGV[2] + local holder_id = ARGV[3] + local token = ARGV[4] + local epoch = ARGV[5] + local lease_ttl = tonumber(ARGV[6]) + local current, current_error = read_leader() + if current_error then + return reply('unavailable', generation, epoch or 0, 0) + end + if not same_leader(current, generation, holder_id, token, epoch) then + return reply('stale', generation, epoch or 0, 0) + end + current.expiresAtMs = now_ms() + lease_ttl + redis.call('SET', KEYS[3], cjson.encode(current), 'PX', lease_ttl) + return reply('ok', generation, epoch, 0) +end + +if operation == 'release_leader' then + local generation = ARGV[2] + local holder_id = ARGV[3] + local token = ARGV[4] + local epoch = ARGV[5] + local current, current_error = read_leader() + if current_error then + return reply('unavailable', generation, epoch or 0, 0) + end + if same_leader(current, generation, holder_id, token, epoch) then + redis.call('DEL', KEYS[3]) + end + return reply('ok', generation, epoch or 0, 0) +end + +if operation == 'acquire_fetch' then + local generation = ARGV[2] + local holder_id = ARGV[3] + local leader_token = ARGV[4] + local epoch = ARGV[5] + local permit_token = ARGV[6] + local request_interval = tonumber(ARGV[7]) + local max_in_flight = tonumber(ARGV[8]) + local permit_ttl = tonumber(ARGV[9]) + local current, current_error = read_leader() + if current_error then + return reply('unavailable', generation, epoch or 0, 0) + end + if not same_leader(current, generation, holder_id, leader_token, epoch) then + return reply('stale', generation, epoch or 0, 0) + end + local now = now_ms() + redis.call('ZREMRANGEBYSCORE', KEYS[5], '-inf', now) + local existing = redis.call('ZSCORE', KEYS[5], permit_token) + if existing then + return reply('ok', generation, epoch, 0) + end + local next_request = redis.call('GET', KEYS[4]) + if next_request and not tonumber(next_request) then + return reply('unavailable', generation, epoch, 0) + end + if next_request and tonumber(next_request) > now then + return reply('rate_limited', generation, epoch, tonumber(next_request) - now) + end + if redis.call('ZCARD', KEYS[5]) >= max_in_flight then + local earliest = redis.call('ZRANGE', KEYS[5], 0, 0, 'WITHSCORES') + local wait_ms = 1 + if earliest[2] then + wait_ms = math.max(tonumber(earliest[2]) - now, 1) + end + return reply('at_capacity', generation, epoch, wait_ms) + end + redis.call('ZADD', KEYS[5], now + permit_ttl, permit_token) + redis.call('PEXPIRE', KEYS[5], permit_ttl + 1000) + if request_interval > 0 then + redis.call('SET', KEYS[4], now + request_interval, 'PX', request_interval) + else + redis.call('DEL', KEYS[4]) + end + return reply('ok', generation, epoch, 0) +end + +if operation == 'release_fetch' then + redis.call('ZREM', KEYS[5], ARGV[2]) + return reply('ok', '', 0, 0) +end + +return reply('invalid', '', 0, 0) diff --git a/internal/config/config.go b/internal/config/config.go index b04c7cb..0c080d1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -163,6 +163,7 @@ type Upstream struct { ProxyAuth ProxyAuth `yaml:"proxyAuth"` Pool Pool `yaml:"pool"` Capacity Capacity `yaml:"capacity"` + Refill Refill `yaml:"refill"` Lifecycle Lifecycle `yaml:"lifecycle"` Fetch Fetch `yaml:"fetch"` Check Check `yaml:"check"` @@ -217,20 +218,27 @@ type Capacity struct { MaxConcurrencyPerProxy int `yaml:"maxConcurrencyPerProxy"` } +type Refill struct { + ReconcileInterval Duration `yaml:"reconcileInterval"` + MinimumAvailableSlots int64 `yaml:"minimumAvailableSlots"` + TargetAvailableSlots int64 `yaml:"targetAvailableSlots"` +} + type Lifecycle struct { TTL Duration `yaml:"ttl"` AllocationSafetyMargin Duration `yaml:"allocationSafetyMargin"` } type Fetch struct { - RequestInterval Duration `yaml:"requestInterval"` - Timeout Duration `yaml:"timeout"` - MaxAttempts int `yaml:"maxAttempts"` - MaxInFlight int `yaml:"maxInFlight"` - MaxTotal int `yaml:"maxTotal"` - MaxResponseBytes int64 `yaml:"maxResponseBytes"` - TemplateTimeout Duration `yaml:"templateTimeout"` - Retry Backoff `yaml:"retry"` + EstimatedIPsPerCall int `yaml:"estimatedIPsPerCall"` + RequestInterval Duration `yaml:"requestInterval"` + Timeout Duration `yaml:"timeout"` + MaxAttempts int `yaml:"maxAttempts"` + MaxInFlight int `yaml:"maxInFlight"` + MaxTotal int `yaml:"maxTotal"` + MaxResponseBytes int64 `yaml:"maxResponseBytes"` + TemplateTimeout Duration `yaml:"templateTimeout"` + Retry Backoff `yaml:"retry"` } type Backoff struct { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 031d831..b8e4ebc 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -61,10 +61,15 @@ upstreams: maxSize: 100 capacity: maxConcurrencyPerProxy: 10 + refill: + reconcileInterval: 1s + minimumAvailableSlots: 200 + targetAvailableSlots: 500 lifecycle: ttl: 120s allocationSafetyMargin: 10s fetch: + estimatedIPsPerCall: 20 requestInterval: 1s timeout: 3s maxAttempts: 5 @@ -86,7 +91,10 @@ func TestLoadStrictValidConfiguration(t *testing.T) { if err != nil { t.Fatalf("Load(): %v", err) } - if cfg.Version != 1 || cfg.Upstreams["provider-a"].Pool.MaxSize != 100 { + provider := cfg.Upstreams["provider-a"] + if cfg.Version != 1 || provider.Pool.MaxSize != 100 || provider.Fetch.EstimatedIPsPerCall != 20 || + provider.Refill.ReconcileInterval.Value() != time.Second || provider.Refill.MinimumAvailableSlots != 200 || + provider.Refill.TargetAvailableSlots != 500 { t.Fatalf("unexpected config: %+v", cfg) } } @@ -529,6 +537,47 @@ func TestValidateRejectsInvalidConfigurationMatrix(t *testing.T) { }, want: "fetch.maxInFlight", }, + { + name: "zero estimated IPs per call", + mutate: func(cfg *Config) { + updateUpstream(cfg, func(upstream *Upstream) { upstream.Fetch.EstimatedIPsPerCall = 0 }) + }, + want: "estimatedIPsPerCall", + }, + { + name: "estimated IPs exceed pool size", + mutate: func(cfg *Config) { + updateUpstream(cfg, func(upstream *Upstream) { + upstream.Fetch.EstimatedIPsPerCall = upstream.Pool.MaxSize + 1 + }) + }, + want: "estimatedIPsPerCall", + }, + { + name: "zero refill interval", + mutate: func(cfg *Config) { + updateUpstream(cfg, func(upstream *Upstream) { upstream.Refill.ReconcileInterval = 0 }) + }, + want: "refill.reconcileInterval", + }, + { + name: "refill target does not exceed minimum", + mutate: func(cfg *Config) { + updateUpstream(cfg, func(upstream *Upstream) { + upstream.Refill.TargetAvailableSlots = upstream.Refill.MinimumAvailableSlots + }) + }, + want: "targetAvailableSlots", + }, + { + name: "refill target exceeds theoretical capacity", + mutate: func(cfg *Config) { + updateUpstream(cfg, func(upstream *Upstream) { + upstream.Refill.TargetAvailableSlots = int64(upstream.Pool.MaxSize)*int64(upstream.Capacity.MaxConcurrencyPerProxy) + 1 + }) + }, + want: "targetAvailableSlots", + }, { name: "negative fetch response limit", mutate: func(cfg *Config) { diff --git a/internal/config/store_test.go b/internal/config/store_test.go index 45eed28..fe75ac2 100644 --- a/internal/config/store_test.go +++ b/internal/config/store_test.go @@ -4,6 +4,7 @@ import ( "errors" "sync" "testing" + "time" ) func TestStorePublishesAndReturnsDetachedConfigurations(t *testing.T) { @@ -77,8 +78,13 @@ func storeTestConfig(upstreamName string) *Config { API: ProviderAPI{Auth: ProviderAuth{Type: "none"}}, ProxyAuth: ProxyAuth{Type: "response"}, Pool: Pool{MaxSize: 10}, Capacity: Capacity{MaxConcurrencyPerProxy: 1}, + Refill: Refill{ + ReconcileInterval: Duration(time.Second), MinimumAvailableSlots: 1, TargetAvailableSlots: 2, + }, Lifecycle: Lifecycle{TTL: Duration(60_000_000_000), AllocationSafetyMargin: Duration(10_000_000_000)}, - Fetch: Fetch{Timeout: Duration(1_000_000_000), MaxAttempts: 1, MaxInFlight: 1}, + Fetch: Fetch{ + EstimatedIPsPerCall: 1, Timeout: Duration(time.Second), MaxAttempts: 1, MaxInFlight: 1, + }, }, }, Routing: []Routing{{ diff --git a/internal/config/validate.go b/internal/config/validate.go index ab5df72..964afae 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -2,6 +2,7 @@ package config import ( "fmt" + "math" "net" "net/url" "regexp" @@ -338,6 +339,22 @@ func validateUpstream(name string, upstream Upstream) error { if err := requirePositive(scope+" capacity.maxConcurrencyPerProxy", upstream.Capacity.MaxConcurrencyPerProxy); err != nil { return err } + if err := requirePositive(scope+" refill.reconcileInterval", upstream.Refill.ReconcileInterval); err != nil { + return err + } + if err := requirePositive(scope+" refill.minimumAvailableSlots", upstream.Refill.MinimumAvailableSlots); err != nil { + return err + } + if upstream.Refill.TargetAvailableSlots <= upstream.Refill.MinimumAvailableSlots { + return fmt.Errorf("validate %s refill.targetAvailableSlots: must be greater than minimumAvailableSlots", scope) + } + if int64(upstream.Pool.MaxSize) > math.MaxInt64/int64(upstream.Capacity.MaxConcurrencyPerProxy) { + return fmt.Errorf("validate %s refill.targetAvailableSlots: theoretical capacity overflows int64", scope) + } + theoreticalSlots := int64(upstream.Pool.MaxSize) * int64(upstream.Capacity.MaxConcurrencyPerProxy) + if upstream.Refill.TargetAvailableSlots > theoreticalSlots { + return fmt.Errorf("validate %s refill.targetAvailableSlots: exceeds theoretical capacity", scope) + } if err := requirePositive(scope+" lifecycle.ttl", upstream.Lifecycle.TTL); err != nil { return err } @@ -350,6 +367,12 @@ func validateUpstream(name string, upstream Upstream) error { if err := validateFetch(scope+" fetch", upstream.Fetch); err != nil { return err } + if upstream.Fetch.EstimatedIPsPerCall > upstream.Pool.MaxSize { + return fmt.Errorf("validate %s fetch.estimatedIPsPerCall: cannot exceed pool.maxSize", scope) + } + if upstream.Fetch.MaxTotal > 0 && upstream.Fetch.EstimatedIPsPerCall > upstream.Fetch.MaxTotal { + return fmt.Errorf("validate %s fetch.estimatedIPsPerCall: cannot exceed fetch.maxTotal", scope) + } if err := validateCheck(scope+" check", upstream.Check); err != nil { return err } @@ -377,6 +400,9 @@ func validateUpstream(name string, upstream Upstream) error { } func validateFetch(scope string, fetch Fetch) error { + if err := requirePositive(scope+".estimatedIPsPerCall", fetch.EstimatedIPsPerCall); err != nil { + return err + } if err := requireNonNegative(scope+".requestInterval", fetch.RequestInterval); err != nil { return err } @@ -442,7 +468,7 @@ func validateCheck(scope string, check Check) error { } func fetchConfigured(fetch Fetch) bool { - return fetch.RequestInterval != 0 || fetch.Timeout != 0 || fetch.MaxAttempts != 0 || + return fetch.EstimatedIPsPerCall != 0 || fetch.RequestInterval != 0 || fetch.Timeout != 0 || fetch.MaxAttempts != 0 || fetch.MaxInFlight != 0 || fetch.MaxTotal != 0 || fetch.MaxResponseBytes != 0 || fetch.TemplateTimeout != 0 || fetch.Retry.Initial != 0 || fetch.Retry.Max != 0 || fetch.Retry.Jitter != 0 diff --git a/internal/controller/admin/configuration_test.go b/internal/controller/admin/configuration_test.go index e104e95..241fc8a 100644 --- a/internal/controller/admin/configuration_test.go +++ b/internal/controller/admin/configuration_test.go @@ -150,7 +150,8 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 2000} capacity: {maxConcurrencyPerProxy: 20} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 100, targetAvailableSlots: 200} lifecycle: {ttl: 5m, allocationSafetyMargin: 30s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + fetch: {estimatedIPsPerCall: 100, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} ` diff --git a/internal/controller/admin/service_test.go b/internal/controller/admin/service_test.go index 94ee622..f4f116d 100644 --- a/internal/controller/admin/service_test.go +++ b/internal/controller/admin/service_test.go @@ -473,8 +473,13 @@ func validReloadUpstream(secret string) config.Upstream { API: config.ProviderAPI{Auth: config.ProviderAuth{Type: "none"}}, ProxyAuth: config.ProxyAuth{Type: "static", Username: "user", Password: secret}, Pool: config.Pool{MaxSize: 10}, Capacity: config.Capacity{MaxConcurrencyPerProxy: 2}, + Refill: config.Refill{ + ReconcileInterval: config.Duration(time.Second), MinimumAvailableSlots: 1, TargetAvailableSlots: 2, + }, Lifecycle: config.Lifecycle{TTL: config.Duration(time.Minute), AllocationSafetyMargin: config.Duration(10 * time.Second)}, - Fetch: config.Fetch{Timeout: config.Duration(time.Second), MaxAttempts: 2, MaxInFlight: 1}, + Fetch: config.Fetch{ + EstimatedIPsPerCall: 1, Timeout: config.Duration(time.Second), MaxAttempts: 2, MaxInFlight: 1, + }, } } diff --git a/internal/controller/bootstrap/bootstrap_test.go b/internal/controller/bootstrap/bootstrap_test.go index 4cabe30..440f62c 100644 --- a/internal/controller/bootstrap/bootstrap_test.go +++ b/internal/controller/bootstrap/bootstrap_test.go @@ -196,8 +196,9 @@ upstreams: proxyAuth: {type: response} pool: {maxSize: 100} capacity: {maxConcurrencyPerProxy: 10} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 100, targetAvailableSlots: 200} lifecycle: {ttl: 2m, allocationSafetyMargin: 10s} - fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 1000} + fetch: {estimatedIPsPerCall: 10, requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 1000} check: interval: 30s jitter: 20 diff --git a/internal/controller/pool/fetch_budget.go b/internal/controller/pool/fetch_budget.go index 7626594..619c707 100644 --- a/internal/controller/pool/fetch_budget.go +++ b/internal/controller/pool/fetch_budget.go @@ -9,10 +9,12 @@ import ( ) var ( - ErrInvalidFetchBudget = errors.New("invalid fetch budget") - ErrInvalidFetchCompletion = errors.New("invalid fetch completion") - ErrFetchPermitFinished = errors.New("fetch permit is already finished") - ErrInvalidManagedRelease = errors.New("invalid managed proxy release") + ErrInvalidFetchBudget = errors.New("invalid fetch budget") + ErrInvalidFetchCompletion = errors.New("invalid fetch completion") + ErrFetchPermitFinished = errors.New("fetch permit is already finished") + ErrInvalidManagedRelease = errors.New("invalid managed proxy release") + ErrInvalidManagedSynchronization = errors.New("invalid managed proxy synchronization") + ErrManagedSynchronizationInFlight = errors.New("managed proxy synchronization has pending fetches") ) type FetchBudgetConfig struct { @@ -112,6 +114,22 @@ func (b *FetchBudget) Snapshot() FetchBudgetSnapshot { return b.usage } +// SynchronizeManaged replaces the local current-inventory count with the +// authoritative activity-store observation. Pending requests and cumulative +// fetch usage remain owned by this budget. +func (b *FetchBudget) SynchronizeManaged(managed int) error { + if b == nil || managed < 0 { + return ErrInvalidManagedSynchronization + } + b.mu.Lock() + defer b.mu.Unlock() + if b.usage.PendingExpected > 0 { + return ErrManagedSynchronizationInFlight + } + b.usage.Managed = managed + return nil +} + // ReleaseManaged returns current-inventory capacity after extraction, expiry, // or removal. It deliberately does not restore the cumulative fetch quota. func (b *FetchBudget) ReleaseManaged(count int) error { diff --git a/internal/controller/pool/fetch_budget_test.go b/internal/controller/pool/fetch_budget_test.go index c6ddb9b..493267d 100644 --- a/internal/controller/pool/fetch_budget_test.go +++ b/internal/controller/pool/fetch_budget_test.go @@ -61,6 +61,48 @@ func TestFetchBudgetRejectsManagedCounterUnderflow(t *testing.T) { } } +func TestFetchBudgetSynchronizesAuthoritativeManagedInventory(t *testing.T) { + budget, err := NewFetchBudget(FetchBudgetConfig{ + UpstreamID: "a", MaxSize: 5, ExpectedPerFetch: 2, Managed: 3, FetchedTotal: 7, + }) + if err != nil { + t.Fatalf("NewFetchBudget(): %v", err) + } + permit, ok, err := budget.ReserveFetch("a") + if err != nil || !ok { + t.Fatalf("ReserveFetch() = (_, %v, %v), want permit", ok, err) + } + + if err := budget.SynchronizeManaged(1); !errors.Is(err, ErrManagedSynchronizationInFlight) { + t.Fatalf("SynchronizeManaged() error = %v, want ErrManagedSynchronizationInFlight", err) + } + usage := budget.Snapshot() + if usage.Managed != 3 || usage.PendingExpected != 2 || usage.FetchedTotal != 7 { + t.Fatalf("Snapshot() = %+v, want managed=3 pending=2 fetched=7", usage) + } + if err := permit.Cancel(); err != nil { + t.Fatalf("Cancel(): %v", err) + } + if err := budget.SynchronizeManaged(1); err != nil { + t.Fatalf("SynchronizeManaged() after cancel: %v", err) + } + if usage := budget.Snapshot(); usage.Managed != 1 || usage.PendingExpected != 0 || usage.FetchedTotal != 7 { + t.Fatalf("Snapshot() after synchronization = %+v, want managed=1 pending=0 fetched=7", usage) + } +} + +func TestFetchBudgetRejectsNegativeManagedSynchronization(t *testing.T) { + budget, err := NewFetchBudget(FetchBudgetConfig{ + UpstreamID: "a", MaxSize: 10, ExpectedPerFetch: 1, + }) + if err != nil { + t.Fatalf("NewFetchBudget(): %v", err) + } + if err := budget.SynchronizeManaged(-1); !errors.Is(err, ErrInvalidManagedSynchronization) { + t.Fatalf("SynchronizeManaged() error = %v, want ErrInvalidManagedSynchronization", err) + } +} + func TestFetchBudgetRequiresWholeExpectedBatchToFitLimits(t *testing.T) { tests := []struct { name string diff --git a/internal/controller/pool/reconciler.go b/internal/controller/pool/reconciler.go index 7108bba..cc73821 100644 --- a/internal/controller/pool/reconciler.go +++ b/internal/controller/pool/reconciler.go @@ -2,6 +2,8 @@ package pool import ( "errors" + "math" + "sync" "time" "proxy-pool/internal/domain/upstream" @@ -11,7 +13,9 @@ var ErrInvalidReconcilePolicy = errors.New("invalid pool reconcile policy") type ReconcilePolicy struct { MinimumAvailableSlots int64 + TargetAvailableSlots int64 ExpectedPerFetch int + ExpectedSlotsPerFetch int64 SafetyMargin time.Duration } @@ -24,41 +28,83 @@ type ReconcileDecision struct { PendingExpected int FetchedTotal int64 FetchAllowance int + EffectiveSlots int64 Triggered bool } type Reconciler struct { - policy ReconcilePolicy - budget *FetchBudget - notifier FetchNotifier + policy ReconcilePolicy + budget *FetchBudget + notifier FetchNotifier + mu sync.Mutex + refilling bool + slotsPerProxy int64 } func NewReconciler(policy ReconcilePolicy, budget *FetchBudget, notifier FetchNotifier) (*Reconciler, error) { - if policy.MinimumAvailableSlots <= 0 || policy.ExpectedPerFetch <= 0 || + if policy.MinimumAvailableSlots <= 0 || policy.TargetAvailableSlots <= policy.MinimumAvailableSlots || + policy.ExpectedPerFetch <= 0 || policy.ExpectedSlotsPerFetch <= 0 || policy.SafetyMargin < 0 || budget == nil || notifier == nil { return nil, ErrInvalidReconcilePolicy } if budget.expected != policy.ExpectedPerFetch { return nil, ErrInvalidReconcilePolicy } - return &Reconciler{policy: policy, budget: budget, notifier: notifier}, nil + if policy.ExpectedSlotsPerFetch%int64(policy.ExpectedPerFetch) != 0 { + return nil, ErrInvalidReconcilePolicy + } + return &Reconciler{ + policy: policy, budget: budget, notifier: notifier, + slotsPerProxy: policy.ExpectedSlotsPerFetch / int64(policy.ExpectedPerFetch), + }, nil } // Reconcile centralizes the cold-path decision. The notifier may coalesce many // calls; Provider Reconciler atomically reserves the budget before doing I/O. func (r *Reconciler) Reconcile(now time.Time, inventory upstream.Inventory) ReconcileDecision { usage := r.budget.Snapshot() + availableSlots := inventory.AvailableSlots(now, r.policy.SafetyMargin) + pendingSlots := saturatingMultiply(int64(usage.PendingExpected), r.slotsPerProxy) decision := ReconcileDecision{ - AvailableSlots: inventory.AvailableSlots(now, r.policy.SafetyMargin), + AvailableSlots: availableSlots, PendingExpected: usage.PendingExpected, FetchedTotal: usage.FetchedTotal, FetchAllowance: r.budget.FetchAllowance(), + EffectiveSlots: saturatingAdd(availableSlots, pendingSlots), } - if decision.AvailableSlots >= r.policy.MinimumAvailableSlots || - decision.FetchAllowance < r.policy.ExpectedPerFetch { + r.mu.Lock() + if r.refilling { + if usage.PendingExpected == 0 && decision.AvailableSlots >= r.policy.TargetAvailableSlots { + r.refilling = false + } + } else if decision.AvailableSlots < r.policy.MinimumAvailableSlots { + r.refilling = true + } + trigger := r.refilling && + decision.EffectiveSlots < r.policy.TargetAvailableSlots && + decision.FetchAllowance >= r.policy.ExpectedPerFetch + r.mu.Unlock() + if !trigger { return decision } r.notifier.Notify() decision.Triggered = true return decision } + +func saturatingMultiply(left, right int64) int64 { + if left <= 0 || right <= 0 { + return 0 + } + if left > math.MaxInt64/right { + return math.MaxInt64 + } + return left * right +} + +func saturatingAdd(left, right int64) int64 { + if left >= math.MaxInt64-right { + return math.MaxInt64 + } + return left + right +} diff --git a/internal/controller/pool/reconciler_test.go b/internal/controller/pool/reconciler_test.go index ca2c6bc..2dd71e9 100644 --- a/internal/controller/pool/reconciler_test.go +++ b/internal/controller/pool/reconciler_test.go @@ -18,7 +18,9 @@ func TestPoolReconcilerSignalsOnlyWhenSlotsAreLowAndWholeFetchFits(t *testing.T) notifier := &recordingFetchNotifier{} reconciler, err := NewReconciler(ReconcilePolicy{ MinimumAvailableSlots: 5, + TargetAvailableSlots: 8, ExpectedPerFetch: 2, + ExpectedSlotsPerFetch: 2, SafetyMargin: 10 * time.Second, }, budget, notifier) if err != nil { @@ -56,7 +58,8 @@ func TestPoolReconcilerUsesBudgetPendingAndCumulativeCounters(t *testing.T) { defer permit.Cancel() notifier := &recordingFetchNotifier{} reconciler, err := NewReconciler(ReconcilePolicy{ - MinimumAvailableSlots: 1, ExpectedPerFetch: 2, + MinimumAvailableSlots: 1, TargetAvailableSlots: 2, + ExpectedPerFetch: 2, ExpectedSlotsPerFetch: 2, }, budget, notifier) if err != nil { t.Fatalf("NewReconciler(): %v", err) @@ -71,6 +74,79 @@ func TestPoolReconcilerUsesBudgetPendingAndCumulativeCounters(t *testing.T) { } } +func TestPoolReconcilerUsesTargetWatermarkUntilRefillCompletes(t *testing.T) { + budget, err := NewFetchBudget(FetchBudgetConfig{ + UpstreamID: "provider-a", MaxSize: 10, ExpectedPerFetch: 2, + }) + if err != nil { + t.Fatalf("NewFetchBudget(): %v", err) + } + notifier := &recordingFetchNotifier{} + reconciler, err := NewReconciler(ReconcilePolicy{ + MinimumAvailableSlots: 3, + TargetAvailableSlots: 8, + ExpectedPerFetch: 2, + ExpectedSlotsPerFetch: 2, + }, budget, notifier) + if err != nil { + t.Fatalf("NewReconciler(): %v", err) + } + now := time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC) + + belowMinimum := reconciler.Reconcile(now, inventoryWithSlots(now, 2)) + betweenWatermarks := reconciler.Reconcile(now, inventoryWithSlots(now, 5)) + atTarget := reconciler.Reconcile(now, inventoryWithSlots(now, 8)) + aboveMinimumAfterCompletion := reconciler.Reconcile(now, inventoryWithSlots(now, 5)) + + if !belowMinimum.Triggered || !betweenWatermarks.Triggered || atTarget.Triggered || aboveMinimumAfterCompletion.Triggered { + t.Fatalf("triggered states = [%v %v %v %v], want [true true false false]", + belowMinimum.Triggered, betweenWatermarks.Triggered, atTarget.Triggered, aboveMinimumAfterCompletion.Triggered) + } + if notifier.calls != 2 { + t.Fatalf("Notify() calls = %d, want 2", notifier.calls) + } +} + +func TestPoolReconcilerPendingEstimatePausesWithoutEndingRefillEpisode(t *testing.T) { + budget, err := NewFetchBudget(FetchBudgetConfig{ + UpstreamID: "provider-a", MaxSize: 10, ExpectedPerFetch: 2, + }) + if err != nil { + t.Fatalf("NewFetchBudget(): %v", err) + } + notifier := &recordingFetchNotifier{} + reconciler, err := NewReconciler(ReconcilePolicy{ + MinimumAvailableSlots: 3, TargetAvailableSlots: 8, + ExpectedPerFetch: 2, ExpectedSlotsPerFetch: 6, + }, budget, notifier) + if err != nil { + t.Fatalf("NewReconciler(): %v", err) + } + now := time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC) + if decision := reconciler.Reconcile(now, inventoryWithSlots(now, 2)); !decision.Triggered { + t.Fatalf("initial Reconcile() = %+v, want trigger", decision) + } + permit, ok, err := budget.ReserveFetch("provider-a") + if err != nil || !ok { + t.Fatalf("ReserveFetch() = (_, %v, %v), want permit", ok, err) + } + if decision := reconciler.Reconcile(now, inventoryWithSlots(now, 2)); decision.Triggered || decision.EffectiveSlots != 8 { + t.Fatalf("pending Reconcile() = %+v, want paused at target estimate", decision) + } + if err := permit.Complete(2, 1); err != nil { + t.Fatalf("Complete(): %v", err) + } + if decision := reconciler.Reconcile(now, inventoryWithSlots(now, 5)); !decision.Triggered { + t.Fatalf("post-fetch Reconcile() = %+v, want refill episode to continue", decision) + } +} + +func inventoryWithSlots(now time.Time, slots int64) upstream.Inventory { + return upstream.Inventory{Proxies: []upstream.ProxyCapacity{{ + State: proxyDomain.StateAvailable, ExpiresAt: now.Add(time.Minute), Max: slots, + }}} +} + type recordingFetchNotifier struct{ calls int } func (n *recordingFetchNotifier) Notify() { n.calls++ } diff --git a/internal/controller/provider/coordination.go b/internal/controller/provider/coordination.go new file mode 100644 index 0000000..a1f80b6 --- /dev/null +++ b/internal/controller/provider/coordination.go @@ -0,0 +1,41 @@ +package provider + +import ( + "context" + "errors" + "time" +) + +var ( + ErrInvalidCoordination = errors.New("invalid provider coordination") + ErrCoordinationUnavailable = errors.New("provider coordination unavailable") + ErrLeadershipLost = errors.New("provider leadership lost") + ErrLeaderWorkStopped = errors.New("provider leader work stopped") +) + +type CoordinationLimits struct { + RequestInterval time.Duration + MaxInFlight int + MaxAttemptDuration time.Duration +} + +type Fence struct { + Generation string + Epoch uint64 +} + +type Coordinator interface { + // RunLeader waits for leadership and runs work only while its fencing lease + // is valid. The work context is canceled before a known lease expiry. + RunLeader(context.Context, string, CoordinationLimits, func(context.Context, LeaderSession) error) error +} + +type LeaderSession interface { + Fence() Fence + AcquireFetch(context.Context) (RequestPermit, error) +} + +type RequestPermit interface { + // Release is idempotent. A failed release expires automatically in storage. + Release(context.Context) error +} diff --git a/scripts/test-redis.ps1 b/scripts/test-redis.ps1 index 74f5b2a..a258cb4 100644 --- a/scripts/test-redis.ps1 +++ b/scripts/test-redis.ps1 @@ -14,7 +14,7 @@ try { $env:PROXY_POOL_TEST_REDIS_URL = "redis://127.0.0.1:16379/15" Push-Location $repositoryRoot try { - go test -count=1 -tags=integration -timeout 60s ./internal/adapters/redisactivity/... + go test -count=1 -tags=integration -timeout 60s ./internal/adapters/redisactivity/... ./internal/adapters/redisprovider/... if ($LASTEXITCODE -ne 0) { throw "Redis integration tests failed with exit code $LASTEXITCODE" }