chore(pipeline): Create pipeline pool even without specific buffer passed

.github/workflows/govulncheck.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v7
         with:
-          go-version: "1.26.x"
+          go-version: "stable"
           cache: true
 
       - name: Install govulncheck

autopipeline_test.go

@@ -1874,8 +1874,24 @@ func TestAutoPipelineRetriesOnNetworkError(t *testing.T) {
 	}
 	defer ap.Close()
 
-	// Arm after the handshake: the pooled conn is healthy, so it passes the
-	// pool health check, and the batch's first write dies on the wire.
+	// Warm the DEDICATED PIPELINE POOL connection first: creating the
+	// autopipeliner created the pipeline pool (lazily), and BATCHES run there,
+	// not on the main-pool conn the Ping dialed. Without this warm-up the
+	// batch's first dispatch dials a fresh pipeline conn and the dial count
+	// reads one high for a reason unrelated to the retry. The warm-up must be
+	// a real multi-command batch: a lone command takes the solo fast path
+	// (Process on the main pool) and would not touch the pipeline pool.
+	w1 := ap.Set(ctx, "apr:warm", 1, 0)
+	w2 := ap.Incr(ctx, "apr:warm2")
+	if err := w1.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := w2.Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Arm after the handshake: the pipeline-pool conn is healthy, so it passes
+	// the pool health check, and the batch's first write dies on the wire.
 	dialsBefore := dials.Load()
 	failNextWrite.Store(true)
 
