Tune redis proxy connection capacity

adapter/redis_peer_limiter_test.go

@@ -86,7 +86,7 @@ func TestRedisLeaderClientPoolStaysBelowPeerLimit(t *testing.T) {
 }
 
 func TestRedisLeaderClientPoolUsesSmallDefault(t *testing.T) {
-	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(8))
+	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(defaultRedisPerPeerConnectionCap))
 	client := server.getOrCreateLeaderClient("127.0.0.1:6379")
 	defer client.Close()
 
@@ -103,8 +103,8 @@ func TestRedisLeaderClientPoolsSharePeerBudget(t *testing.T) {
 	}{
 		{name: "low cap", limit: 2, wantNormal: 1, wantBlocking: 1},
 		{name: "four cap", limit: 4, wantNormal: 2, wantBlocking: 2},
-		{name: "default cap", limit: 8, wantNormal: 4, wantBlocking: 4},
-		{name: "raised cap", limit: defaultRedisPerPeerConnectionCap, wantNormal: 4, wantBlocking: 4},
+		{name: "legacy cap", limit: 8, wantNormal: defaultRedisLeaderClientPoolSize, wantBlocking: defaultRedisBlockingLeaderClientPoolSize},
+		{name: "default cap", limit: defaultRedisPerPeerConnectionCap, wantNormal: defaultRedisLeaderClientPoolSize, wantBlocking: defaultRedisBlockingLeaderClientPoolSize},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(tc.limit))
@@ -116,7 +116,7 @@ func TestRedisLeaderClientPoolsSharePeerBudget(t *testing.T) {
 }
 
 func TestRedisBlockingLeaderClientUsesDedicatedBudgetedPool(t *testing.T) {
-	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(8))
+	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(defaultRedisPerPeerConnectionCap))
 	shared := server.getOrCreateLeaderClient("127.0.0.1:6379")
 	defer shared.Close()
 	blocking := server.getOrCreateBlockingLeaderClient("127.0.0.1:6379")

docs/design/2026_04_24_implemented_workload_isolation.md

@@ -26,9 +26,11 @@ Implementation status:
   `adapter/redis_peer_limiter.go`, wired through `RedisServer.Run` accept and
   close hooks. Default cap is 8 per peer IP and is configurable via
   `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` /
-  `WithRedisPerPeerConnectionLimit`. Redis leader-proxy clients use a small
-  explicit go-redis pool below that default cap, and Pub/Sub detached sockets
-  stay counted until their Pub/Sub cleanup path closes.
+  `WithRedisPerPeerConnectionLimit`. The redis-proxy deployment can raise the
+  cap explicitly on ElasticKV nodes before increasing the proxy's ElasticKV
+  pool size. Redis leader-proxy clients use a small explicit go-redis pool
+  below that default cap, and Pub/Sub detached sockets stay counted until their
+  Pub/Sub cleanup path closes.
 - Shipped: Layer 4 stream entry-per-key layout in `store/stream_helpers.go`,
   `adapter/redis_stream_cmds.go`, and `adapter/redis_compat_helpers.go`.
   XREAD now range-scans `!stream|entry|...` after the requested ID instead of

docs/redis-proxy-deployment.md

@@ -414,7 +414,7 @@ groups:
 | Parameter | Value | Description |
 |-----------|-------|-------------|
 | Redis connection pool size | 128 | Default go-redis pool size for Redis |
-| ElasticKV connection pool size | 64 | Default per-leader command pool; leave server per-peer headroom for dedicated PubSub connections |
+| ElasticKV connection pool size | 64 | Default per-leader command pool; regular and blocking replay pools share the server per-peer budget, while PubSub uses dedicated sockets outside the pool |
 | Dial timeout | 5s | Backend connection timeout |
 | Read timeout | 3s | Backend read timeout |
 | Write timeout | 3s | Backend write timeout |
