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

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

options.go

@@ -246,15 +246,27 @@ 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 is enough to create the dedicated pipeline pool; the
+	// pipeline buffer sizes then default to ReadBufferSize/WriteBufferSize.
 	//
 	// 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 back to the main pool
+	// instead of queueing. Its connections use DefaultPipelineBufferSize
+	// buffers unless the pipeline buffer sizes are set explicitly.
 	//
-	// 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: 0 (= DefaultPipelinePoolSize connections)
 	PipelinePoolSize int
 
 	// AutoPipelineOptions is the default config for BOTH autopipeliner faces:
@@ -452,6 +464,23 @@ 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
+
 func (opt *Options) init() {
 	if opt.Addr == "" {
 		opt.Addr = "localhost:6379"
@@ -497,6 +526,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 {