feat(pool): keep a warm spare for full-duplex (#4101)

* feat(pool): keep a warm spare for full-duplex

The full-duplex engine holds one pipeline-pool connection for its whole
session, and the pipeline pool never kept idle connections. After a
connection error the replacement had to dial first, while every unacked
command waited.

Add Options.PipelineMinIdleConns (default 0, also on Cluster, Ring,
Failover and Universal options and the URL parsers) to keep idle
connections in the pipeline pool. A full-duplex engine raises it to at
least 1 through a new ConnPool.RaiseMinIdleConns; the pool refills the
spare in the background whenever it is leased. A negative value opts
out. Clients that never use full-duplex keep a zero idle footprint.

The spare saves the dial (TCP/TLS), not the connection init
(HELLO/AUTH), which still runs on first lease.

* fix(pool): keep the warm spare fresh

The pool checks ConnMaxIdleTime only when it hands a connection out.
Under LIFO the full-duplex spare is never leased, so it aged past
ConnMaxIdleTime (30m by default) and was discarded by the very lease it
exists for, which then dialed.

In a pool whose min-idle floor was raised, Get/TryGet now replace the
oldest idle connection once it is half-way to ConnMaxIdleTime, but only
when another idle connection serves the lease. The close goes through
the stale path, and the floor refills in the background. One atomic
load for all other pools.

Also fix the warm-spare test race seen on CI: IdleLen counts a min-idle
connection as soon as its dial starts, so the waits returned while the
spare was still dialing. Wait on Len as well.

* fix(pool): add the FD spare only when it fits

Add the warm spare only when the pipeline pool has room for it next to
the engines: PipelinePoolSize must exceed the engine count, or the
session spills to the main pool, and MaxIdleConns (if set) must too, or
every recycle evicts the returned session connection and dials again.

Also refresh the spare before its ConnMaxLifetime ends, not only before
ConnMaxIdleTime; re-check a raise that races an early-return min-idle
check; and fix the pipeline pool docs (permanent raise, the
FullDuplexMaxHold bound, cluster pool lifecycle, pipelinePoolOptions).

* fix(pool): use fdPipeCount after master merge

* fix(pool): refresh the spare after the lease

Closing the aged spare before popIdle could leave this Get with no
connection (a concurrent Get took the other idle one, or the newest one
failed the health check) and force a dial.

* test(pool): fold warm-spare tests into existing files

* docs(pool): note lifetime jitter for the FD spare

autopipeline_fullduplex.go

@@ -819,9 +819,31 @@ func newFDEngine(ap *AutoPipeliner, client *Client) *fdEngine {
 	} else {
 		fd.reprocess = client.processStartingAt
 	}
+	// Keep one warm spare in the dedicated pipeline pool. The engine holds one
+	// pipeline connection for its whole session, so without an idle spare the
+	// replacement after a connection error must dial first, while every unacked
+	// command waits. The pool refills the spare in the background whenever it is
+	// leased. A negative PipelineMinIdleConns opts out; on the main pool (no
+	// dedicated pipeline pool) the user's MinIdleConns applies instead.
+	if ref := client.loadPipelinePool(); ref != nil && client.opt.PipelineMinIdleConns >= 0 &&
+		fdSpareFits(ref.pool.Size(), client.opt.MaxIdleConns, fdPipeCount(ap.config)) {
+		ref.pool.RaiseMinIdleConns(1)
+	}
 	return fd
 }
 
+// fdSpareFits reports whether a pipeline pool has room for a warm spare next to
+// the engines' held connections. The spare must not take a slot an engine needs
+// (the session would spill to the main pool), and the idle limit must hold the
+// spare plus a returned session connection, or every recycle would evict that
+// connection and dial again. maxIdle <= 0 means no idle limit.
+func fdSpareFits(poolSize, maxIdle, engines int) bool {
+	if poolSize <= engines {
+		return false
+	}
+	return maxIdle <= 0 || maxIdle > engines
+}
+
 // submit enqueues a command onto the ordered stream and returns its batch.
 // Blocks when the queue is full (backpressure) or bails if the engine is
 // closing. The caller (AutoPipeliner.submit) stamps setReady on the async face.

autopipeline_fullduplex_test.go

@@ -326,23 +326,27 @@ func TestFullDuplexIdleReturnsConn(t *testing.T) {
 		t.Fatalf("initial set: %v", err)
 	}
 	// Immediately after the command the engine still holds the conn (idle gap not
