Tune redis proxy connection capacity

adapter/redis_peer_limiter_test.go

@@ -10,6 +10,23 @@ const (
 	testPeerLimit = 2
 )
 
+func TestDefaultRedisPeerLimiterUsesNoisyClientBoundedDefault(t *testing.T) {
+	t.Setenv(redisPerPeerLimitEnv, "")
+
+	limiter := newDefaultRedisPeerLimiter()
+	require.NotNil(t, limiter)
+	require.Equal(t, 8, limiter.limit)
+	require.Equal(t, defaultRedisPerPeerConnectionCap, limiter.limit)
+}
+
+func TestDefaultRedisPeerLimiterCanRaiseProxyDeploymentCap(t *testing.T) {
+	t.Setenv(redisPerPeerLimitEnv, "64")
+
+	limiter := newDefaultRedisPeerLimiter()
+	require.NotNil(t, limiter)
+	require.Equal(t, 64, limiter.limit)
+}
+
 func TestRedisPeerLimiterRejectsAndReleases(t *testing.T) {
 	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(testPeerLimit))
 	c1 := &remoteCommandRecorder{remote: "192.168.0.64:10001"}
@@ -78,7 +95,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()
 
@@ -95,7 +112,7 @@ 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: "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))
@@ -107,7 +124,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")

cmd/redis-proxy/main_test.go

@@ -72,10 +72,10 @@ func TestDeriveSecondaryConcurrency(t *testing.T) {
 			name:                    "dual write derives from ElasticKV pool",
 			mode:                    proxy.ModeDualWrite,
 			primaryPoolSize:         128,
-			elasticKVPoolSize:       4,
-			wantWriteConcurrency:    2,
-			wantScriptConcurrency:   1,
-			wantBlockingConcurrency: 2,
+			elasticKVPoolSize:       16,
+			wantWriteConcurrency:    8,
+			wantScriptConcurrency:   4,
+			wantBlockingConcurrency: 8,
 		},
 		{
 			name:                    "shadow mode derives from ElasticKV pool",

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 | 4 | Default per-leader pool; keep within the server per-peer connection limit |
+| ElasticKV connection pool size | 4 | Default per-leader pool; keep within the server per-peer connection limit and leave room for dedicated sockets |
 | Dial timeout | 5s | Backend connection timeout |
 | Read timeout | 3s | Backend read timeout |
 | Write timeout | 3s | Backend write timeout |
@@ -439,7 +439,8 @@ 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.
-- Increase the ElasticKV pool only together with `ELASTICKV_REDIS_PER_PEER_CONNECTIONS`; keep `-secondary-write-concurrency` at or below the pool size.
+- Increase the ElasticKV pool only together with `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` when the proxy pool can exceed the server's per-peer cap; keep `-secondary-write-concurrency` at or below the pool size.
+- For the production proxy shape that needs more backend connections, first deploy ElasticKV nodes with `ELASTICKV_REDIS_PER_PEER_CONNECTIONS=64`, then raise the proxy with `REDIS_PROXY_ELASTICKV_POOL_SIZE=16` or an explicit `-elastickv-pool-size=16`, keeping `-secondary-write-concurrency` below the pool size, for example `8`. Do not roll a larger proxy pool before the server-side cap is active on every node.
 - 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,9 +72,10 @@ func DefaultBackendOptions() BackendOptions {
 }
 
 // DefaultElasticKVBackendOptions returns defaults for proxy backends that
-// connect to ElasticKV's Redis adapter. ElasticKV limits concurrent Redis
-// connections per peer by default, so keep the pool below that cap unless the
-// operator also raises ELASTICKV_REDIS_PER_PEER_CONNECTIONS on the cluster.
+// connect to ElasticKV's Redis adapter. Keep the default within ElasticKV's
+// server-side per-peer cap while leaving room for dedicated Pub/Sub and
+// blocking-command sockets. Operators can still raise this together with
+// ELASTICKV_REDIS_PER_PEER_CONNECTIONS after the cluster is configured for it.
 func DefaultElasticKVBackendOptions() BackendOptions {
 	opts := DefaultBackendOptions()
 	opts.PoolSize = defaultElasticKVPoolSize

proxy/leader_aware_backend.go

@@ -75,8 +75,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 +106,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),
@@ -389,6 +389,15 @@ 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.RefreshLeaderNow(ctx)
+	return b.currentClient()
+}
+
 // 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 +413,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 +435,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 +446,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 +526,7 @@ 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 {
 		return nil
 	}

proxy/leader_aware_backend_test.go

@@ -553,6 +553,51 @@ func TestLeaderAwareRedisBackend_EvictsOldestNonProtectedClient(t *testing.T) {
 	assert.Contains(t, addrs, seed.addr, "seed must never be evicted")
 }
 
+func TestLeaderAwareRedisBackend_InitialCommandWaitsForLeaderDiscovery(t *testing.T) {
+	aliveLeader := newFakeElasticKVNode(t)
+	aliveLeader.SetLeader(aliveLeader.addr)
+	gate := make(chan struct{})
+	aliveLeader.SetInfoGate(gate)
+
+	backend := NewLeaderAwareRedisBackendWithInterval(
+		[]string{"127.0.0.1:1", aliveLeader.addr},
+		"elastickv",
+		DefaultBackendOptions(),
+		time.Hour, 200*time.Millisecond,
+		testLogger,
+	)
+	t.Cleanup(func() { _ = backend.Close() })
+
+	require.Empty(t, backend.CurrentLeader(), "initial commands must not target the first seed before discovery")
+
+	done := make(chan error, 1)
+	go func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		done <- backend.Do(ctx, "SET", "k", "v").Err()
+	}()
+
+	select {
+	case err := <-done:
+		t.Fatalf("command returned before leader discovery completed: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	close(gate)
+	var err error
+	require.Eventually(t, func() bool {
+		select {
+		case err = <-done:
+			return true
+		default:
+			return false
+		}
+	}, 2*time.Second, 10*time.Millisecond)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), aliveLeader.commands.Load(), "initial command must reach the discovered leader")
+	require.Equal(t, aliveLeader.addr, backend.CurrentLeader())
+}
+
 func TestLeaderAwareRedisBackend_FallsBackToSeedOnProbeFailure(t *testing.T) {
 	// Seed 1 is unreachable; seed 2 is alive and reports itself as leader.
 	aliveLeader := newFakeElasticKVNode(t)

proxy/proxy_test.go

@@ -784,7 +784,7 @@ func TestDualWriter_Blocking_BZPopReplayShortTimeoutStillAttemptsZRem(t *testing
 
 func TestNoEffectReplayRetryLimitIncludesJitterBudget(t *testing.T) {
 	limit := noEffectReplayRetryLimit(context.Background(), blockingReplayNoEffectRetryWindow)
-	assert.Positive(t, limit)
+	assert.Equal(t, 5, limit)
 
 	var spent time.Duration
 	backoff := compactedRetryInitialBackoff