@@ -439,7 +439,7 @@ Recommended shutdown order: `redis-proxy -> application -> Redis / ElasticKV`.
 ### Secondary writes are falling behind
 - Check `proxy_async_queue_depth`, `proxy_async_queue_delay_seconds`, and `proxy_async_drops_by_queue_total` before increasing concurrency.
 - Check `proxy_backend_pool_pending_requests` and the `waits`/`timeouts` pool events. Pool waits mean concurrency is too high for the configured pool.
-- Keep `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` above `-elastickv-pool-size`; PubSub and shadow PubSub use dedicated connections outside the command pool. Keep `-secondary-write-concurrency` at or below the pool size.
+- Keep `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` above `-elastickv-pool-size`; PubSub and shadow PubSub use dedicated connections outside the command pool and detached PubSub sockets may remain counted until cleanup. Keep `-secondary-write-concurrency` at or below the pool size.
 - A sustained `expired` rate means secondary throughput is below ingress. Increasing queue size only delays the loss; profile ElasticKV before raising concurrency.
 
 ### High divergence count

proxy/backend.go

@@ -72,11 +72,12 @@ func DefaultBackendOptions() BackendOptions {
 }
 
 // DefaultElasticKVBackendOptions returns defaults for proxy backends that
-// connect to ElasticKV's Redis adapter. Production dual-write deployments
-// should run the cluster with ELASTICKV_REDIS_PER_PEER_CONNECTIONS above this
-// pool size because PubSub and shadow PubSub use dedicated connections outside
-// the go-redis command pool. Lower the proxy pool instead for clusters that
-// keep the server-side per-peer cap below the default.
+// connect to ElasticKV's Redis adapter. The regular command pool and blocking
+// replay pool share ElasticKV's per-peer connection budget. PubSub and shadow
+// PubSub use dedicated sockets outside those pools, and detached sockets may
+// remain counted until cleanup. Run the cluster with
+// ELASTICKV_REDIS_PER_PEER_CONNECTIONS above this pool size before using the
+// default in production, or lower the proxy pool to fit the configured cap.
 func DefaultElasticKVBackendOptions() BackendOptions {
 	opts := DefaultBackendOptions()
 	opts.PoolSize = defaultElasticKVPoolSize

proxy/leader_aware_backend.go

@@ -59,6 +59,7 @@ type LeaderAwareRedisBackend struct {
 	refreshMu     sync.Mutex
 	refreshDone   chan struct{}
 	refreshClosed bool
+	lastRefreshAt time.Time
 	refreshCtx    context.Context
 	refreshCancel context.CancelFunc
 
@@ -75,8 +76,8 @@ type LeaderAwareRedisBackend struct {
 }
 
 // NewLeaderAwareRedisBackend creates a LeaderAwareRedisBackend with the given
-// seed addresses. The first seed is used as the initial target until the
-// first refresh completes. At least one seed is required.
+// seed addresses. The first command waits for leader discovery instead of
+// sending traffic to a seed that may be down. At least one seed is required.
 func NewLeaderAwareRedisBackend(seeds []string, name string, opts BackendOptions, logger *slog.Logger) *LeaderAwareRedisBackend {
 	return NewLeaderAwareRedisBackendWithInterval(seeds, name, opts, defaultLeaderRefreshInterval, defaultLeaderRefreshTimeout, logger)
 }
@@ -106,7 +107,7 @@ func NewLeaderAwareRedisBackendWithInterval(seeds []string, name string, opts Ba
 		clients:         make(map[string]*redis.Client, len(normalized)),
 		clientOrder:     make([]string, 0, len(normalized)),
 		seedProtect:     seedProtect,
-		leader:          normalized[0],
+		leader:          "",
 		stopCh:          make(chan struct{}),
 		done:            make(chan struct{}),
 		refreshCh:       make(chan struct{}, 1),
@@ -209,6 +210,7 @@ func (b *LeaderAwareRedisBackend) runLeaderRefresh(done chan struct{}) {
 
 	b.refreshMu.Lock()
 	b.refreshDone = nil
+	b.lastRefreshAt = time.Now()
 	close(done)
 	b.refreshMu.Unlock()
 }
@@ -244,6 +246,30 @@ func (b *LeaderAwareRedisBackend) RefreshLeaderNow(ctx context.Context) {
 	b.refreshLeader(ctx)
 }
 
+func (b *LeaderAwareRedisBackend) refreshLeaderNowIfDue(ctx context.Context) {
+	b.refreshMu.Lock()
+	if b.refreshClosed {
+		b.refreshMu.Unlock()
+		return
+	}
+	done := b.refreshDone
+	if done == nil {
+		if !b.lastRefreshAt.IsZero() && time.Since(b.lastRefreshAt) < b.refreshInterval {
+			b.refreshMu.Unlock()
+			return
+		}
+		done = make(chan struct{})
+		b.refreshDone = done
+		go b.runLeaderRefresh(done)
+	}
+	b.refreshMu.Unlock()
+
+	select {
+	case <-done:
+	case <-ctx.Done():
+	}
+}
+
 func (b *LeaderAwareRedisBackend) probeLeader(ctx context.Context, addr string) (string, error) {
 	cli := b.getOrCreateClient(addr)
 	if cli == nil {
@@ -389,6 +415,29 @@ func (b *LeaderAwareRedisBackend) currentClient() *redis.Client {
 	return b.clients[b.leader]
 }
 
+func (b *LeaderAwareRedisBackend) currentClientOrRefresh(ctx context.Context) *redis.Client {
+	cli := b.currentClient()
+	if cli != nil {
+		return cli
+	}
+	b.refreshLeaderNowIfDue(ctx)
+	return b.currentClient()
+}
+
+func (b *LeaderAwareRedisBackend) firstSeedClient() *redis.Client {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	if b.closed {
+		return nil
+	}
+	for _, seed := range b.seeds {
+		if cli := b.clients[seed]; cli != nil {
+			return cli
+		}
+	}
+	return nil
+}
+
 // Do forwards a single command to the current leader. NOTLEADER refreshes the
 // cached leader for the next command, but the current command is not replayed:
 // leadership-loss errors can be returned after an operation has already applied.
@@ -404,7 +453,7 @@ func (b *LeaderAwareRedisBackend) Do(ctx context.Context, args ...any) *redis.Cm
 }
 
 func (b *LeaderAwareRedisBackend) doOnce(ctx context.Context, args ...any) *redis.Cmd {
-	cli := b.currentClient()
+	cli := b.currentClientOrRefresh(ctx)
 	if cli == nil {
 		cmd := redis.NewCmd(ctx, args...)
 		cmd.SetErr(ErrNoLeaderBackend)
@@ -426,7 +475,7 @@ func (b *LeaderAwareRedisBackend) DoWithTimeout(ctx context.Context, timeout tim
 }
 
 func (b *LeaderAwareRedisBackend) doWithTimeoutOnce(ctx context.Context, timeout time.Duration, args ...any) *redis.Cmd {
-	cli := b.currentClient()
+	cli := b.currentClientOrRefresh(ctx)
 	if cli == nil {
 		cmd := redis.NewCmd(ctx, args...)
 		cmd.SetErr(ErrNoLeaderBackend)
@@ -437,7 +486,7 @@ func (b *LeaderAwareRedisBackend) doWithTimeoutOnce(ctx context.Context, timeout
 
 // Pipeline forwards a batch to the current leader.
 func (b *LeaderAwareRedisBackend) Pipeline(ctx context.Context, cmds [][]any) ([]*redis.Cmd, error) {
-	cli := b.currentClient()
+	cli := b.currentClientOrRefresh(ctx)
 	if cli == nil {
 		return nil, ErrNoLeaderBackend
 	}
@@ -517,7 +566,10 @@ func isLeaderRefreshTransportError(err error) bool {
 
 // NewPubSub opens a subscribe connection on the current leader.
 func (b *LeaderAwareRedisBackend) NewPubSub(ctx context.Context) *redis.PubSub {
-	cli := b.currentClient()
+	cli := b.currentClientOrRefresh(ctx)
+	if cli == nil {
+		cli = b.firstSeedClient()
+	}
 	if cli == nil {
 		return nil
 	}