fix(autopipeline): TS.READ positional scan, FD child GC, batch guardrail

Four fixes from PR #4002 review comments:

- isBlockingCmd's ts.read case scanned every argument for "block",
  including args[1]/args[2] (key, timestamp) — TS.READ's two fixed
  positional args before the option section (see TSReadWithArgs). A
  nonblocking TSRead(ctx, "block", ts) call was misclassified as
  blocking and diverted off the ordered pipe, breaking async-face
  ordering (codex — a gap in the earlier STREAMS/BLOCK positional fix,
  da956f24, which covered XREAD/XREADGROUP but not TS.READ). Scan only
  args[3:], mirroring the XREADGROUP fix.

- clusterFDRouter.children was keyed by node *Client and only cleared
  when the parent closes. Topology GC can close a node client (and its
  cached child) while the router stays up; once the cluster's slot map
  stops pointing at that *Client, getOrCreateChild never sees it again,
  so a closed child (its FD engine, submit channel, held connection)
  sat retained in the map for the router's whole life on a cluster that
  scales down or replaces node addresses (cursor bugbot). Sweep stale
  closed entries for OTHER nodes while getOrCreateChild already holds
  the write lock to build a fresh one.

- DefaultPipelinePoolTimeout's GoDoc claimed a residual
  maintnotifications drainer-handoff budget. Traced every acquisition
  against the dedicated pipeline pool (withPipelineConn's borrow, the
  FD engine's session lease) — both use TryGet, never the blocking
  Get, and TryGet's non-wait branch returns before waitForDrainer is
  ever reached. The constant has no live effect there; doc corrected
  (codex). A second, related stale comment in pipelinePoolOptions
  claimed the same wrong "short wait then ErrPoolTimeout" behavior;
  fixed to match the correct comment already 600 lines away in the
  same file.

- MaxBatchBytes defaulted to 0 (unbounded). With FullDuplex, the
  reader cannot start draining replies until the writer finishes
  flushing the WHOLE batch, so a batch with >= 2 large-payload
  commands (e.g. big ECHOs) can deadlock both directions: Redis blocks
  writing an early large reply while the client is still blocked
  writing the rest of the batch, resolved only by WriteTimeout, with
  recovery then possibly replaying an already-executed command
  (codex). MaxBatchBytes now defaults to 128 KiB — a guardrail, not a
  throughput knob; for ordinary small commands it never binds
  (MaxBatchSize's 200-command cap already keeps a typical batch under
  ~15 KiB). There is no "unbounded" setting anymore; pass a
  deliberately large value to opt out. Bumped DefaultPipelineBufferSize
  64 KiB -> 128 KiB alongside it for the matching bufio headroom before
  a write() to a slow-draining peer blocks; the aggregate memory cost
  stays negligible since the pool it backs holds few connections
  (dedicated pipeline pool never pre-dials; full-duplex holds exactly
  one connection per node).

This does not fully close the underlying gap: MaxBatchBytes bounds
REQUEST bytes, not expected REPLY bytes, so a batch of plain GETs
against large values is still exposed regardless of this default. The
real fix — draining inflight incrementally as the writer flushes,
instead of only after the whole batch — is a separate, larger change
to the write/flush protocol and is intentionally out of scope here.

Adds TestClusterFullDuplexEvictsStaleChildOnRebuild,
TestBlockingDetectionTSReadKeyNamedBlock, and
TestAutoPipelineMaxBatchBytesDefaultsToGuardrail. Updates
TestAutoPipelineMaxBatchBytes's uncapped control, which relied on the
old 0-means-unbounded default. All three new/changed fixes verified
against a temporary revert to confirm they actually catch the
regression.

autopipeline.go

@@ -32,14 +32,31 @@ type AutoPipelineOptions struct {
 	// DefaultBlockingAutoPipelineOptions, uses 300).
 	MaxBatchSize int
 
-	// MaxBatchBytes, when > 0, caps a batch by APPROXIMATE payload volume: the
+	// MaxBatchBytes caps a batch by APPROXIMATE payload volume: the
 	// accumulator stops waiting once the queued commands' argument bytes reach
 	// it, so many large values flush as several bounded writes instead of one
 	// huge burst (300 x 64KiB is ~19MB written down one connection before any
 	// reply is read — enough to stall a constrained link past its write
 	// deadline). Like MaxBatchSize it is a soft threshold, not a hard cap.
 	// The estimate counts string/[]byte argument lengths plus a small
-	// per-argument overhead. Default: 0 (no byte cap).
+	// per-argument overhead.
+	//
+	// Default: 128 KiB (unset/<=0 gets this default; there is no "unbounded"
+	// setting — pass a deliberately large value instead). This is a
+	// full-duplex safety guardrail, not a throughput knob: with FullDuplex,
+	// the reader cannot start draining replies until the writer finishes
+	// flushing the WHOLE batch (see autopipeline_fullduplex.go), so a batch
+	// with ≥2 large-payload commands can deadlock both directions — Redis
+	// blocks writing an early large reply while the client is still blocked
+	// writing the rest of the batch, resolved only by WriteTimeout, and
+	// recovery may then replay a command Redis already executed (cursor/codex
+	// on #4002). The cap bounds this for large-REQUEST commands (big ECHO,
+	// large SET values); it does NOT bound expected REPLY size, so a batch of
+	// plain GETs against large values is still exposed regardless of this
+	// setting — closing that gap needs the reader to drain incrementally, not
+	// a byte cap. For ordinary small commands this default rarely binds:
+	// MaxBatchSize's 200-command cap already keeps a typical batch under
+	// ~15 KiB, far below this threshold.
 	MaxBatchBytes int
 
 	// MaxConcurrentBatches is the maximum number of pipeline batches that may
@@ -351,8 +368,9 @@ func numAutoPipelineShards() int {
 func DefaultAutoPipelineOptions() *AutoPipelineOptions {
 	return &AutoPipelineOptions{
 		MaxBatchSize:         200,
-		MaxConcurrentBatches: 1, // ordered by default
-		MaxFlushDelay:        0, // lowest latency; no coalescing wait (batch via in-flight backpressure)
+		MaxBatchBytes:        128 * 1024, // see MaxBatchBytes doc: full-duplex deadlock guardrail, not a throughput knob
+		MaxConcurrentBatches: 1,          // ordered by default
+		MaxFlushDelay:        0,          // lowest latency; no coalescing wait (batch via in-flight backpressure)
 	}
 }
 
@@ -374,6 +392,7 @@ func DefaultAutoPipelineOptions() *AutoPipelineOptions {
 func DefaultBlockingAutoPipelineOptions() *AutoPipelineOptions {
 	return &AutoPipelineOptions{
 		MaxBatchSize:         300,
+		MaxBatchBytes:        128 * 1024, // see MaxBatchBytes doc: full-duplex deadlock guardrail, not a throughput knob
 		MaxConcurrentBatches: 1,
 	}
 }
@@ -1129,6 +1148,14 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 		config.MaxBatchSize = 200
 	}
 
+	if config.MaxBatchBytes <= 0 {
+		// Full-duplex deadlock guardrail, not a throughput knob — see the
+		// MaxBatchBytes field doc. Applies here too so a caller-constructed
+		// config that leaves this zero (rather than going through
+		// DefaultAutoPipelineOptions) still gets the safety net.
+		config.MaxBatchBytes = 128 * 1024
+	}
+
 	if config.MaxConcurrentBatches <= 0 {
 		// Default to an ordered single stream. Callers raise this (with
 		// Unordered:true) to opt into parallel-batch throughput.
@@ -1754,11 +1781,16 @@ func isBlockingCmd(cmd Cmder) bool {
 		// them positionally rather than matching on value.
 		return blockOptionBeforeStreams(args, 4)
 	case "ts.read":
-		// TS.READ has no STREAMS terminator at all, so there is nothing to
-		// stop the scan early for: the key and timestamp are arbitrary
-		// values (e.g. a key literally named "streams") that must not be
-		// treated as one. Just look for the BLOCK keyword anywhere.
-		for _, arg := range args {
+		// TS.READ has no STREAMS terminator, but it DOES have two fixed
+		// positional args before the option section: args[1] is the key and
+		// args[2] is the timestamp (see TSReadWithArgs), and either can be
+		// arbitrary user data (e.g. a key literally named "block"). Scanning
+		// from args[0] would mistake that key for the BLOCK option and divert
+		// a non-blocking TS.READ off the ordered pipe (codex on #4002).
+		if len(args) < 3 {
+			return false
+		}
+		for _, arg := range args[3:] {
 			if internal.ToLower(blockingArgString(arg)) == "block" {
 				return true
 			}

autopipeline_cluster_fd.go

@@ -263,6 +263,21 @@ func (r *clusterFDRouter) getOrCreateChild(nc *Client) (*AutoPipeliner, bool) {
 		_ = child.Close()
 		return nil, true
 	}
+	// Evict stale entries for OTHER node clients while the write lock is
+	// already held: topology GC can close a node client (and its cached
+	// child) while this router stays up, and once the cluster's slot map
+	// stops pointing at that *Client, getOrCreateChild is never called with
+	// it again — nothing else would notice it went stale. Without this sweep
+	// a closed child (and its FD engine, submit channel, held connection)
+	// sits retained in this map for the router's whole life on a cluster that
+	// scales down or replaces node addresses (cursor bugbot on #4002). The
+	// child is already closed (its own node-client close hook got there
+	// first), so this only drops the reference for GC — no Close() needed.
+	for k, v := range r.children {
+		if k != nc && v.IsClosed() {
+			delete(r.children, k)
+		}
+	}
 	// Prefer a live child another goroutine cached first; otherwise store ours.
 	if cached := r.children[nc]; cached != nil && !cached.IsClosed() {
 		child = cached

autopipeline_cluster_fd_internal_test.go

@@ -300,6 +300,101 @@ func TestClusterFullDuplexChildCloseSelfHeals(t *testing.T) {
 	}
 }
 
+// nodeClientFor reports the *Client that owns key's slot in the current
+// topology, without creating or touching any FD child (unlike childFor).
+// Used by eviction tests to find a key that routes to a node different from
+// one already cached.
+func nodeClientFor(t *testing.T, ctx context.Context, r *clusterFDRouter, key string) (*Client, bool) {
+	t.Helper()
+	slot := r.cc.cmdSlot(NewStatusCmd(ctx, "get", key), -1)
+	if slot < 0 {
+		return nil, false
+	}
+	state, err := r.cc.state.Get(ctx)
+	if err != nil {
+		return nil, false
+	}
+	node, err := state.slotMasterNode(slot)
+	if err != nil || node == nil {
+		return nil, false
+	}
+	return node.Client, true
+}
+
+// TestClusterFullDuplexEvictsStaleChildOnRebuild pins that a closed node
+// child for a DIFFERENT node than the one currently being (re)built gets
+// swept out of clusterFDRouter.children. Topology GC can close a node client
+// (and its cached child) while the router stays up; once the cluster stops
+// routing to that *Client, getOrCreateChild is never called with it again, so
+// nothing else would notice it went stale — without the sweep it (and its FD
+// engine, submit channel, held connection) would sit retained in the map for
+// the router's whole life (cursor bugbot on #4002).
+func TestClusterFullDuplexEvictsStaleChildOnRebuild(t *testing.T) {
+	cc := dialClusterFDTest(t)
+	defer cc.Close()
+	ctx := context.Background()
+
+	ap, err := cc.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true})
+	if err != nil {
+		t.Fatalf("AsyncAutoPipelineWithOptions: %v", err)
+	}
+	defer ap.Close()
+
+	// Seed one key to populate its node's child, then hunt for a second tag
+	// that lands on a DIFFERENT node — a first-time build for that second
+	// node is what forces getOrCreateChild's write-lock (sweep) path without
+	// needing to touch the second node's own cache state.
+	const seedKey = "cfd:{evictseed}:k"
+	if err := ap.Set(ctx, seedKey, "1", 0).Err(); err != nil {
+		t.Fatalf("seed SET: %v", err)
+	}
+	ap.clusterFD.mu.RLock()
+	var staleClient *Client
+	var staleChild *AutoPipeliner
+	for c, ch := range ap.clusterFD.children {
+		staleClient, staleChild = c, ch
+	}
+	ap.clusterFD.mu.RUnlock()
+	if staleClient == nil {
+		t.Fatal("no FD child cached after seed SET")
+	}
+
+	var secondKey string
+	for i := 0; i < 50; i++ {
+		k := fmt.Sprintf("cfd:{evicttarget%d}:k", i)
+		nc, ok := nodeClientFor(t, ctx, ap.clusterFD, k)
+		if ok && nc != staleClient {
+			secondKey = k
+			break
+		}
+	}
+	if secondKey == "" {
+		t.Skip("could not find a tag routing to a different master after 50 tries")
+	}
+
+	// Simulate topology GC: close the seed node's child directly, but leave
+	// its (now stale) entry in the map exactly as a real close hook would.
+	if err := staleChild.Close(); err != nil {
+		t.Fatalf("staleChild.Close: %v", err)
+	}
+
+	// First-time build for the second node: write-lock path, sweep runs.
+	if err := ap.Set(ctx, secondKey, "1", 0).Err(); err != nil {
+		t.Fatalf("SET on second node: %v", err)
+	}
+
+	ap.clusterFD.mu.RLock()
+	_, stillPresent := ap.clusterFD.children[staleClient]
+	n := len(ap.clusterFD.children)
+	ap.clusterFD.mu.RUnlock()
+	if stillPresent {
+		t.Error("stale closed child was not evicted from clusterFDRouter.children")
+	}
+	if n != 1 {
+		t.Errorf("clusterFDRouter.children has %d entries, want 1 (only the second node's fresh child)", n)
+	}
+}
+
 // TestClusterFullDuplexGatedOffForReplicaRouting asserts the construction gate:
 // a ClusterClient configured for replica routing (ReadOnly / RouteByLatency /
 // RouteRandomly) must NOT engage cluster FD, because the router only routes to

autopipeline_internal_test.go

@@ -1580,6 +1580,21 @@ func TestBlockingDetectionTSReadKeyNamedStreams(t *testing.T) {
 	}
 }
 
+// TestBlockingDetectionTSReadKeyNamedBlock pins that TS.READ's scan starts
+// AFTER its two fixed positional args (key, timestamp — see TSReadWithArgs),
+// not from args[0]: a key literally named "block" must not be mistaken for
+// the BLOCK option, which would divert a non-blocking TS.READ off the
+// ordered pipe and break ordering for the async face (codex on #4002).
+func TestBlockingDetectionTSReadKeyNamedBlock(t *testing.T) {
+	ctx := context.Background()
+	if isBlockingCmd(NewCmd(ctx, "ts.read", "block", "-", "MAX_COUNT", 10)) {
+		t.Error(`isBlockingCmd(ts.read key="block", no real BLOCK option) = true, want false`)
+	}
+	if !isBlockingCmd(NewCmd(ctx, "ts.read", "block", "-", "BLOCK", 100, 1)) {
+		t.Error(`isBlockingCmd(ts.read key="block" ... BLOCK) = false, want true`)
+	}
+}
+
 // TestDivertRegistrationRacesClose hammers the window between a diverted
 // command's closed check and its registration against Close: every submitted
 // command must end with a definite outcome, and Close must never report

autopipeline_test.go

@@ -3641,6 +3641,39 @@ func TestAutoPipelineClientLevelOptionsApply(t *testing.T) {
 	}
 }
 
+// TestAutoPipelineMaxBatchBytesDefaultsToGuardrail pins that an unset (or
+// explicitly zero) MaxBatchBytes resolves to the 128 KiB full-duplex deadlock
+// guardrail (see the field's GoDoc), not to "unbounded" — the previous
+// behavior. Checked via both preset constructors and the actual resolved
+// config on a constructed autopipeliner, since MaxBatchSize's own default is
+// applied redundantly in three places and MaxBatchBytes now mirrors that.
+func TestAutoPipelineMaxBatchBytesDefaultsToGuardrail(t *testing.T) {
+	const want = 128 * 1024
+
+	if got := redis.DefaultAutoPipelineOptions().MaxBatchBytes; got != want {
+		t.Errorf("DefaultAutoPipelineOptions().MaxBatchBytes = %d, want %d", got, want)
+	}
+	if got := redis.DefaultBlockingAutoPipelineOptions().MaxBatchBytes; got != want {
+		t.Errorf("DefaultBlockingAutoPipelineOptions().MaxBatchBytes = %d, want %d", got, want)
+	}
+
+	c := redis.NewClient(&redis.Options{
+		Addr: apTestAddr(),
+		AutoPipelineOptions: &redis.AutoPipelineOptions{
+			MaxBatchSize: 50, // leave MaxBatchBytes at its zero value
+		},
+	})
+	defer c.Close()
+
+	ap, err := c.AutoPipeline()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := ap.Config().MaxBatchBytes; got != want {
+		t.Fatalf("resolved MaxBatchBytes = %d, want %d (zero must default to the guardrail, not stay unbounded)", got, want)
+	}
+}
+
 // TestAsyncAutoPipelineBlockingCommand covers the readTimeout diversion on the
 // DEFERRED face: BLPOP must bypass the batch (it would stall a whole pipeline)
 // and still deliver its result through the usual accessors.
@@ -4560,7 +4593,10 @@ func TestAutoPipelineMaxBatchBytes(t *testing.T) {
 			t.Fatalf("capped run had a %d-command batch (%v) — cap not enforced", maxBatch, capped)
 		}
 
-		uncapped := run(0)
+		// 0 no longer means "uncapped" (it now gets the 128 KiB default
+		// guardrail — see MaxBatchBytes's GoDoc); opt out explicitly with a
+		// value comfortably above this run's total payload (10 x 64 KiB).
+		uncapped := run(10 * 1024 * 1024)
 		utotal := 0
 		for _, n := range uncapped {
 			utotal += n

options.go

@@ -248,7 +248,7 @@ type Options struct {
 	// PipelinePoolSize is the pool size for the separate pipeline connection pool.
 	// 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 /
+	// DefaultPipelineBufferSize (128 KiB), unless PipelineReadBufferSize /
 	// PipelineWriteBufferSize are set.
 	//
 	// Pipelining typically needs fewer connections than regular operations because
@@ -502,9 +502,19 @@ const DefaultPipelinePoolSize = 10
 // (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
+// throughput plateaus around 64 KiB and gains nothing past ~128 KiB for
+// TYPICAL (small-command) traffic, while very large buffers (>=512 KiB) can
+// regress it.
+//
+// Set to 128 KiB anyway: the extra headroom is not about typical-traffic
+// throughput but about full-duplex's large-payload backpressure guardrail
+// (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.
+const DefaultPipelineBufferSize = 128 * 1024
 
 // DefaultPipelinePoolTimeout is the dedicated pipeline pool's PoolTimeout
 // (pipelinePoolOptions caps the pipeline clone's PoolTimeout at this value but honors
@@ -514,13 +524,17 @@ const DefaultPipelineBufferSize = 64 * 1024
 // in Options and withPipelineConn), never waiting this timeout, and the spilled op
 // then uses the MAIN pool's own PoolTimeout, not this one.
 //
-// Its remaining live effect is the budget for a pipeline connection's maintnotifications
-// drainer handoff: PoolTimeout is one budget for both a pool turn and a drainer handoff
-// (see internal/pool), and TryGet still respects a drainer's bounded claim up to
-// PoolTimeout. So a pipeline connection that needs a handoff gets this short budget
-// rather than the main pool's (tens of seconds) — acceptable because pipeline
-// connections are disposable burst capacity that a burst can spill past anyway. It is
-// deliberately short for the same reason.
+// It currently has NO other live effect either. Every acquisition against the
+// dedicated pipeline pool — withPipelineConn's per-round-trip borrow above, and
+// the full-duplex engine's own session lease (autopipeline_fullduplex.go) — uses
+// TryGet, never the blocking Get. TryGet's non-wait branch returns ErrPoolTryFull
+// at once for BOTH a saturated pool turn and an active maintnotifications drainer
+// claim, before waitForDrainer or this deadline is ever consulted (see
+// ConnPool.getConn/waitTurn in internal/pool). A previous version of this doc
+// claimed a residual drainer-handoff budget; that was inaccurate (codex on #4002)
+// — there is no code path that currently waits out this value. It is kept short
+// anyway in case a future acquisition path, or a caller that obtains the pool via
+// getPipelinePool's exported pool.Pooler interface, uses the blocking Get.
 const DefaultPipelinePoolTimeout = 100 * time.Millisecond
 
 func (opt *Options) init() {

redis.go

@@ -693,12 +693,14 @@ func pipelinePoolOptions(opt *Options) *Options {
 	// 0 the pipeline pool never itself returns ErrPoolExhausted; a burst wider
 	// than PoolSize spills on ErrPoolTimeout instead — see withPipelineConn.)
 	pipelineOpt.MaxActiveConns = 0
-	// Spill, don't queue: when every pipeline connection is busy, fall back to the
-	// main pool after a short wait rather than blocking for the (main) PoolTimeout,
-	// which the clone would otherwise inherit (tens of seconds). withPipelineConn
-	// spills on the resulting ErrPoolTimeout. Cap at DefaultPipelinePoolTimeout but
-	// honor a caller's SHORTER PoolTimeout, so a client tuned to spill faster is not
-	// forced to wait the full default before falling back.
+	// Spill, don't queue: withPipelineConn acquires from the pipeline pool with
+	// TryGet, which never waits out PoolTimeout — a saturated pool returns
+	// ErrPoolTryFull AT ONCE and the caller spills to the main pool immediately
+	// (see withPipelineConn, DefaultPipelinePoolTimeout). This value therefore
+	// has no live effect on that acquisition; it only bounds what PoolTimeout
+	// ends up stored on the pipeline pool's Options, capped here so a caller
+	// tuned to a short PoolTimeout is not silently widened to the (main)
+	// default (tens of seconds), which the clone would otherwise inherit.
 	pipelineOpt.PoolTimeout = DefaultPipelinePoolTimeout
 	if opt.PoolTimeout > 0 && opt.PoolTimeout < DefaultPipelinePoolTimeout {
 		pipelineOpt.PoolTimeout = opt.PoolTimeout