@@ -3927,7 +3943,6 @@ func TestAutoPipelineHookPostNextErrorPartialBatch(t *testing.T) {
 	if err := c.Set(ctx, "pnp:present", "v", 0).Err(); err != nil {
 		t.Fatal(err)
 	}
-	c.AddHook(postNextErrorHook{err: errInjected})
 
 	// Wide flush window so all three commands deterministically land in ONE
 	// pipeline batch (the rule is per-batch: hooks fire per batch).
@@ -3940,6 +3955,23 @@ func TestAutoPipelineHookPostNextErrorPartialBatch(t *testing.T) {
 	}
 	defer ap.Close()
 
+	// Initialize the dedicated pipeline-pool connection BEFORE installing the
+	// error-injecting hook: connection init runs its handshake pipeline through
+	// the client's hook chain (newConn shares hooksMixin), so a hook that
+	// unconditionally injects an error would fail the pipeline conn's init and
+	// this test would measure init poisoning instead of the post-next rule.
+	// Must be a real multi-command batch — a lone command takes the solo fast
+	// path (Process on the main pool) and would not init the pipeline conn.
+	w1 := ap.Set(ctx, "pnp:warm", 1, 0)
+	w2 := ap.Set(ctx, "pnp:warm2", 1, 0)
+	if err := w1.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := w2.Err(); err != nil {
+		t.Fatal(err)
+	}
+	c.AddHook(postNextErrorHook{err: errInjected})
+
 	runWithWatchdog(t, 30*time.Second, func() {
 		// One batch: a hit, a miss (redis.Nil), and a write.
 		hit := ap.Get(ctx, "pnp:present")

commands_test.go

@@ -83,11 +83,27 @@ var _ = Describe("Commands", func() {
 			Expect(cmds[0].Err().Error()).To(authErr)
 
 			stats := rawClient.PoolStats()
-			Expect(stats.Hits).To(Equal(uint32(2)))
-			Expect(stats.Misses).To(Equal(uint32(1)))
-			Expect(stats.Timeouts).To(Equal(uint32(0)))
-			Expect(stats.TotalConns).To(Equal(uint32(1)))
-			Expect(stats.IdleConns).To(Equal(uint32(1)))
+			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
+				// served only the BeforeEach FlushDB.
+				Expect(stats.PipelineStats.Hits).To(Equal(uint32(1)))
+				Expect(stats.PipelineStats.Misses).To(Equal(uint32(1)))
+				Expect(stats.PipelineStats.Timeouts).To(Equal(uint32(0)))
+				Expect(stats.PipelineStats.TotalConns).To(Equal(uint32(1)))
+				Expect(stats.PipelineStats.IdleConns).To(Equal(uint32(1)))
+				Expect(stats.Hits).To(Equal(uint32(0)))
+				Expect(stats.Misses).To(Equal(uint32(1)))
+			} else {
+				// No dedicated pipeline pool: FlushDB dialed the one connection
+				// (miss) and both Pipelined calls reused it (hits).
+				Expect(stats.Hits).To(Equal(uint32(2)))
+				Expect(stats.Misses).To(Equal(uint32(1)))
+				Expect(stats.Timeouts).To(Equal(uint32(0)))
+				Expect(stats.TotalConns).To(Equal(uint32(1)))
+				Expect(stats.IdleConns).To(Equal(uint32(1)))
+			}
 		})
 
 		It("should hello", func() {
@@ -393,8 +409,16 @@ var _ = Describe("Commands", func() {
 				}()
 				pipe.ClientSetInfo(ctx, libInfo)
 			}).To(Panic())
-			// Test setting the default options for libName, libName suffix and libVer
-			clientInfo := rawClient.ClientInfo(ctx).Val()
+			// Test setting the default options for libName, libName suffix and libVer.
+			// CLIENT SETINFO is per-connection state, so read CLIENT INFO through the
+			// same pipeline path that issued the SETINFOs above: when the client has a
+			// dedicated pipeline pool (the autopipeline subject faces create one
+			// lazily), those ran on a pipeline-pool connection and rawClient.ClientInfo
+			// would inspect a different, main-pool connection.
+			infoCmd := pipe.ClientInfo(ctx)
+			_, err = pipe.Exec(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			clientInfo := infoCmd.Val()
 			Expect(clientInfo.LibName).To(ContainSubstring("go-redis(go-redis,"))
 			// Test setting the libName suffix in options
 			opt := redisOptions()

himport_mock_test.go

@@ -356,10 +356,13 @@ func TestHImportLazyReplay(t *testing.T) {
 	ctx := context.Background()
 
 	client := redis.NewClient(&redis.Options{
-		Addr:            srv.addr(),
-		Protocol:        2,
-		PoolSize:        1, // deterministic: every command runs on the same connection
-		DisableIdentity: true,
+		Addr:     srv.addr(),
+		Protocol: 2,
+		PoolSize: 1, // deterministic: every command runs on the same connection
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		DisableIdentity:  true,
 	})
 	defer client.Close()
 
@@ -450,10 +453,13 @@ func TestHImportPipelineRecoversAfterSessionLoss(t *testing.T) {
 	ctx := context.Background()
 
 	client := redis.NewClient(&redis.Options{
-		Addr:            srv.addr(),
-		Protocol:        2,
-		PoolSize:        1,
-		DisableIdentity: true,
+		Addr:     srv.addr(),
+		Protocol: 2,
+		PoolSize: 1,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		DisableIdentity:  true,
 	})
 	defer client.Close()
 
@@ -500,11 +506,14 @@ func TestHImportPipelineReissueTransportErrorScoped(t *testing.T) {
 	ctx := context.Background()
 
 	client := redis.NewClient(&redis.Options{
-		Addr:            srv.addr(),
-		Protocol:        2,
-		PoolSize:        1,
-		MaxRetries:      -1,
-		DisableIdentity: true,
+		Addr:     srv.addr(),
+		Protocol: 2,
+		PoolSize: 1,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		MaxRetries:       -1,
+		DisableIdentity:  true,
 	})
 	defer client.Close()
 
@@ -553,10 +562,13 @@ func TestHImportTxSurfacesSessionLoss(t *testing.T) {
 	ctx := context.Background()
 
 	client := redis.NewClient(&redis.Options{
-		Addr:            srv.addr(),
-		Protocol:        2,
-		PoolSize:        1,
-		DisableIdentity: true,
+		Addr:     srv.addr(),
+		Protocol: 2,
+		PoolSize: 1,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		DisableIdentity:  true,
 	})
 	defer client.Close()
 
@@ -602,10 +614,13 @@ func TestHImportLazyDiscardPropagation(t *testing.T) {
 	ctx := context.Background()
 
 	client := redis.NewClient(&redis.Options{
-		Addr:            srv.addr(),
-		Protocol:        2,
-		PoolSize:        2,
-		DisableIdentity: true,
+		Addr:     srv.addr(),
+		Protocol: 2,
+		PoolSize: 2,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		DisableIdentity:  true,
 	})
 	defer client.Close()
 
@@ -708,8 +723,11 @@ func TestHImportRingFanOut(t *testing.T) {
 			"shard1": srv1.addr(),
 			"shard2": srv2.addr(),
 		},
-		PoolSize:        1,
-		DisableIdentity: true,
+		PoolSize: 1,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		DisableIdentity:  true,
 	})
 	defer ring.Close()
 
@@ -790,11 +808,14 @@ func TestHImportInjectedPrepareWithPushNotification(t *testing.T) {
 	ctx := context.Background()
 
 	client := redis.NewClient(&redis.Options{
-		Addr:            srv.addr(),
-		Protocol:        3,
-		PoolSize:        1,
-		MaxRetries:      -1, // fail BOOM fast; the injected PREPARE needs no retries
-		DisableIdentity: true,
+		Addr:     srv.addr(),
+		Protocol: 3,
+		PoolSize: 1,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
+		MaxRetries:       -1, // fail BOOM fast; the injected PREPARE needs no retries
+		DisableIdentity:  true,
 		// The mock is not a real cluster; keep maintenance-notification
 		// machinery out of the connection lifecycle.
 		MaintNotificationsConfig: &maintnotifications.Config{Mode: maintnotifications.ModeDisabled},
@@ -847,6 +868,9 @@ func TestHImportPipelineInjectedReplyFailureStampsBatch(t *testing.T) {
 		Addr:     srv.addr(),
 		Protocol: 2,
 		PoolSize: 1,
+		// This file choreographs exact per-connection sequences (armed booms,
+		// session counts) on the MAIN pool; keep pipelines there too.
+		PipelinePoolSize: -1,
 		// Exhaust the budget on the first attempt: stamping must not
 		// depend on a later attempt reaching the read path.
 		MaxRetries:      -1,

options.go

@@ -246,15 +246,35 @@ type Options struct {
 	PipelineWriteBufferSize int
 
 	// PipelinePoolSize is the pool size for the separate pipeline connection pool.
-	// Only used if PipelineReadBufferSize or PipelineWriteBufferSize is set.
+	// Setting this alone still sizes the (now always-created) dedicated pipeline
+	// pool; its buffers default to the larger of the regular buffer size and
+	// DefaultPipelineBufferSize (64 KiB), unless PipelineReadBufferSize /
+	// PipelineWriteBufferSize are set.
 	//
 	// Pipelining typically needs fewer connections than regular operations because
 	// batching reduces connection contention. A smaller pool saves memory while
 	// maintaining high throughput.
 	//
-	// If not set (0), defaults to 10 connections.
+	// 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
+	// concurrent pipelines wider than the cap spills to the main pool after a
+	// short wait (DefaultPipelinePoolTimeout) rather than queueing for the full
+	// PoolTimeout. Its connections use DefaultPipelineBufferSize buffers unless
+	// the pipeline buffer sizes are set explicitly. It does not inherit
+	// MaxActiveConns: rather than the ~2x total ceiling that inheriting it
+	// verbatim would allow, the pipeline pool adds at most PipelinePoolSize
+	// connections on top of the main pool's MaxActiveConns (so the effective
+	// ceiling is MaxActiveConns + PipelinePoolSize — a small, bounded addition),
+	// and the main pool the burst spills to still enforces MaxActiveConns.
 	//
-	// default: 10
+	// Set to a negative value to opt out of the dedicated pool entirely:
+	// pipelines then run on the main pool, as they did before the pool
+	// existed.
+	//
+	// default: DefaultPipelinePoolSize (10) connections
 	PipelinePoolSize int
 
 	// AutoPipelineOptions is the default config for BOTH autopipeliner faces:
@@ -452,6 +472,37 @@ const (
 	CSCStrategySharedTracking CSCStrategy = iota
 )
 
+// 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.
+const DefaultPipelinePoolSize = 10
+
+// DefaultPipelineBufferSize is the per-connection read/write buffer size for
+// the dedicated pipeline pool when no explicit pipeline buffer size is set
+// (the larger of this and the regular buffer size is used). Pipeline
+// connections move whole batches per round trip, so they earn bigger buffers
+// than regular per-command traffic: measured on the autopipeline engine,
+// throughput plateaus around 64 KiB and gains nothing past ~128 KiB, while
+// very large buffers (>=512 KiB) can regress it.
+const DefaultPipelineBufferSize = 64 * 1024
+
+// DefaultPipelinePoolTimeout bounds how long a pipeline waits for a pipeline-pool
+// connection before spilling to the main pool. The pipeline pool is burst
+// capacity, so when every one of its connections is busy a further pipeline
+// should fall back to the main pool promptly rather than queue for the full
+// (main) PoolTimeout, which can be tens of seconds. It is deliberately short:
+// staying under it costs a little extra latency on a saturated pipeline pool
+// (the spill), never correctness. See pipelinePoolOptions / withPipelineConn.
+//
+// Note: PoolTimeout is also the budget for a connection's drainer handoff
+// (maintnotifications), so a pipeline connection that needs a handoff gets this
+// short budget rather than the main pool's — acceptable because pipeline
+// connections are disposable burst capacity that a burst can spill past anyway.
+const DefaultPipelinePoolTimeout = 100 * time.Millisecond
+
 func (opt *Options) init() {
 	if opt.Addr == "" {
 		opt.Addr = "localhost:6379"
@@ -497,6 +548,7 @@ func (opt *Options) init() {
 	if opt.PoolSize == 0 {
 		opt.PoolSize = 10 * runtime.GOMAXPROCS(0)
 	}
+
 	if opt.MaxConcurrentDials <= 0 {
 		opt.MaxConcurrentDials = opt.PoolSize
 	} else if opt.MaxConcurrentDials > opt.PoolSize {
@@ -862,6 +914,12 @@ func setupConnParams(u *url.URL, o *Options) (*Options, error) {
 	o.MaxIdleConns = q.int("max_idle_conns")
 	o.MaxActiveConns = q.int("max_active_conns")
 	o.MaxConcurrentDials = q.int("max_concurrent_dials")
+	// Pipeline pool (created by default): allow URL-configured clients to opt out
+	// (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.PipelineReadBufferSize = q.int("pipeline_read_buffer_size")
+	o.PipelineWriteBufferSize = q.int("pipeline_write_buffer_size")
 	if q.has("conn_max_idle_time") {
 		o.ConnMaxIdleTime = q.duration("conn_max_idle_time")
 	} else {

options_test.go

@@ -607,3 +607,35 @@ func TestUniversalOptionsSimpleCopiesClientSideCache(t *testing.T) {
 		t.Fatal("Simple did not copy ClientSideCacheStrategy")
 	}
 }
+
+// TestParseURLPipelinePoolOptions verifies the URL parsers accept the pipeline
+// pool settings — notably pipeline_pool_size=-1 to opt out of the now-default
+// dedicated pipeline pool — instead of rejecting them as unexpected options
+// (#3959). Covers ParseURL, ParseClusterURL and ParseFailoverURL.
+func TestParseURLPipelinePoolOptions(t *testing.T) {
+	const q = "pipeline_pool_size=-1&pipeline_read_buffer_size=131072&pipeline_write_buffer_size=65536"
+
+	o, err := ParseURL("redis://localhost:6379?" + q)
+	if err != nil {
+		t.Fatalf("ParseURL: %v", err)
+	}
+	if o.PipelinePoolSize != -1 || o.PipelineReadBufferSize != 131072 || o.PipelineWriteBufferSize != 65536 {
+		t.Fatalf("ParseURL pipeline opts: size=%d rbuf=%d wbuf=%d", o.PipelinePoolSize, o.PipelineReadBufferSize, o.PipelineWriteBufferSize)
+	}
+
+	co, err := ParseClusterURL("redis://localhost:6379?" + q)
+	if err != nil {
+		t.Fatalf("ParseClusterURL: %v", err)
+	}
+	if co.PipelinePoolSize != -1 || co.PipelineReadBufferSize != 131072 || co.PipelineWriteBufferSize != 65536 {
+		t.Fatalf("ParseClusterURL pipeline opts: size=%d rbuf=%d wbuf=%d", co.PipelinePoolSize, co.PipelineReadBufferSize, co.PipelineWriteBufferSize)
+	}
+
+	fo, err := ParseFailoverURL("redis://localhost:6379?master_name=mymaster&" + q)
+	if err != nil {
+		t.Fatalf("ParseFailoverURL: %v", err)
+	}
+	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)
+	}
+}