feat: add Redis provider coordination

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

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

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

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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

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}

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

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

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

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")
+	}
+}

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

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

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)

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,12 +218,19 @@ 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 {
+	EstimatedIPsPerCall int      `yaml:"estimatedIPsPerCall"`
 	RequestInterval     Duration `yaml:"requestInterval"`
 	Timeout             Duration `yaml:"timeout"`
 	MaxAttempts         int      `yaml:"maxAttempts"`

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

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

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

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

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,
+		},
 	}
 }
 

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

internal/controller/pool/fetch_budget.go

@@ -13,6 +13,8 @@ var (
 	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 {

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

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,6 +28,7 @@ type ReconcileDecision struct {
 	PendingExpected int
 	FetchedTotal    int64
 	FetchAllowance  int
+	EffectiveSlots  int64
 	Triggered       bool
 }
 
@@ -31,34 +36,75 @@ type Reconciler struct {
 	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 ||
+	r.mu.Lock()
-		decision.FetchAllowance < r.policy.ExpectedPerFetch {
+	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
+}

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

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

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