-	// yet elapsed) — it is checked out of the pool, so IdleLen is 0. This is the
-	// baseline that makes the return below meaningful (the assertion would not
-	// hold if lease/return were removed — the conn would stay held, never idle).
-	if got := pp.IdleLen(); got != 0 {
-		t.Fatalf("expected the FD conn held (IdleLen=0) right after a command, got %d", got)
+	// yet elapsed). This is the baseline that makes the return below meaningful
+	// (it would not hold if lease/return were removed — the conn would stay held).
+	// Checked on the engine, not the pool: the pool also keeps the engine's warm
+	// spare idle, and its refill dial is asynchronous.
+	if ap.fd.curConn.Load() == nil {
+		t.Fatal("expected the FD conn held right after a command")
 	}
 
+	// Returned: the engine holds nothing and the pool has the returned conn plus
+	// the warm spare idle.
 	returned := false
 	for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
-		if pp.IdleLen() >= 1 {
+		if ap.fd.curConn.Load() == nil && pp.IdleLen() >= 2 {
 			returned = true
 			break
 		}
 		time.Sleep(10 * time.Millisecond)
 	}
 	if !returned {
-		t.Fatalf("pipeline conn not returned to the pool after idle (IdleLen=%d)", pp.IdleLen())
+		t.Fatalf("pipeline conn not returned to the pool after idle (held=%v, IdleLen=%d)",
+			ap.fd.curConn.Load() != nil, pp.IdleLen())
 	}
 
 	if v, err := ap.Get(ctx, "fd:idle:k").Result(); err != nil || v != "v" {
@@ -7028,3 +7032,201 @@ func TestFullDuplexPipesClusterChildForcedToOne(t *testing.T) {
 		t.Fatalf("Config().FullDuplexPipes = %d, want 1 (one pipe per master)", got)
 	}
 }
+
+// warmSpareClient returns a client whose dialer counts dials and, once slow is
+// set, takes slowDial per dial.
+func warmSpareClient(t *testing.T, minIdle int, slow *atomic.Bool) *Client {
+	t.Helper()
+	if err := probeRedis(internalTestRedisAddr()); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+	client := NewClient(&Options{
+		Addr:                 internalTestRedisAddr(),
+		PipelineMinIdleConns: minIdle,
+		Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			if slow.Load() {
+				time.Sleep(slowDial)
+			}
+			var d net.Dialer
+			return d.DialContext(ctx, network, addr)
+		},
+	})
+	t.Cleanup(func() { _ = client.Close() })
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+	return client
+}
+
+const slowDial = 500 * time.Millisecond
+
+// waitFDSession waits until the engine holds a session connection and the
+// pipeline pool has want idle connections whose dials have finished. IdleLen
+// counts a min-idle connection as soon as its dial starts; Len only once it is
+// in the pool, so Len must also reach the held connection plus the idle ones.
+func waitFDSession(t *testing.T, ap *AutoPipeliner, client *Client, want int) {
+	t.Helper()
+	pp := client.loadPipelinePool().pool
+	deadline := time.Now().Add(5 * time.Second)
+	for ap.fd.curConn.Load() == nil || pp.IdleLen() != want || pp.Len() != want+1 {
+		if time.Now().After(deadline) {
+			t.Fatalf("session held=%v, pipeline Len=%d IdleLen=%d, want held with %d idle",
+				ap.fd.curConn.Load() != nil, pp.Len(), pp.IdleLen(), want)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// TestFDWarmSpareReplacesWithoutDial checks the full-duplex engine keeps an
+// idle spare in the pipeline pool, so after its connection fails the
+// replacement does not wait for a dial.
+func TestFDWarmSpareReplacesWithoutDial(t *testing.T) {
+	ctx := context.Background()
+	var slow atomic.Bool
+	client := warmSpareClient(t, 0, &slow)
+
+	ap, err := newAutoPipeliner(client, &AutoPipelineOptions{FullDuplex: true}, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ap.Close()
+	if ap.fd == nil {
+		t.Skip("full-duplex engine not enabled for this client")
+	}
+	if err := ap.Set(ctx, "fdspare:k", "1", 0).Err(); err != nil {
+		t.Fatal(err)
+	}
+	waitFDSession(t, ap, client, 1)
+
+	// From now on a dial is slow. Break the held connection: the next command
+	// fails its write and is replayed on the replacement.
+	slow.Store(true)
+	_ = ap.fd.curConn.Load().Close()
+
+	start := time.Now()
+	if err := ap.Set(ctx, "fdspare:k", "2", 0).Err(); err != nil {
+		t.Fatalf("command after connection failure: %v", err)
+	}
+	if d := time.Since(start); d >= slowDial/2 {
+		t.Fatalf("recovery took %v; the replacement dialed instead of leasing the warm spare", d)
+	}
+	if v, err := client.Get(ctx, "fdspare:k").Result(); err != nil || v != "2" {
+		t.Fatalf("get = %q, %v; want \"2\"", v, err)
+	}
+}
+
+// TestFDWarmSpareOptOut checks a negative PipelineMinIdleConns keeps the
+// pipeline pool at zero idle connections under full-duplex.
+func TestFDWarmSpareOptOut(t *testing.T) {
+	ctx := context.Background()
+	var slow atomic.Bool
+	client := warmSpareClient(t, -1, &slow)
+
+	ap, err := newAutoPipeliner(client, &AutoPipelineOptions{FullDuplex: true}, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ap.Close()
+	if ap.fd == nil {
+		t.Skip("full-duplex engine not enabled for this client")
+	}
+	if err := ap.Set(ctx, "fdspare:o", "1", 0).Err(); err != nil {
+		t.Fatal(err)
+	}
+	waitFDSession(t, ap, client, 0)
+	time.Sleep(50 * time.Millisecond)
+	if n := client.loadPipelinePool().pool.IdleLen(); n != 0 {
+		t.Fatalf("pipeline IdleLen = %d with PipelineMinIdleConns -1, want 0", n)
+	}
+}
+
+// TestFDSpareFits checks the spare is only added when it does not take a slot
+// an engine needs and the idle limit can hold it next to a returned session
+// connection (Codex on #4101).
+func TestFDSpareFits(t *testing.T) {
+	for _, tc := range []struct {
+		poolSize, maxIdle, engines int
+		want                       bool
+	}{
+		{10, 0, 1, true},  // defaults
+		{1, 0, 1, false},  // the only slot belongs to the engine
+		{2, 0, 2, false},  // NumShards == PipelinePoolSize
+		{3, 0, 2, true},   // room for two engines and a spare
+		{10, 1, 1, false}, // spare would fill the only idle slot
+		{10, 2, 1, true},  // spare + returned session conn
+		{10, 2, 2, false}, // two engines need both idle slots
+	} {
+		if got := fdSpareFits(tc.poolSize, tc.maxIdle, tc.engines); got != tc.want {
+			t.Errorf("fdSpareFits(pool=%d, maxIdle=%d, engines=%d) = %v, want %v",
+				tc.poolSize, tc.maxIdle, tc.engines, got, tc.want)
+		}
+	}
+}
+
+// TestFDNoSpareWhenItDoesNotFit checks a client whose pipeline pool cannot fit
+// a spare gets none, and that the full-duplex session then stays on the
+// pipeline pool instead of spilling to the main pool.
+func TestFDNoSpareWhenItDoesNotFit(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		opt  func(*Options)
+	}{
+		{"PipelinePoolSize 1", func(o *Options) { o.PipelinePoolSize = 1 }},
+		{"MaxIdleConns 1", func(o *Options) { o.MaxIdleConns = 1 }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			if err := probeRedis(internalTestRedisAddr()); err != nil {
+				t.Skipf("no redis: %v", err)
+			}
+			opt := &Options{Addr: internalTestRedisAddr()}
+			tc.opt(opt)
+			client := NewClient(opt)
+			defer client.Close()
+			if err := client.Ping(ctx).Err(); err != nil {
+				t.Skipf("no redis: %v", err)
+			}
+			ap, err := newAutoPipeliner(client, &AutoPipelineOptions{FullDuplex: true}, false)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer ap.Close()
+			if ap.fd == nil {
+				t.Skip("full-duplex engine not enabled for this client")
+			}
+			if err := ap.Set(ctx, "fdspare:fit", "1", 0).Err(); err != nil {
+				t.Fatal(err)
+			}
+			waitFDSession(t, ap, client, 0)
+			time.Sleep(50 * time.Millisecond)
+			pp := client.loadPipelinePool().pool
+			if n := pp.IdleLen(); n != 0 {
+				t.Fatalf("pipeline IdleLen = %d, want 0 (no spare)", n)
+			}
+			if ap.fd.curConnSpilled.Load() {
+				t.Fatal("full-duplex session spilled to the main pool")
+			}
+		})
+	}
+}
+
+// TestPipelineMinIdleConnsWithoutFD checks PipelineMinIdleConns alone keeps
+// idle connections in the pipeline pool, and the default keeps none.
+func TestPipelineMinIdleConnsWithoutFD(t *testing.T) {
+	var slow atomic.Bool
+	client := warmSpareClient(t, 2, &slow)
+	pp := client.loadPipelinePool().pool
+	deadline := time.Now().Add(5 * time.Second)
+	for pp.IdleLen() != 2 || pp.Len() != 2 {
+		if time.Now().After(deadline) {
+			t.Fatalf("pipeline Len=%d IdleLen=%d, want 2/2", pp.Len(), pp.IdleLen())
+		}
+		time.Sleep(time.Millisecond)
+	}
+
+	def := warmSpareClient(t, 0, &slow)
+	time.Sleep(50 * time.Millisecond)
+	if n := def.loadPipelinePool().pool.IdleLen(); n != 0 {
+		t.Fatalf("default pipeline IdleLen = %d, want 0", n)
+	}
+}

commands_test.go

@@ -83,7 +83,18 @@ var _ = Describe("Commands", func() {
 			Expect(cmds[0].Err().Error()).To(authErr)
 
 			stats := rawClient.PoolStats()
-			if stats.PipelineStats != nil {
+			if stats.PipelineStats != nil && isFullDuplexSubject() {
+				// Full-duplex keeps a warm spare in the pipeline pool, refilled
+				// by a background dial, so whether a lease hits the spare or
+				// dials races that refill, and the held session can carry the
+				// second Pipelined without a pool operation. Pin the intent
+				// instead: the pipelines ran on the pipeline pool, and the main
+				// pool served only the BeforeEach FlushDB.
+				Expect(stats.PipelineStats.Hits + stats.PipelineStats.Misses).To(BeNumerically(">=", 1))
+				Expect(stats.PipelineStats.Timeouts).To(Equal(uint32(0)))
+				Expect(stats.Hits).To(Equal(uint32(0)))
+				Expect(stats.Misses).To(Equal(uint32(1)))
+			} else if stats.PipelineStats != nil {
 				// The autopipeline subject faces create the dedicated pipeline
 				// pool lazily, so the two Pipelined calls above ran there: the
 				// first dialed (miss), the second reused (hit). The main pool

internal/pool/pool.go

@@ -433,6 +433,10 @@ type ConnPool struct {
 	idleCheckInProgress atomic.Bool
 	idleCheckNeeded     atomic.Bool
 
+	// minIdleFloor is a runtime lower bound on cfg.MinIdleConns, raised by
+	// RaiseMinIdleConns (the full-duplex engine keeps a warm spare). 0 = none.
+	minIdleFloor atomic.Int32
+
 	stats          Stats
 	waitDurationNs atomic.Int64
 
@@ -527,6 +531,80 @@ func (p *ConnPool) RemovePoolHook(hook PoolHook) {
 	}
 }
 
+// minIdleConns returns the effective minimum idle connections: the configured
+// MinIdleConns, or the runtime floor if higher.
+func (p *ConnPool) minIdleConns() int32 {
+	if f := p.minIdleFloor.Load(); f > p.cfg.MinIdleConns {
+		return f
+	}
+	return p.cfg.MinIdleConns
+}
+
+// RaiseMinIdleConns raises the pool's minimum idle connections to at least n
+// and starts filling it in the background. It never lowers the configured
+// MinIdleConns or an earlier raise. Like MinIdleConns, the idle connections
+// count against PoolSize and are dialed but not initialized: init runs on
+// first Get. A raised pool also keeps its spare fresh: see refreshStaleSpare.
+func (p *ConnPool) RaiseMinIdleConns(n int32) {
+	for {
+		cur := p.minIdleFloor.Load()
+		if n <= cur {
+			return
+		}
+		if p.minIdleFloor.CompareAndSwap(cur, n) {
+			break
+		}
+	}
+	if p.closed() {
+		return
+	}
+	p.checkMinIdleConns()
+}
+
+// refreshStaleSpare replaces the oldest idle connection once it is half-way to
+// ConnMaxIdleTime or to the end of its ConnMaxLifetime, in a pool whose min-idle
+// floor was raised (RaiseMinIdleConns, the full-duplex warm spare). Under LIFO
+// such a spare is never leased, so it would age out and be discarded by the
+// very lease it exists for, which then dials. Get runs it only after the lease
+// has its healthy connection, so it never closes this lease's connection or its
+// fallback. O(1): under LIFO the oldest idle connection is idleConns[0].
+func (p *ConnPool) refreshStaleSpare(ctx context.Context, nowNs int64) {
+	if p.minIdleFloor.Load() == 0 || p.cfg.PoolFIFO ||
+		(p.cfg.ConnMaxIdleTime <= 0 && p.cfg.ConnMaxLifetime <= 0) {
+		return
+	}
+	p.connsMu.Lock()
+	if len(p.idleConns) == 0 {
+		p.connsMu.Unlock()
+		return
+	}
+	oldest := p.idleConns[0]
+	if !p.spareAging(oldest, nowNs) || !oldest.TryAcquire() {
+		p.connsMu.Unlock()
+		return
+	}
+	copy(p.idleConns, p.idleConns[1:])
+	p.idleConns = p.idleConns[:len(p.idleConns)-1]
+	p.idleConnsLen.Add(-1)
+	p.connsMu.Unlock()
+
+	_ = p.CloseConn(ctx, oldest, CloseReasonStale, MetricStateIdle)
+	p.checkMinIdleConns()
+}
+
+// spareAging reports whether cn is half-way to ConnMaxIdleTime or has less than
+// half of its ConnMaxLifetime left, so a refresh now lands before isHealthyConn
+// would reject it.
+func (p *ConnPool) spareAging(cn *Conn, nowNs int64) bool {
+	if idle := p.cfg.ConnMaxIdleTime; idle > 0 && nowNs-cn.UsedAtNs() >= int64(idle/2) {
+		return true
+	}
+	if life := p.cfg.ConnMaxLifetime; life > 0 && cn.expiresAt.UnixNano()-nowNs < int64(life/2) {
+		return true
+	}
+	return false
+}
+
 func (p *ConnPool) checkMinIdleConns() {
 	// If a check is already in progress, mark that we need another check and return
 	if !p.idleCheckInProgress.CompareAndSwap(false, true) {
@@ -534,8 +612,14 @@ func (p *ConnPool) checkMinIdleConns() {
 		return
 	}
 
-	if p.cfg.MinIdleConns == 0 {
+	if p.minIdleConns() == 0 {
 		p.idleCheckInProgress.Store(false)
+		// A RaiseMinIdleConns between the read above and the store may have
+		// found the check in progress and only set idleCheckNeeded: run again
+		// so the new floor is filled now, not on a later Get.
+		if p.idleCheckNeeded.Load() && p.minIdleConns() > 0 {
+			p.checkMinIdleConns()
+		}
 		return
 	}
 
@@ -544,10 +628,13 @@ func (p *ConnPool) checkMinIdleConns() {
 	for {
 		// Clear the "check needed" flag before we start
 		p.idleCheckNeeded.Store(false)
+		// Re-read each pass: a RaiseMinIdleConns during a check only sets
+		// idleCheckNeeded, so the new floor must be picked up here.
+		minIdle := p.minIdleConns()
 
 		// Only create idle connections if we haven't reached the total pool size limit
 		// MinIdleConns should be a subset of PoolSize, not additional connections
-		for p.poolSize.Load() < p.cfg.PoolSize && p.idleConnsLen.Load() < p.cfg.MinIdleConns {
+		for p.poolSize.Load() < p.cfg.PoolSize && p.idleConnsLen.Load() < minIdle {
 			// Try to acquire a semaphore token
 			if !p.semaphore.TryAcquire() {
 				// Semaphore is full, can't create more connections right now
@@ -996,6 +1083,8 @@ retryIdle:
 			cb(ctx, -1, cn, poolName)
 		}
 
+		// Only now, with this lease's connection taken and checked.
+		p.refreshStaleSpare(ctx, nowNs)
 		return cn, nil
 	}
 

internal/pool/pool_test.go

@@ -1968,3 +1968,368 @@ func TestTryGetReturnsTryFullWithoutCountingTimeout(t *testing.T) {
 func init() {
 	logging.Disable()
 }
+
+// waitPool waits until the pool has total connections, idle of them idle.
+// IdleLen counts a min-idle connection as soon as its dial STARTS; Len only
+// once the dial finished and the connection is in the pool, so both are needed
+// to know the idle connections are really there.
+func waitPool(t *testing.T, p *pool.ConnPool, total, idle int) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for p.Len() != total || p.IdleLen() != idle {
+		if time.Now().After(deadline) {
+			t.Fatalf("Len=%d IdleLen=%d, want %d/%d", p.Len(), p.IdleLen(), total, idle)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func newRaiseTestPool(t *testing.T, minIdle int32, dials *atomic.Int32) *pool.ConnPool {
+	t.Helper()
+	p := pool.NewConnPool(&pool.Options{
+		Dialer: func(ctx context.Context) (net.Conn, error) {
+			dials.Add(1)
+			return dummyDialer(ctx)
+		},
+		PoolSize:           4,
+		MinIdleConns:       minIdle,
+		MaxConcurrentDials: 4,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = p.Close() })
+	return p
+}
+
+// TestRaiseMinIdleConnsFillsAndRefills checks RaiseMinIdleConns pre-dials up to
+// the new floor and that the pool refills it after the idle connection is taken.
+func TestRaiseMinIdleConnsFillsAndRefills(t *testing.T) {
+	var dials atomic.Int32
+	p := newRaiseTestPool(t, 0, &dials)
+	if p.IdleLen() != 0 || dials.Load() != 0 {
+		t.Fatalf("fresh pool: IdleLen=%d dials=%d, want 0/0", p.IdleLen(), dials.Load())
+	}
+
+	p.RaiseMinIdleConns(1)
+	waitPool(t, p, 1, 1)
+
+	cn, err := p.Get(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The leased spare is replaced in the background.
+	waitPool(t, p, 2, 1)
+	if got := dials.Load(); got != 2 {
+		t.Fatalf("dials = %d, want 2 (spare + refill)", got)
+	}
+	p.Put(context.Background(), cn)
+}
+
+// spareRefreshPool returns a LIFO pool with a short ConnMaxIdleTime, set up as
+// the full-duplex engine leaves it: one leased-and-returned connection on top
+// (newest) and, below it, the min-idle spare (oldest). It returns the pool and
+// the returned connection, which the next Get should hand out.
+func spareRefreshPool(t *testing.T, raise bool, dials *atomic.Int32) (*pool.ConnPool, *pool.Conn) {
+	t.Helper()
+	return spareRefreshPoolWith(t, raise, dials, 400*time.Millisecond, 0)
+}
+
+func spareRefreshPoolWith(t *testing.T, raise bool, dials *atomic.Int32, maxIdle, maxLife time.Duration) (*pool.ConnPool, *pool.Conn) {
+	t.Helper()
+	p := pool.NewConnPool(&pool.Options{
+		Dialer: func(ctx context.Context) (net.Conn, error) {
+			dials.Add(1)
+			return dummyDialer(ctx)
+		},
+		PoolSize:           4,
+		MaxConcurrentDials: 4,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    maxIdle,
+		ConnMaxLifetime:    maxLife,
+	})
+	t.Cleanup(func() { _ = p.Close() })
+	ctx := context.Background()
+	if raise {
+		p.RaiseMinIdleConns(1)
+		waitPool(t, p, 1, 1)
+	}
+	cn, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if raise {
+		waitPool(t, p, 2, 1) // the leased spare was replaced
+	}
+	p.Put(ctx, cn)
+	return p, cn
+}
+
+// TestRefreshStaleSpareReplacesAgedSpare checks a spare past half of
+// ConnMaxIdleTime is replaced on the next Get, which still gets the newest
+// idle connection, so the spare never ages out while it waits.
+func TestRefreshStaleSpareReplacesAgedSpare(t *testing.T) {
+	var dials atomic.Int32
+	p, returned := spareRefreshPool(t, true, &dials)
+	before := dials.Load()
+
+	time.Sleep(300 * time.Millisecond) // > ConnMaxIdleTime/2, < ConnMaxIdleTime
+	cn, err := p.Get(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cn != returned {
+		t.Fatal("Get did not return the newest idle connection")
+	}
+	// The aged spare was closed and a fresh one dialed in the background.
+	waitPool(t, p, 2, 1)
+	if got := dials.Load() - before; got != 1 {
+		t.Fatalf("dials after refresh = %d, want 1", got)
+	}
+	p.Put(context.Background(), cn)
+}
+
+// TestRefreshStaleSpareBeforeLifetimeEnds checks the spare is also replaced
+// once less than half of its ConnMaxLifetime is left, with ConnMaxIdleTime off
+// (Codex on #4101).
+func TestRefreshStaleSpareBeforeLifetimeEnds(t *testing.T) {
+	var dials atomic.Int32
+	p, returned := spareRefreshPoolWith(t, true, &dials, -1, 800*time.Millisecond)
+	before := dials.Load()
+
+	time.Sleep(500 * time.Millisecond) // < half of the lifetime left, not expired
+	cn, err := p.Get(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cn != returned {
+		t.Fatal("Get did not return the newest idle connection")
+	}
+	waitPool(t, p, 2, 1)
+	if got := dials.Load() - before; got != 1 {
+		t.Fatalf("dials after lifetime refresh = %d, want 1", got)
+	}
+	p.Put(context.Background(), cn)
+}
+
+// TestRefreshStaleSpareKeepsYoungSpare checks a spare younger than half of
+// ConnMaxIdleTime is kept.
+func TestRefreshStaleSpareKeepsYoungSpare(t *testing.T) {
+	var dials atomic.Int32
+	p, _ := spareRefreshPool(t, true, &dials)
+	before := dials.Load()
+
+	cn, err := p.Get(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(20 * time.Millisecond)
+	if got := dials.Load() - before; got != 0 {
+		t.Fatalf("dials = %d for a young spare, want 0", got)
+	}
+	p.Put(context.Background(), cn)
+}
+
+// TestRefreshStaleSpareOnlyRaisedPools checks a pool whose floor was never
+// raised keeps its old idle connections (no proactive close).
+func TestRefreshStaleSpareOnlyRaisedPools(t *testing.T) {
+	var dials atomic.Int32
+	p, _ := spareRefreshPool(t, false, &dials)
+	ctx := context.Background()
+	// Two idle connections, the older one aged past half of ConnMaxIdleTime.
+	extra, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	p.Put(ctx, extra)
+	cn, err := p.Get(ctx) // newest
+	if err != nil {
+		t.Fatal(err)
+	}
+	other, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	p.Put(ctx, other)
+	p.Put(ctx, cn)
+	before := dials.Load()
+	lenBefore := p.Len()
+
+	time.Sleep(300 * time.Millisecond)
+	got, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(20 * time.Millisecond)
+	if d := dials.Load() - before; d != 0 || p.Len() != lenBefore {
+		t.Fatalf("unraised pool: dials=%d Len=%d (was %d), want no refresh", d, p.Len(), lenBefore)
+	}
+	p.Put(ctx, got)
+}
+
+// TestRefreshStaleSpareWithConfiguredMinIdle checks a raise to a floor the
+// configured MinIdleConns already meets still enables the spare refresh: the
+// floor is tracked apart from MinIdleConns (Codex on #4101).
+func TestRefreshStaleSpareWithConfiguredMinIdle(t *testing.T) {
+	var dials atomic.Int32
+	p := pool.NewConnPool(&pool.Options{
+		Dialer: func(ctx context.Context) (net.Conn, error) {
+			dials.Add(1)
+			return dummyDialer(ctx)
+		},
+		PoolSize:           4,
+		MinIdleConns:       1,
+		MaxConcurrentDials: 4,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    400 * time.Millisecond,
+	})
+	t.Cleanup(func() { _ = p.Close() })
+	ctx := context.Background()
+	waitPool(t, p, 1, 1)
+	p.RaiseMinIdleConns(1)
+
+	returned, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitPool(t, p, 2, 1) // the leased spare was replaced
+	p.Put(ctx, returned)
+	before := dials.Load()
+
+	time.Sleep(300 * time.Millisecond) // > ConnMaxIdleTime/2, < ConnMaxIdleTime
+	cn, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cn != returned {
+		t.Fatal("Get did not return the newest idle connection")
+	}
+	waitPool(t, p, 2, 1)
+	if got := dials.Load() - before; got != 1 {
+		t.Fatalf("dials after refresh = %d, want 1", got)
+	}
+	p.Put(ctx, cn)
+}
+
+// TestRefreshStaleSpareKeepsFallback checks the refresh does not close the aged
+// spare before the lease has a connection: when the newest idle connection is
+// dead, Get falls back to the spare instead of dialing (Bugbot on #4101).
+func TestRefreshStaleSpareKeepsFallback(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = ln.Close() })
+	var mu sync.Mutex
+	servers := map[string]net.Conn{}
+	go func() {
+		for {
+			c, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			mu.Lock()
+			servers[c.RemoteAddr().String()] = c
+			mu.Unlock()
+		}
+	}()
+	t.Cleanup(func() {
+		mu.Lock()
+		defer mu.Unlock()
+		for _, c := range servers {
+			_ = c.Close()
+		}
+	})
+
+	var dials atomic.Int32
+	var clients []net.Conn
+	p := pool.NewConnPool(&pool.Options{
+		Dialer: func(ctx context.Context) (net.Conn, error) {
+			c, err := (&net.Dialer{}).DialContext(ctx, "tcp", ln.Addr().String())
+			if err == nil {
+				mu.Lock()
+				clients = append(clients, c)
+				mu.Unlock()
+				dials.Add(1)
+			}
+			return c, err
+		},
+		PoolSize:           4,
+		MaxConcurrentDials: 4,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    400 * time.Millisecond,
+	})
+	t.Cleanup(func() { _ = p.Close() })
+	ctx := context.Background()
+	p.RaiseMinIdleConns(1)
+	waitPool(t, p, 1, 1)
+	returned, err := p.Get(ctx) // the first dial
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitPool(t, p, 2, 1) // the leased spare was replaced
+	p.Put(ctx, returned)
+
+	// Kill the server side of the newest idle connection.
+	mu.Lock()
+	addr := clients[0].LocalAddr().String()
+	srv := servers[addr]
+	mu.Unlock()
+	if srv == nil {
+		t.Fatal("server side of the first connection not found")
+	}
+	_ = srv.Close()
+	before := dials.Load()
+
+	time.Sleep(300 * time.Millisecond) // > ConnMaxIdleTime/2, < ConnMaxIdleTime
+	cn, err := p.Get(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cn == returned {
+		t.Fatal("Get returned the dead connection")
+	}
+	// Only the background refill dials: Get took the aged spare.
+	waitPool(t, p, 2, 1)
+	time.Sleep(50 * time.Millisecond)
+	if got := dials.Load() - before; got != 1 {
+		t.Fatalf("dials = %d, want 1 (refill only, no synchronous dial)", got)
+	}
+	p.Put(ctx, cn)
+}
+
+// TestRaiseMinIdleConnsNeverLowers checks a raise below the configured
+// MinIdleConns (or an earlier raise) changes nothing.
+func TestRaiseMinIdleConnsNeverLowers(t *testing.T) {
+	var dials atomic.Int32
+	p := newRaiseTestPool(t, 3, &dials)
+	waitPool(t, p, 3, 3)
+
+	p.RaiseMinIdleConns(1)
+	p.RaiseMinIdleConns(0)
+	p.RaiseMinIdleConns(-1)
+	time.Sleep(20 * time.Millisecond)
+	if got := p.IdleLen(); got != 3 {
+		t.Fatalf("IdleLen = %d after lower raises, want 3", got)
+	}
+	if got := dials.Load(); got != 3 {
+		t.Fatalf("dials = %d, want 3", got)
+	}
+}
+
+// TestRaiseMinIdleConnsClosedPool checks a raise on a closed pool dials
+// nothing.
+func TestRaiseMinIdleConnsClosedPool(t *testing.T) {
+	var dials atomic.Int32
+	p := newRaiseTestPool(t, 0, &dials)
+	_ = p.Close()
+
+	p.RaiseMinIdleConns(1)
+	time.Sleep(20 * time.Millisecond)
+	if got := dials.Load(); got != 0 {
+		t.Fatalf("dials = %d on a closed pool, want 0", got)
+	}
+}

main_test.go

@@ -613,6 +613,14 @@ func cleanupTLSCluster() {
 	// TLS cluster is auto-managed by the container, no cleanup needed
 }
 
+// isFullDuplexSubject reports whether GOREDIS_TEST_SUBJECT selects a
+// full-duplex face (ap-fd, ap-fd-blocking). Those keep a warm spare in the
+// pipeline pool, so exact pipeline-pool counts differ.
+func isFullDuplexSubject() bool {
+	s := os.Getenv("GOREDIS_TEST_SUBJECT")
+	return s == "ap-fd" || s == "ap-fd-blocking"
+}
+
 // newUniversalSubject returns the redis.UniversalClient that a parametrized
 // command suite runs against, selected by the GOREDIS_TEST_SUBJECT env var so a
 // whole suite can be replayed through the autopipeliner without touching each

options.go

@@ -268,9 +268,10 @@ type Options struct {
 	//
 	// The dedicated pipeline pool is created unconditionally at NewClient —
 	// like the pubsub pool — so pipelines never compete with regular commands
-	// for main-pool connections. It never pre-dials (MinIdleConns is forced
-	// to 0 on it), so the size is a cap on burst capacity, not a standing
-	// footprint: an unused pipeline pool holds zero connections. A burst of
+	// for main-pool connections. By default it never pre-dials (it does not
+	// inherit MinIdleConns; see PipelineMinIdleConns), so the size is a cap on
+	// burst capacity, not a standing footprint: an unused pipeline pool holds
+	// zero connections. A burst of
 	// concurrent pipelines wider than the cap spills to the main pool IMMEDIATELY
 	// (a non-blocking TryGet on the pipeline pool) rather than waiting a grace
 	// period — so a saturated pipeline pool never adds latency before falling back,
@@ -290,6 +291,31 @@ type Options struct {
 	// default: DefaultPipelinePoolSize (10) connections
 	PipelinePoolSize int
 
+	// PipelineMinIdleConns is the minimum number of idle connections kept in the
+	// dedicated pipeline pool, like MinIdleConns for the main pool. They count
+	// against PipelinePoolSize and are dialed but not initialized (init runs on
+	// first use).
+	//
+	// Creating a full-duplex autopipeliner (AutoPipelineOptions.FullDuplex)
+	// raises it to at least 1 for the rest of the client's life: the engine
+	// holds one pipeline connection for its whole session, so without an idle
+	// spare a replacement after a connection error must dial first. The spare is
+	// added only when it fits: PipelinePoolSize and MaxIdleConns (if set) must
+	// both exceed the number of full-duplex engines (NumShards).
+	//
+	// The spare is replaced when the session connection is returned and leased
+	// again (at least every FullDuplexMaxHold) once it is half-way to
+	// ConnMaxIdleTime or to the end of ConnMaxLifetime, so it does not age out
+	// unused. Keep FullDuplexMaxHold below half of those limits, and below
+	// ConnMaxLifetime minus ConnMaxLifetimeJitter, for that to hold; the
+	// defaults (5s against 30m) do.
+	//
+	// Set a negative value to keep the pool at zero idle connections even with
+	// full-duplex.
+	//
+	// default: 0
+	PipelineMinIdleConns int
+
 	// AutoPipelineOptions is the default config for BOTH autopipeliner faces:
 	// AutoPipeline and AsyncAutoPipeline use it when called without an
 	// explicit config, falling back to their per-face defaults
@@ -589,9 +615,10 @@ const (
 // DefaultPipelinePoolSize is the pipeline pool size used when
 // PipelinePoolSize is not set. Pipelining batches many commands per round
 // trip, so it needs far fewer connections than regular traffic. The pool is
-// pure burst capacity: it never pre-dials idle connections (MinIdleConns is
-// forced to 0 on it), so an unused pipeline pool holds no connections at all
-// and the size is only a cap — bursts wider than it spill to the main pool.
+// pure burst capacity: by default it never pre-dials idle connections (it does
+// not inherit MinIdleConns; see PipelineMinIdleConns), so an unused pipeline
+// pool holds no connections at all and the size is only a cap — bursts wider
+// than it spill to the main pool.
 const DefaultPipelinePoolSize = 10
 
 // DefaultPipelineBufferSize is the per-connection read/write buffer size for
@@ -608,9 +635,9 @@ const DefaultPipelinePoolSize = 10
 // (see MaxBatchBytes's default) — more bufio headroom before a write() to a
 // slow-draining peer blocks widens the margin before that guardrail's cap is
 // reached. The aggregate memory cost stays negligible because the pool this
-// backs holds few connections: the dedicated pipeline pool never pre-dials
-// (DefaultPipelinePoolSize, MinIdleConns forced to 0), and full-duplex holds
-// exactly one connection per node regardless of pool size.
+// backs holds few connections: the dedicated pipeline pool does not pre-dial by
+// default (DefaultPipelinePoolSize, PipelineMinIdleConns 0), and full-duplex
+// holds one connection per engine plus one idle spare, regardless of pool size.
 const DefaultPipelineBufferSize = 128 * 1024
 
 // DefaultPipelinePoolTimeout is the dedicated pipeline pool's PoolTimeout
@@ -1104,6 +1131,7 @@ func setupConnParams(u *url.URL, o *Options) (*Options, error) {
 	// (pipeline_pool_size=-1) or tune it, otherwise these would be rejected as
 	// unexpected options. q.int accepts a negative value.
 	o.PipelinePoolSize = q.int("pipeline_pool_size")
+	o.PipelineMinIdleConns = q.int("pipeline_min_idle_conns")
 	o.PipelineReadBufferSize = q.int("pipeline_read_buffer_size")
 	o.PipelineWriteBufferSize = q.int("pipeline_write_buffer_size")
 	if q.has("conn_max_idle_time") {

options_test.go

@@ -742,4 +742,15 @@ func TestParseURLPipelinePoolOptions(t *testing.T) {
 	if fo.PipelinePoolSize != -1 || fo.PipelineReadBufferSize != 131072 || fo.PipelineWriteBufferSize != 65536 {
 		t.Fatalf("ParseFailoverURL pipeline opts: size=%d rbuf=%d wbuf=%d", fo.PipelinePoolSize, fo.PipelineReadBufferSize, fo.PipelineWriteBufferSize)
 	}
+
+	const mi = "pipeline_min_idle_conns=2"
+	if o, err := ParseURL("redis://localhost:6379?" + mi); err != nil || o.PipelineMinIdleConns != 2 {
+		t.Fatalf("ParseURL pipeline_min_idle_conns: %v, %+v", err, o)
+	}
+	if co, err := ParseClusterURL("redis://localhost:6379?" + mi); err != nil || co.PipelineMinIdleConns != 2 {
+		t.Fatalf("ParseClusterURL pipeline_min_idle_conns: %v, %+v", err, co)
+	}
+	if fo, err := ParseFailoverURL("redis://localhost:6379?master_name=mymaster&" + mi); err != nil || fo.PipelineMinIdleConns != 2 {
+		t.Fatalf("ParseFailoverURL pipeline_min_idle_conns: %v, %+v", err, fo)
+	}
 }

osscluster.go

@@ -156,13 +156,16 @@ type ClusterOptions struct {
 	// default: 32KiB (32768 bytes)
 	WriteBufferSize int
 
-	// PipelineReadBufferSize, PipelineWriteBufferSize and PipelinePoolSize
-	// configure an optional separate connection pool used for pipelining on
-	// each node, with its own (typically larger) buffers. See the same-named
-	// fields on Options for details. The pool is created only when PipelineReadBufferSize or PipelineWriteBufferSize is set (PipelinePoolSize alone does not enable it).
+	// PipelineReadBufferSize, PipelineWriteBufferSize, PipelinePoolSize and
+	// PipelineMinIdleConns configure the separate connection pool used for
+	// pipelining on each node, with its own (typically larger) buffers. See the
+	// same-named fields on Options for details. Each node client creates this
+	// pool by default; set PipelinePoolSize < 0 to opt out (pipelines run on the
+	// main pool).
 	PipelineReadBufferSize  int
 	PipelineWriteBufferSize int
 	PipelinePoolSize        int
+	PipelineMinIdleConns    int
 
 	// AutoPipelineOptions is the default config for BOTH autopipeliner faces
 	// (AutoPipeline and AsyncAutoPipeline), applied when they are called
@@ -413,6 +416,7 @@ func setupClusterQueryParams(u *url.URL, o *ClusterOptions) (*ClusterOptions, er
 	// Pipeline pool (per node, created by default): allow URL opt-out
 	// (pipeline_pool_size=-1) / tuning, else rejected as unexpected options.
 	o.PipelinePoolSize = q.int("pipeline_pool_size")
+	o.PipelineMinIdleConns = q.int("pipeline_min_idle_conns")
 	o.PipelineReadBufferSize = q.int("pipeline_read_buffer_size")
 	o.PipelineWriteBufferSize = q.int("pipeline_write_buffer_size")
 	o.PoolTimeout = q.duration("pool_timeout")
@@ -495,6 +499,7 @@ func (opt *ClusterOptions) clientOptions() *Options {
 		PipelineReadBufferSize:  opt.PipelineReadBufferSize,
 		PipelineWriteBufferSize: opt.PipelineWriteBufferSize,
 		PipelinePoolSize:        opt.PipelinePoolSize,
+		PipelineMinIdleConns:    opt.PipelineMinIdleConns,
 		DisableIdentity:         opt.DisableIdentity,
 		DisableIndentity:        opt.DisableIndentity,
 		IdentitySuffix:          opt.IdentitySuffix,

pipeline_pool_gate_test.go

@@ -82,9 +82,10 @@ func TestPipelinePoolSharedByClones(t *testing.T) {
 //   - buffers default to DefaultPipelineBufferSize (not the regular 32 KiB) —
 //     pipeline connections move whole batches per round trip — but never
 //     SHRINK a larger explicitly-configured regular buffer;
-//   - MinIdleConns is forced to 0: the pool is burst capacity, and inheriting
-//     the main pool's MinIdleConns would pre-dial that many pipeline
-//     connections at creation, silently doubling the client's idle footprint;
+//   - MinIdleConns comes from PipelineMinIdleConns (default 0), never from the
+//     main pool's MinIdleConns: the pool is burst capacity, and inheriting it
+//     would pre-dial that many pipeline connections at creation, silently
+//     doubling the client's idle footprint;
 //   - the regular buffers and pool size of the MAIN pool are never touched.
 func TestPipelinePoolOptionsResolution(t *testing.T) {
 	t.Run("buffers default to DefaultPipelineBufferSize", func(t *testing.T) {
@@ -131,6 +132,16 @@ func TestPipelinePoolOptionsResolution(t *testing.T) {
 		}
 	})
 
+	t.Run("PipelineMinIdleConns sets it; negative means 0", func(t *testing.T) {
+		for _, tc := range []struct{ in, want int }{{0, 0}, {2, 2}, {-1, 0}} {
+			opt := &Options{Addr: "127.0.0.1:0", PipelinePoolSize: 4, MinIdleConns: 8, PipelineMinIdleConns: tc.in}
+			opt.init()
+			if po := pipelinePoolOptions(opt); po.MinIdleConns != tc.want {
+				t.Fatalf("PipelineMinIdleConns=%d: pipeline MinIdleConns = %d, want %d", tc.in, po.MinIdleConns, tc.want)
+			}
+		}
+	})
+
 	t.Run("main pool options are untouched", func(t *testing.T) {
 		opt := &Options{Addr: "127.0.0.1:0", PipelinePoolSize: 4, MinIdleConns: 8}
 		opt.init()

redis.go

@@ -423,8 +423,9 @@ type baseClient struct {
 	// operations (Pipeline, TxPipeline and autopipeline batches), created
 	// unconditionally at NewClient/NewFailoverClient — like pubSubPool — with
 	// pipeline-appropriate options (see pipelinePoolOptions): larger buffers,
-	// no pre-dialing, a small connection cap. It is pure burst capacity: an
-	// unused pipeline pool holds zero connections. PipelinePoolSize < 0 opts
+	// no pre-dialing by default (PipelineMinIdleConns), a small connection cap.
+	// It is pure burst capacity: an unused pipeline pool holds zero connections
+	// (a full-duplex engine keeps one idle spare). PipelinePoolSize < 0 opts
 	// out; nil means pipelines use connPool (opt-out, and the internal
 	// Conn/Tx/Sentinel wrappers which never create one). The field is set
 	// before the client is visible to any goroutine and never mutated after,
@@ -725,11 +726,12 @@ func (c *baseClient) poolForConn(cn *pool.Conn) pool.Pooler {
 //     plateaus around 64 KiB and very large buffers can regress it). The
 //     RESP3 minimum clamp applies as on the main pool.
 //   - PoolSize: PipelinePoolSize when set, DefaultPipelinePoolSize otherwise.
-//   - MinIdleConns: always 0. The pipeline pool is burst capacity — its
-//     connections dial on demand and there is nothing to keep warm before
-//     the first pipeline runs. Without this the clone would inherit the
-//     main pool's MinIdleConns and pre-dial that many pipeline connections
-//     at creation, silently doubling a client's idle footprint.
+//   - MinIdleConns: PipelineMinIdleConns (negative means 0), never the main
+//     pool's MinIdleConns. The pipeline pool is burst capacity — by default its
+//     connections dial on demand. Inheriting MinIdleConns would pre-dial that
+//     many pipeline connections at creation, silently doubling a client's idle
+//     footprint. A full-duplex engine raises the floor at runtime
+//     (RaiseMinIdleConns) to keep a warm spare.
 func pipelinePoolOptions(opt *Options) *Options {
 	pipelineOpt := opt.clone()
 	if opt.PipelineReadBufferSize > 0 {
@@ -753,7 +755,9 @@ func pipelinePoolOptions(opt *Options) *Options {
 	} else {
 		pipelineOpt.PoolSize = DefaultPipelinePoolSize
 	}
-	pipelineOpt.MinIdleConns = 0
+	// No pre-dial by default (an unused pipeline pool holds nothing); a negative
+	// value means 0 here and also turns off the full-duplex spare.
+	pipelineOpt.MinIdleConns = max(opt.PipelineMinIdleConns, 0)
 	// Re-resolve MaxConcurrentDials for the pipeline pool. Options.init capped it to
 	// the MAIN pool size, so a pipeline pool larger than the main pool (e.g.
 	// PoolSize:1 with the default 10-slot pipeline pool) would otherwise dial one
@@ -2483,7 +2487,7 @@ func NewClient(opt *Options) *Client {
 	c.pubSubPool = pubSubPool
 
 	// Create the dedicated pipeline pool unconditionally, like pubSubPool: it
-	// is pure burst capacity (no pre-dialing, small cap, larger buffers — see
+	// is pure burst capacity (no pre-dialing by default, small cap, larger buffers — see
 	// pipelinePoolOptions), so an unused pipeline pool holds zero connections
 	// and costs nothing. Pipelines stop competing with regular commands for
 	// main-pool connections; a burst wider than the pool's cap spills back to

ring.go

@@ -147,14 +147,15 @@ type RingOptions struct {
 	// default: 32KiB (32768 bytes)
 	WriteBufferSize int
 
-	// PipelineReadBufferSize, PipelineWriteBufferSize and PipelinePoolSize
-	// configure the separate connection pool used for pipelining on each shard,
-	// with its own (typically larger) buffers. See the same-named fields on
-	// Options for details. Each shard client (a NewClient) creates this pool by
+	// PipelineReadBufferSize, PipelineWriteBufferSize, PipelinePoolSize and
+	// PipelineMinIdleConns configure the separate connection pool used for
+	// pipelining on each shard, with its own (typically larger) buffers. See the
+	// same-named fields on Options for details. Each shard client (a NewClient) creates this pool by
 	// default; set PipelinePoolSize < 0 to opt out (pipelines run on the main pool).
 	PipelineReadBufferSize  int
 	PipelineWriteBufferSize int
 	PipelinePoolSize        int
+	PipelineMinIdleConns    int
 
 	TLSConfig *tls.Config
 	Limiter   Limiter
@@ -263,6 +264,7 @@ func (opt *RingOptions) clientOptions() *Options {
 		PipelineReadBufferSize:  opt.PipelineReadBufferSize,
 		PipelineWriteBufferSize: opt.PipelineWriteBufferSize,
 		PipelinePoolSize:        opt.PipelinePoolSize,
+		PipelineMinIdleConns:    opt.PipelineMinIdleConns,
 
 		TLSConfig: opt.TLSConfig,
 		Limiter:   opt.Limiter,

sentinel.go

@@ -127,14 +127,15 @@ type FailoverOptions struct {
 	// default: 32KiB (32768 bytes)
 	WriteBufferSize int
 
-	// PipelineReadBufferSize, PipelineWriteBufferSize and PipelinePoolSize
-	// configure the separate connection pool used for pipelining, with its own
-	// (typically larger) buffers. See the same-named fields on Options for
-	// details. NewFailoverClient creates this pool by default; set
+	// PipelineReadBufferSize, PipelineWriteBufferSize, PipelinePoolSize and
+	// PipelineMinIdleConns configure the separate connection pool used for
+	// pipelining, with its own (typically larger) buffers. See the same-named
+	// fields on Options for details. NewFailoverClient creates this pool by default; set
 	// PipelinePoolSize < 0 to opt out (pipelines then run on the main pool).
 	PipelineReadBufferSize  int
 	PipelineWriteBufferSize int
 	PipelinePoolSize        int
+	PipelineMinIdleConns    int
 
 	// AutoPipelineOptions is the default config for the client's autopipeliner
 	// faces. See Options.AutoPipelineOptions.
@@ -222,6 +223,7 @@ func (opt *FailoverOptions) clientOptions() *Options {
 		PipelineReadBufferSize:  opt.PipelineReadBufferSize,
 		PipelineWriteBufferSize: opt.PipelineWriteBufferSize,
 		PipelinePoolSize:        opt.PipelinePoolSize,
+		PipelineMinIdleConns:    opt.PipelineMinIdleConns,
 		AutoPipelineOptions:     opt.AutoPipelineOptions,
 
 		DialTimeout:        opt.DialTimeout,
@@ -344,6 +346,7 @@ func (opt *FailoverOptions) clusterOptions() *ClusterOptions {
 		PipelineReadBufferSize:  opt.PipelineReadBufferSize,
 		PipelineWriteBufferSize: opt.PipelineWriteBufferSize,
 		PipelinePoolSize:        opt.PipelinePoolSize,
+		PipelineMinIdleConns:    opt.PipelineMinIdleConns,
 		AutoPipelineOptions:     opt.AutoPipelineOptions,
 
 		DialTimeout:        opt.DialTimeout,
@@ -490,6 +493,7 @@ func setupFailoverConnParams(u *url.URL, o *FailoverOptions) (*FailoverOptions,
 	// Pipeline pool (created by default): allow URL opt-out
 	// (pipeline_pool_size=-1) / tuning, else rejected as unexpected options.
 	o.PipelinePoolSize = q.int("pipeline_pool_size")
+	o.PipelineMinIdleConns = q.int("pipeline_min_idle_conns")
 	o.PipelineReadBufferSize = q.int("pipeline_read_buffer_size")
 	o.PipelineWriteBufferSize = q.int("pipeline_write_buffer_size")
 	o.ConnMaxLifetime = q.duration("conn_max_lifetime")

universal.go

@@ -108,11 +108,12 @@ type UniversalOptions struct {
 
 	// PipelineReadBufferSize / PipelineWriteBufferSize size the dedicated pipeline
 	// pool's per-connection buffers. PipelinePoolSize sizes that pool; a negative
-	// value opts out of the dedicated pipeline pool. See the same fields on
-	// Options for details.
+	// value opts out of the dedicated pipeline pool. PipelineMinIdleConns keeps
+	// idle connections in it. See the same fields on Options for details.
 	PipelineReadBufferSize  int
 	PipelineWriteBufferSize int
 	PipelinePoolSize        int
+	PipelineMinIdleConns    int
 
 	// PoolFIFO uses FIFO mode for each node connection pool GET/PUT (default LIFO).
 	PoolFIFO bool
@@ -267,6 +268,7 @@ func (o *UniversalOptions) Cluster() *ClusterOptions {
 		PipelineReadBufferSize:  o.PipelineReadBufferSize,
 		PipelineWriteBufferSize: o.PipelineWriteBufferSize,
 		PipelinePoolSize:        o.PipelinePoolSize,
+		PipelineMinIdleConns:    o.PipelineMinIdleConns,
 
 		PoolFIFO:              o.PoolFIFO,
 		PoolSize:              o.PoolSize,
@@ -339,6 +341,7 @@ func (o *UniversalOptions) Failover() *FailoverOptions {
 		PipelineReadBufferSize:  o.PipelineReadBufferSize,
 		PipelineWriteBufferSize: o.PipelineWriteBufferSize,
 		PipelinePoolSize:        o.PipelinePoolSize,
+		PipelineMinIdleConns:    o.PipelineMinIdleConns,
 
 		PoolFIFO:              o.PoolFIFO,
 		PoolSize:              o.PoolSize,
@@ -404,6 +407,7 @@ func (o *UniversalOptions) Simple() *Options {
 		PipelineReadBufferSize:  o.PipelineReadBufferSize,
 		PipelineWriteBufferSize: o.PipelineWriteBufferSize,
 		PipelinePoolSize:        o.PipelinePoolSize,
+		PipelineMinIdleConns:    o.PipelineMinIdleConns,
 
 		PoolFIFO:              o.PoolFIFO,
 		PoolSize:              o.PoolSize,