feat(autopipeline): run N full-duplex engines on one client (#4022)

* feat(autopipeline): spike N full-duplex engines per client

SPIKE, not for merge. Answers whether the ~2x measured for
"multi-connection full duplex" is a property of N engines or an artifact
of the benchmark harness, which fakes multi-conn by building N separate
*redis.Client instances and round-robining across them. That holds six
sockets per client to do I/O on one -- 48 for eight wires -- and keeps
the routing outside the library, so key routing could not be tested.

NumShards>1 under FullDuplex now means N engines on ONE client: one pool,
one hook chain, one invalidation stream, N held wires. Commands route by
key hash, so same-key commands keep one wire and per-key order holds
without an Unordered opt-in. Keyless commands round-robin. A whole
FDPipelined batch pins to one engine, since splitting it would lose the
internal order a pipeline exists to guarantee.

The change is TWO-SITED, which is the finding. Relaxing the Validate
rejection alone silently turns full duplex OFF: fdOn is separately gated
on nShards == 1, so NumShards>1 fell through to half-duplex round-robin
shards, which are unordered, and a same-key write-then-read returned
redis.Nil. That is exactly what the deleted error warned about ("reject
it rather than silently falling back to half-duplex"). Both the
validation and the construction gate have to move together, with
nShards forced back to 1 so the shard array stays flusherless.

Measured at 1024 callers, 10 commands per Exec, one binary:

  proxy, 8 clients    1,686,080 ops/s   48 held sockets
  in-client, 8        1,791,369 ops/s   12 held sockets

Known gap before this could ship: engines lease from the pipeline pool
and DefaultPipelinePoolSize is 10, so more than ~10 engines spill to the
main pool. A real feature must size that pool from the engine count.
cmdFirstKey is also deliberately crude (args[1]) rather than using
declared key positions.

* refactor(autopipeline): document FD sharding, guard the pool

Drops the exploratory framing from the multi-connection full-duplex code
and closes two of its three known gaps.

Key extraction no longer assumes args[1]. It calls
cmdFirstKeyPosWithInfo(cmd, nil), the same helper client-side caching
uses, so keyless commands, the static command table, and eval/evalsha
variants whose key position depends on the runtime numkeys argument all
route correctly instead of hashing a non-key argument.

Engine count is now checked against the pipeline pool. Each engine holds
one pooled connection for as long as it runs, and DefaultPipelinePoolSize
is 10, so 11 or more engines left the surplus permanently spilling onto
the main pool and competing with ordinary commands. Construction fails
with a message naming PipelinePoolSize instead.

The two-sited rule is now stated at both sites: relaxing the Validate
rejection without relaxing the fdOn gate leaves full duplex silently OFF
for NumShards>1, so commands fall through to unordered half-duplex shards
and a same-key write-then-read can read a stale value. Tests cover it
from both directions -- reply correlation and per-key ordering across
eight engines, and Config() reporting FullDuplex true rather than having
fallen back.

Still open, and why this stays draft: NumShards:0 means one engine and
sizing is not automatic. A fixed count above 1 costs low-concurrency
callers about 30% (eight engines at eight callers), so the count should
be derived from observed load, roughly one wire per 256-512 callers.

* fix(autopipeline): fix FD engine routing edges

- The NumShards>1 ordering check exempted any requested FullDuplex.
  Without a pipeline pool full duplex does not engage and construction
  falls back to round-robin shards, which need Unordered. The exemption
  now follows the effective state.
- Keys are read with stringArg, as Ring does, so []byte, *string and
  numeric keys route like the string form, and an empty key routes by
  hash instead of round-robin.
- Engine index uses uint32 modulo; converting to int first goes
  negative on 32-bit platforms.
- Document that N engines keep order per first key only, so multi-key
  commands are not ordered across engines, and that NumShards>1 under
  full duplex is new, so the mode is opt-in.

* docs(autopipeline): FD engine ordering and window

Document that with NumShards > 1 keyless commands, including ones that
change every key (FLUSHDB, SWAPDB), round-robin and are not ordered
against keyed commands; that keys route through stringArg like Ring and
the cluster slot hash; and that FullDuplexWindow applies per engine.

* docs(autopipeline): pool check is per autopipeliner

The NumShards pipeline-pool check counts only the autopipeliner being
built. The blocking and async faces and WithTimeout clones share one
pipeline pool, so it must hold the sum of their engines. Say so; a
client-wide engine count is a follow-up.

* fix(command): route raw FCALL by its first key

FCALL has the EVAL layout (name, function, numkeys, keys...), but the key
resolver only special-cased EVAL. The typed FCall sets its key position;
a raw Do/Process FCALL fell back to position 1 and hashed the function
name. That misrouted it on Ring (wrong shard), on cluster (a MOVED away)
and across full-duplex engines (a following GET of its key could run
first). A zero-key FCALL now resolves as keyless, like a zero-key EVAL.

* docs(autopipeline): document multi-engine NumShards

The exported FullDuplex and NumShards docs still described a single
full-duplex stream and said NumShards > 1 needs Unordered. Describe the
multi-engine mode: engines routed by first key, per-first-key order
only, per-engine window, and the pipeline pool sizing.

* fix(autopipeline): keep FD batches on their keys' engine

With several full-duplex engines, FDPipelined sent the whole batch to
its first command's engine. A later command in the batch whose key
hashes elsewhere could then run ahead of an earlier, unawaited command
for that key on that key's engine.

A batch now rides an engine only when every keyed command hashes to it
(keyless commands ride along). Otherwise FDPipelined returns
ErrFDPipelineSpansEngines and Pipeline() runs an ordinary pipeline.

* docs(autopipeline): note key types in engine routing

With several full-duplex engines a command is routed by a hash of its
first key in string form (fmt for non-string arguments, as Ring does),
so the same Redis key passed as different Go types (true and "1") can
land on different engines. Say so on NumShards.

* test(autopipeline): use the configured test address

The full-duplex shard tests dialed 127.0.0.1:6379 directly and skipped
under the REDIS_PORT override. Use apTestAddr and internalTestRedisAddr
like the other autopipeline tests.

* fix(command): route raw XGROUP and ZUNION by key

The raw-key resolver fell back to args[1] for XGROUP, whose key follows
the subcommand, and for ZUNION, ZINTER and ZDIFF, whose key follows
numkeys. With several full-duplex engines a raw command could then run
on a different engine from earlier writes to the same key, and Ring
picked the wrong shard. The typed constructors already set position 2.

* fix(command): route raw keyed commands by key

The raw-key resolver fell back to args[1] for SDIFFCARD, SUNIONCARD,
TS.NRANGE and TS.NREVRANGE (key after numkeys) and HIMPORT SET (key
after the subcommand), where the typed constructors set position 2.

HSCAN, SSCAN and ZSCAN with a hash tag in MATCH routed by the pattern,
a hint meant for the keyless SCAN. The key decides the node: a cluster
scan took a MOVED hop, a Ring scan read the wrong shard, and a
multi-engine full-duplex scan could pass earlier writes to its key.

* docs(autopipeline): note the cost of FD engines

NumShards is a fixed engine count, and several engines fragment small
batches. Give the measured trade on NumShards: 8 engines were 30%
slower than 1 at 8 callers, broke even at about 128, and were 2.2x
faster at 1024.

autopipeline.go

@@ -97,9 +97,11 @@ type AutoPipelineOptions struct {
 	// Also honored by the full-duplex writer, where it is gated on in-flight
 	// depth so it never taxes low-concurrency callers (see fdAccumMinFor).
 	//
-	// Honored on the ordered (Unordered:false, MaxConcurrentBatches<=1),
-	// single-shard face of a standalone *Client that has a pipeline pool — BOTH the
-	// deferred (AsyncAutoPipeline) and the blocking (AutoPipeline) face. A SINGLE
+	// Honored on the ordered (Unordered:false, MaxConcurrentBatches<=1) face of a
+	// standalone *Client that has a pipeline pool — BOTH the deferred
+	// (AsyncAutoPipeline) and the blocking (AutoPipeline) face. NumShards > 1
+	// runs that many full-duplex engines, each on its own held connection; see
+	// NumShards for the ordering that mode keeps. A SINGLE
 	// blocking caller gains nothing: it has one command in flight, so there is
 	// nothing to overlap, and it still pays the held connection and goroutine
 	// overhead; the win needs MANY concurrent blocking callers, whose commands then
@@ -208,6 +210,11 @@ type AutoPipelineOptions struct {
 	// memory until a stalled peer makes in-flight grow. Only used when FullDuplex is
 	// set; 0 means the default (65536, covering ~50ms links at ~1.3M ops/s) and a
 	// negative value is rejected by Validate.
+	//
+	// The window is PER ENGINE. With NumShards > 1 each full-duplex engine gets
+	// its own window and submit queue of this size, so the client-wide bound is
+	// NumShards × FullDuplexWindow in flight, plus as many queued. Size it for
+	// one wire; to cap the total, divide by NumShards.
 	FullDuplexWindow int
 
 	// FullDuplexIdleTimeout is how long the held full-duplex connection may sit
@@ -260,6 +267,29 @@ type AutoPipelineOptions struct {
 	// max(NumShards, MaxConcurrentBatches) — and because shards flush
 	// concurrently, NumShards > 1 on the deferred (async) face requires
 	// Unordered: true (construction fails otherwise).
+	//
+	// With FullDuplex active on a standalone *Client, NumShards means something
+	// else: the number of full-duplex engines, each holding one pipeline-pool
+	// connection with its own window (see FullDuplexWindow). More engines pay
+	// off only with many concurrent callers: measured with 10-command
+	// pipelines, 8 engines were 30% slower than 1 at 8 callers, broke even at about
+	// 128, and 2.2x faster at 1024. Commands are routed
+	// to an engine by a hash of their first key, so Unordered is not required,
+	// but order holds only between commands that share a first key. The key is
+	// hashed in its string form (fmt for non-string arguments, as Ring does), so
+	// pass a key as the same Go type everywhere: true and "1" are the same Redis
+	// key but can land on different engines. Commands
+	// that touch other keys after the first one (COPY, RENAME, MSET, multi-key
+	// DEL, EVAL/FCALL with several keys) and keyless commands (including
+	// FLUSHDB and SWAPDB) are not ordered against the rest; await the first
+	// future before submitting a dependent one, or keep NumShards at 1, which
+	// keeps the whole submit order. A Pipeline() rides the FD wire only when all
+	// its keyed commands hash to one engine; otherwise it runs as an ordinary
+	// pipeline on a pooled connection, not ordered against unawaited commands on
+	// the engines. The pipeline pool must hold at least
+	// NumShards connections for every full-duplex autopipeliner in use
+	// (construction checks this one). If full duplex cannot engage (no
+	// pipeline pool), the Unordered rule above applies.
 	NumShards int
 
 	// MaxFlushDelay is the maximum delay after flushing before checking for more commands.
@@ -413,10 +443,16 @@ func (cfg *AutoPipelineOptions) Validate() error {
 		// half-duplex. contentSharded is exempt: that flag is set by the CLUSTER
 		// wiring (never by users), where the silent fallback IS the documented
 		// behavior, since the options type cannot see the client type.
-		if cfg.NumShards > 1 && !cfg.contentSharded {
-			return fmt.Errorf("redis: AutoPipelineOptions.FullDuplex requires NumShards<=1 "+
-				"(one held FIFO connection is a single stream); got %d", cfg.NumShards)
-		}
+		// NumShards>1 under FullDuplex means N engines, each holding its own
+		// connection, routed by key hash so per-key order still holds (see
+		// autopipeline_fd_shards.go). It is therefore no longer rejected here.
+		//
+		// This is one half of a TWO-SITED rule: the fdOn gate in
+		// newAutoPipeliner must relax with it. Relaxing only this check leaves
+		// full duplex silently OFF for NumShards>1 — commands fall through to
+		// the half-duplex shards, which are unordered, and a same-key
+		// write-then-read can read the stale value.
+		_ = cfg.contentSharded
 	}
 	if cfg.MaxConcurrentBatches > 1 && !cfg.Unordered {
 		return fmt.Errorf("redis: AutoPipelineOptions.MaxConcurrentBatches=%d requires Unordered:true "+
@@ -849,6 +885,12 @@ type AutoPipeliner struct {
 	// submit() streams on one held connection instead of the sharded batch queue
 	// and no shard flusher is started. See autopipeline_fullduplex.go.
 	fd *fdEngine
+	// fds holds every full-duplex engine when NumShards>1 puts several held
+	// connections behind one autopipeliner (see autopipeline_fd_shards.go).
+	// fds[0] is always fd, so every existing `ap.fd != nil` check still reads as
+	// "full duplex is on" and the single-engine path is unchanged.
+	fds  []*fdEngine
+	fdRR atomic.Uint32 // round-robin cursor, keyless commands only
 	// clusterFD, when non-nil, runs ordered full-duplex natively on a
 	// *ClusterClient by routing each command to a per-node FD child autopipeliner
 	// (one held connection per master). Mutually exclusive with fd and with the
@@ -1183,7 +1225,33 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 	// per command, and Submit is rejected there), as is cluster slot sharding
 	// (contentSharded: same-key commands always land in the same shard, so
 	// per-key order holds).
-	if config.NumShards > 1 && !config.Unordered && !blocking && !config.contentSharded {
+	//
+	// Full duplex is exempt for the SAME reason contentSharded is: its engines
+	// are chosen by key hash, so same-key commands always land on one wire and
+	// per-key order holds without an Unordered opt-in. The exemption follows
+	// the EFFECTIVE state (fdOn), not the requested FullDuplex: on a client
+	// without a pipeline pool full duplex does not engage, construction falls
+	// back to the round-robin shards below, and those need the opt-in.
+	//
+	// Ordered full-duplex: the ordered single-shard face on a standalone *Client
+	// with a pipeline pool, async or blocking. When on, submit() streams on one
+	// held connection and no shard flusher runs. The blocking face needs nothing
+	// extra: submit's fd branch skips setReady (the blocking contract) and
+	// processBlocking Waits on the returned batch, as for a half-duplex enqueue.
+	var fdClient *Client
+	fdOn := false
+	// NOT gated on nShards: NumShards>1 selects several engines rather than
+	// disabling full duplex. This is the second half of the two-sited rule
+	// noted in Validate — while this line also required nShards==1, a
+	// NumShards>1 caller got half-duplex round-robin shards instead of the
+	// ordered engine it asked for, silently.
+	if config.FullDuplex && !config.Unordered && config.MaxConcurrentBatches <= 1 {
+		if c, ok := pipeliner.(*Client); ok && c.getPipelinePool() != nil {
+			fdOn, fdClient = true, c
+		}
+	}
+	if config.NumShards > 1 && !config.Unordered && !blocking &&
+		!config.contentSharded && !fdOn {
 		return nil, fmt.Errorf(
 			"redis: AutoPipelineOptions.NumShards=%d requires Unordered:true on the deferred (async) face "+
 				"(commands are distributed round-robin across shards, which flush concurrently and do not preserve submit order)",
@@ -1296,17 +1364,36 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 		perShard = 1
 		remainder = 0
 	}
-	// Ordered full-duplex: the ordered single-shard face on a standalone *Client
-	// with a pipeline pool, async or blocking. When on, submit() streams on one
-	// held connection and no shard flusher runs. The blocking face needs nothing
-	// extra: submit's fd branch skips setReady (the blocking contract) and
-	// processBlocking Waits on the returned batch, as for a half-duplex enqueue.
-	var fdClient *Client
-	fdOn := false
-	if config.FullDuplex && !config.Unordered && config.MaxConcurrentBatches <= 1 && nShards == 1 {
-		if c, ok := pipeliner.(*Client); ok && c.getPipelinePool() != nil {
-			fdOn, fdClient = true, c
+	// fdOn was decided above, before the ordering check. Engines are
+	// flusherless, so nShards is forced to 1 below to keep enqueue's shard
+	// indexing safe; the engine count lives in ap.fds.
+	if fdOn {
+		// Every engine holds one connection leased from the pipeline pool for
+		// as long as it runs, so more engines than that pool can hold would
+		// leave the surplus permanently spilling to the main pool — quietly
+		// competing with ordinary commands instead of pipelining. Fail loudly
+		// instead: the caller can raise PipelinePoolSize or ask for fewer
+		// engines. DefaultPipelinePoolSize is 10, so this bites at 11+.
+		//
+		// The check is per autopipeliner. The blocking and async faces and
+		// WithTimeout clones each run their own engines on the SAME pipeline
+		// pool, so the pool must hold the sum of their NumShards; two faces at
+		// NumShards 8 pass here against a pool of 10, and the surplus spills.
+		// A client-wide engine count is a follow-up; until then size
+		// PipelinePoolSize for every full-duplex autopipeliner in use.
+		if n := fdShardCount(config); n > 1 {
+			if pp := fdClient.getPipelinePool(); pp != nil && n > pp.Size() {
+				return nil, fmt.Errorf(
+					"redis: AutoPipelineOptions.NumShards=%d needs a pipeline pool of at "+
+						"least %d connections (each full-duplex engine holds one), but "+
+						"PipelinePoolSize gives %d; raise Options.PipelinePoolSize or lower "+
+						"NumShards", n, n, pp.Size())
+			}
 		}
+		// One flusherless shard, exactly as the single-engine path does: no
+		// half-duplex flusher runs and enqueue never does a %0, even though
+		// submit routes past it entirely.
+		nShards = 1
 	}
 	// Publish the EFFECTIVE full-duplex state, not the requested one: FullDuplex
 	// engages on a standalone *Client with a pipeline pool (fdOn) or on a
@@ -1362,9 +1449,18 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 	}
 
 	if fdOn {
-		ap.fd = newFDEngine(ap, fdClient)
-		ap.wg.Add(1)
-		go ap.fd.run()
+		// NumShards>1 runs N engines on THIS client, each leasing its own
+		// connection from the pipeline pool. fds[0] is fd, so nothing
+		// downstream needs to know the difference when N==1.
+		n := fdShardCount(config)
+		ap.fds = make([]*fdEngine, 0, n)
+		for i := 0; i < n; i++ {
+			e := newFDEngine(ap, fdClient)
+			ap.fds = append(ap.fds, e)
+			ap.wg.Add(1)
+			go e.run()
+		}
+		ap.fd = ap.fds[0]
 	}
 	if clusterFDOn {
 		ap.clusterFD = newClusterFDRouter(ap, clusterFDCC, config, blocking)
@@ -1938,7 +2034,7 @@ func (ap *AutoPipeliner) submit(ctx context.Context, cmd Cmder) AutoFuture {
 		// Ordered full-duplex: stream on one held connection. enqueue's async
 		// setReady is replicated here since we bypass it. ctx is threaded so a
 		// per-command process-hook host can parent its span correctly.
-		b := ap.fd.submit(ctx, cmd)
+		b := ap.fdFor(cmd).submit(ctx, cmd)
 		if !ap.blocking {
 			cmd.setReady(b)
 		}
@@ -3402,8 +3498,8 @@ func (ap *AutoPipeliner) Len() int {
 	// commands are buffered behind a backpressured/stalled FD writer, and callers
 	// using Len() for monitoring or local backpressure lose the signal in
 	// FullDuplex mode.
-	if ap.fd != nil {
-		total += ap.fd.q.depth()
+	for _, e := range ap.fds {
+		total += e.q.depth()
 	}
 	// Cluster full-duplex accepts commands onto per-node FD children, not the
 	// shard queues; include their backlog for the same monitoring reason.
@@ -3540,7 +3636,8 @@ func (ap *AutoPipeliner) fdPipelineExec(ctx context.Context, cmds []Cmder) error
 	switch {
 	case errors.Is(err, ErrFDPipelineDiverts),
 		errors.Is(err, ErrFDPipelineUnavailable),
-		errors.Is(err, ErrFDPipelineTooLarge):
+		errors.Is(err, ErrFDPipelineTooLarge),
+		errors.Is(err, ErrFDPipelineSpansEngines):
 		// Not a command failure: the batch is simply not eligible.
 		ineligible = true
 	case len(cmds) > 0 && remaining >= 0 && !cmdsContainNoRetry(cmds):

autopipeline_fd_pipeline.go

@@ -52,6 +52,14 @@ var ErrFDPipelineUnavailable = errors.New(
 var ErrFDPipelineDiverts = errors.New(
 	"redis: FDPipelined got a command that must be diverted off the full-duplex pipe")
 
+// ErrFDPipelineSpansEngines is returned when NumShards > 1 runs several
+// full-duplex engines and the batch's keyed commands hash to different ones.
+// The batch can only ride one wire, and keeping per-first-key order needs each
+// command on its own key's engine, so it is refused rather than split.
+// Pipeline() falls back to an ordinary pipeline for such a batch.
+var ErrFDPipelineSpansEngines = errors.New(
+	"redis: FDPipelined batch has keys on different full-duplex engines")
+
 // FDPipelined submits cmds as one contiguous batch on the full-duplex
 // connection and blocks until every reply has landed. It returns the first
 // command error, Nil included, the same as Pipeline.Exec.
@@ -246,7 +254,16 @@ func (ap *AutoPipeliner) fdPipelined(ctx context.Context, cmds []Cmder) (fdPipeR
 			return fdPipeResult{}, ErrFDPipelineDiverts
 		}
 	}
-	batches, err := ap.fd.submitBatch(ctx, cmds)
+	// The WHOLE batch goes to ONE engine: a pipeline split across engines would
+	// lose its internal order, which is the one thing a pipeline guarantees. That
+	// engine must also be the one every keyed command hashes to, or a command
+	// could run ahead of an earlier, unawaited command for the same key on that
+	// key's engine. A batch that spans engines is refused.
+	e, ok := ap.fdForBatch(cmds)
+	if !ok {
+		return fdPipeResult{}, ErrFDPipelineSpansEngines
+	}
+	batches, err := e.submitBatch(ctx, cmds)
 	if err != nil {
 		return fdPipeResult{}, err
 	}

autopipeline_fd_pipeline_test.go

@@ -463,3 +463,57 @@ func TestFDPipelineRefusesRedirectAwareEngine(t *testing.T) {
 		t.Fatalf("get: v=%q err=%v", v, err)
 	}
 }
+
+// With several engines, a batch may ride one only if every keyed command in it
+// hashes to that engine: an earlier, unawaited command for one of its keys went
+// to that key's engine, and the batch must queue behind it. The batch used to
+// ride its first command's engine, so a GET for another key could run ahead of
+// that key's pending SET. A spanning batch is refused, and Pipeline() runs it
+// as an ordinary pipeline.
+func TestFDPipelineMultiEngineStaysOnItsKeysEngine(t *testing.T) {
+	srv := newFDStateServer(t)
+	c := NewClient(&Options{Addr: srv.addr(), Protocol: 2, DisableIdentity: true, PipelinePoolSize: 4})
+	t.Cleanup(func() { _ = c.Close() })
+	ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true, NumShards: 4, MaxConcurrentBatches: 1})
+	if err != nil {
+		t.Fatalf("AsyncAutoPipeline: %v", err)
+	}
+	t.Cleanup(func() { _ = ap.Close() })
+	if len(ap.fds) != 4 {
+		t.Fatalf("want 4 engines, got %d", len(ap.fds))
+	}
+	// Two keys on different engines.
+	a, b := "k0", ""
+	for i := 1; b == ""; i++ {
+		if k := "k" + itoa(i); ap.fdForKey(k) != ap.fdForKey(a) {
+			b = k
+		}
+	}
+
+	ctx := context.Background()
+	span := []Cmder{NewStatusCmd(ctx, "set", a, "1"), NewStringCmd(ctx, "get", b)}
+	if err := ap.FDPipelined(ctx, span); !errors.Is(err, ErrFDPipelineSpansEngines) {
+		t.Fatalf("batch over two engines: err=%v, want ErrFDPipelineSpansEngines", err)
+	}
+
+	// Same engine, plus a keyless command that may ride along: accepted.
+	get := NewStringCmd(ctx, "get", a)
+	if err := ap.FDPipelined(ctx, []Cmder{NewStatusCmd(ctx, "set", a, "v"), get, NewStatusCmd(ctx, "ping")}); err != nil {
+		t.Fatalf("batch on one engine: %v", err)
+	}
+	if v, err := get.Result(); err != nil || v != "v" {
+		t.Fatalf("get %s: v=%q err=%v", a, v, err)
+	}
+
+	// Pipeline() runs a spanning batch as an ordinary pipeline.
+	pipe := ap.Pipeline()
+	pipe.Set(ctx, a, "x", 0)
+	pipe.Set(ctx, b, "y", 0)
+	gb := pipe.Get(ctx, b)
+	if _, err := pipe.Exec(ctx); err != nil {
+		t.Fatalf("Exec over two engines: %v", err)
+	}
+	if v, err := gb.Result(); err != nil || v != "y" {
+		t.Fatalf("get %s: v=%q err=%v", b, v, err)
+	}
+}

autopipeline_fd_shards.go

@@ -0,0 +1,159 @@
+package redis
+
+// Multi-connection full duplex: N engines behind ONE autopipeliner.
+//
+// A single full-duplex engine streams on one held connection, and one
+// connection has a throughput ceiling — measured at roughly 920k ops/s on a
+// 14-core client against a local server, for both this client and rueidis. Past
+// that the engine does not saturate, it degrades, because callers queue behind
+// a wire that is already busy. Raising NumShards above 1 under FullDuplex runs
+// several engines on the same client, each holding its own connection, which is
+// the only way past that ceiling.
+//
+// Everything stays on ONE client: one connection pool, one hook chain, one
+// invalidation stream. Only the engines multiply.
+//
+// ROUTING IS BY KEY, NOT ROUND-ROBIN, and that choice is load-bearing. Commands
+// for the same key always take the same engine, so per-key order holds and no
+// Unordered opt-in is needed — the same exemption the half-duplex path already
+// grants cluster slot sharding (see the contentSharded note in
+// newAutoPipeliner). Round-robin would put `SET k` and `GET k`, issued
+// back-to-back as futures on the deferred face, on two engines with no ordering
+// between them, which is a silent behaviour change on a released face.
+//
+// ORDER IS PER FIRST KEY, NOT PER COMMAND STREAM. A single engine keeps the
+// whole submit order; N engines keep only the order of commands that share a
+// first key. A command that touches other keys after its first one (COPY,
+// RENAME, SMOVE, LMOVE, MSET, DEL/UNLINK/EXISTS with several keys, the
+// *STORE family, EVAL with several keys) is routed by its first key only. So
+// `Copy(src, dst)` followed by `Get(dst)` can run in either order: they may be
+// on different engines. Cluster sharding never meets this case because Redis
+// Cluster rejects multi-key commands across slots (CROSSSLOT); a standalone
+// server accepts them. Callers that chain such commands should await the
+// first future before submitting the dependent one, or keep NumShards at 1.
+//
+// Keyless commands are not ordered against anything either: they round-robin.
+// That includes commands that change state for every key (FLUSHDB, FLUSHALL,
+// SWAPDB, SCRIPT FLUSH, FUNCTION FLUSH/RESTORE), so `FlushDB()` followed by
+// `Get(k)` can run the GET first. The same rule applies: await the first
+// future, or keep NumShards at 1.
+//
+// Keys are read with stringArg, the same conversion Ring and the cluster slot
+// hash use, so all three route the same key the same way. It differs from the
+// wire bytes only for pathological key arguments (a nil key is written as ""
+// but read as "<nil>").
+//
+// For comparison, rueidis spreads keyed commands over its standalone wires at
+// random (mux.go slotfn, 4 wires by default on 4+ cores) and orders nothing
+// across them; its Do is synchronous, so there are no in-flight futures to
+// reorder. Only a DoMulti batch shares one wire.
+//
+// This is an opt-in: before this change, NumShards>1 never ran more than one
+// full-duplex engine, so every existing caller keeps the single ordered stream
+// and only a caller that raises NumShards on a full-duplex autopipeliner
+// accepts per-first-key order.
+//
+// The cost of that choice is measurable and worth stating: pinning a caller to
+// one engine means an unlucky key waits behind that engine's queue while
+// another sits idle. At eight wires and 4096 callers this shows up as a wider
+// tail than a balancing client achieves (p99.9 77.6 ms against 59.9 ms), and
+// the p99.9/p50 ratio rises from 1.2x on a single engine to 3.2x on eight. A
+// single FIFO cannot starve anyone; any split gives that up. Ordering was
+// judged the more valuable property, but a selectable policy is the obvious
+// follow-up for callers who would rather have the tail.
+
+import (
+	"hash/fnv"
+)
+
+// fdShardCount reports how many full-duplex engines to run.
+//
+// 0 and 1 both mean a single engine, which is the default and leaves the
+// single-engine path untouched. Sizing is deliberately NOT automatic yet: the
+// optimum tracks caller count (roughly one wire per 256-512 callers), and a
+// fixed count above 1 costs low-concurrency callers real throughput — eight
+// engines measured ~30% behind one at 8 callers, because a handful of callers
+// cannot fill eight queues and every batch fragments. Until the count can be
+// derived from observed load, it stays the caller's explicit decision.
+func fdShardCount(cfg *AutoPipelineOptions) int {
+	if !cfg.FullDuplex || cfg.NumShards <= 1 {
+		return 1
+	}
+	return cfg.NumShards
+}
+
+// fdFor picks the engine that will carry one command.
+//
+// The single-engine case returns before doing any work, so the default
+// configuration pays one length check per command and nothing else.
+func (ap *AutoPipeliner) fdFor(cmd Cmder) *fdEngine {
+	if len(ap.fds) <= 1 {
+		return ap.fd
+	}
+	// Keyed commands hash to a stable engine so per-key order holds. Keyless
+	// ones have no order to preserve, so they round-robin and keep the engines
+	// evenly loaded. Modulo on uint32: converting to int first goes negative
+	// on 32-bit platforms and panics as a slice index.
+	if k, ok := cmdFirstKeyFor(cmd); ok {
+		return ap.fdForKey(k)
+	}
+	return ap.fds[(ap.fdRR.Add(1)-1)%uint32(len(ap.fds))]
+}
+
+// fdForKey is the engine a key hashes to. Multi-engine only.
+func (ap *AutoPipeliner) fdForKey(k string) *fdEngine {
+	h := fnv.New32a()
+	_, _ = h.Write([]byte(k))
+	return ap.fds[h.Sum32()%uint32(len(ap.fds))]
+}
+
+// fdForBatch picks the one engine a whole FD pipeline batch can ride, or
+// reports false when its keyed commands hash to different engines.
+//
+// A batch is contiguous on ONE wire, so it can only keep the per-first-key
+// order if every keyed command in it belongs to that wire: an earlier,
+// unawaited command for one of its keys went to that key's engine, and the
+// batch must queue behind it. Keyless commands are not ordered against
+// anything (see the header), so they ride along. A batch with no keyed
+// command takes the round-robin pick.
+func (ap *AutoPipeliner) fdForBatch(cmds []Cmder) (*fdEngine, bool) {
+	if len(ap.fds) <= 1 {
+		return ap.fd, true
+	}
+	var e *fdEngine
+	for _, cmd := range cmds {
+		k, ok := cmdFirstKeyFor(cmd)
+		if !ok {
+			continue
+		}
+		if pick := ap.fdForKey(k); e == nil {
+			e = pick
+		} else if pick != e {
+			return nil, false
+		}
+	}
+	if e == nil {
+		e = ap.fdFor(cmds[0])
+	}
+	return e, true
+}
+
+// cmdFirstKeyFor returns the command's first key and whether it has one.
+//
+// Uses the shared key-position logic rather than assuming args[1]: that handles
+// keyless commands, the static command table, and eval/evalsha variants whose
+// key position depends on the runtime numkeys argument. Passing a nil
+// CommandInfo keeps this synchronous and network-free, the same way the
+// client-side-cache path calls it.
+//
+// The key is read with stringArg, as Ring does, so a []byte, *string or
+// numeric key hashes like the same key given as a string. The key position,
+// not the key text, decides whether the command is keyed, so an empty-string
+// key still routes by hash.
+func cmdFirstKeyFor(cmd Cmder) (string, bool) {
+	pos := cmdFirstKeyPosWithInfo(cmd, nil)
+	if pos <= 0 || pos >= len(cmd.Args()) {
+		return "", false
+	}
+	return cmd.stringArg(pos), true
+}

autopipeline_fd_shards_internal_test.go

@@ -0,0 +1,58 @@
+package redis
+
+import (
+	"context"
+	"testing"
+)
+
+// The same key must pick the same engine whatever Go type carries it. A key
+// given as []byte or *string used to look keyless and round-robin, so a
+// same-key write and read could land on two engines with no order between
+// them. An empty-string key used to collide with the keyless sentinel.
+func TestFDForRoutesKeyFormsTogether(t *testing.T) {
+	ctx := context.Background()
+	cl := NewClient(&Options{Addr: internalTestRedisAddr()})
+	defer cl.Close()
+	if err := cl.Ping(ctx).Err(); err != nil {
+		t.Skipf("no local redis: %v", err)
+	}
+	ap, err := cl.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{
+		FullDuplex:           true,
+		NumShards:            8,
+		MaxConcurrentBatches: 1,
+	})
+	if err != nil {
+		t.Fatalf("construct: %v", err)
+	}
+	defer ap.Close()
+	if len(ap.fds) != 8 {
+		t.Fatalf("want 8 engines, got %d", len(ap.fds))
+	}
+
+	for i := 0; i < 200; i++ {
+		k := "fdkey:" + itoa(i)
+		s := k
+		want := ap.fdFor(NewCmd(ctx, "get", k))
+		for _, arg := range []interface{}{[]byte(k), &s} {
+			if got := ap.fdFor(NewCmd(ctx, "get", arg)); got != want {
+				t.Fatalf("key %q as %T picked a different engine than as string", k, arg)
+			}
+		}
+	}
+
+	// A raw FCALL routes by its declared key, not by the function name, so it
+	// shares an engine with a plain read of that key.
+	for i := 0; i < 200; i++ {
+		k := "fdfcall:" + itoa(i)
+		if ap.fdFor(NewCmd(ctx, "fcall", "myfn", "1", k)) != ap.fdFor(NewCmd(ctx, "get", k)) {
+			t.Fatalf("raw FCALL on %q picked a different engine than GET %q", k, k)
+		}
+	}
+
+	first := ap.fdFor(NewCmd(ctx, "get", ""))
+	for i := 0; i < 16; i++ {
+		if ap.fdFor(NewCmd(ctx, "set", "", "v")) != first {
+			t.Fatal("empty-string key round-robined instead of routing by hash")
+		}
+	}
+}

autopipeline_fd_shards_test.go

@@ -0,0 +1,184 @@
+package redis_test
+
+// With N full-duplex engines on one client, every reply must
+// still be the reply to its own command. Mis-correlation is the failure mode
+// that matters -- N engines means N independent FIFO streams, and a command
+// routed to engine A whose reply is read off engine B would corrupt data
+// silently rather than error.
+//
+// Requires a local server (apTestAddr: REDIS_PORT, default 6379).
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+
+	redis "github.com/redis/go-redis/v9"
+)
+
+func TestFDShardsReplyCorrelation(t *testing.T) {
+	ctx := context.Background()
+	cl := redis.NewClient(&redis.Options{Addr: apTestAddr()})
+	defer cl.Close()
+	if err := cl.Ping(ctx).Err(); err != nil {
+		t.Skipf("no local redis: %v", err)
+	}
+
+	shards := 8
+	if v := os.Getenv("FD_SHARDS"); v != "" {
+		// Overridable so the same test can run at ONE engine, which is how the
+		// silent half-duplex fallback was first caught: 1 passed, 8 failed.
+		shards, _ = strconv.Atoi(v)
+	}
+	ap, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
+		FullDuplex:           true,
+		NumShards:            shards,
+		MaxConcurrentBatches: 1,
+	})
+	if err != nil {
+		t.Fatalf("construct: %v", err)
+	}
+	defer ap.Close()
+
+	// Each key's value encodes the key, so a reply landing on the wrong command
+	// is caught rather than merely suspected.
+	const keys = 2000
+	for i := 0; i < keys; i++ {
+		k := fmt.Sprintf("fdshards:{%d}", i)
+		if err := ap.Set(ctx, k, fmt.Sprintf("val-%d", i), 0).Err(); err != nil {
+			t.Fatalf("set %d: %v", i, err)
+		}
+	}
+
+	var wg sync.WaitGroup
+	errs := make(chan error, keys)
+	for w := 0; w < 64; w++ {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			for i := w; i < keys; i += 64 {
+				k := fmt.Sprintf("fdshards:{%d}", i)
+				want := fmt.Sprintf("val-%d", i)
+				got, err := ap.Get(ctx, k).Result()
+				if err != nil {
+					errs <- fmt.Errorf("get %s: %w", k, err)
+					return
+				}
+				if got != want {
+					errs <- fmt.Errorf("MIS-CORRELATED: %s got %q want %q", k, got, want)
+					return
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+	close(errs)
+	n := 0
+	for e := range errs {
+		if n < 5 {
+			t.Error(e)
+		}
+		n++
+	}
+	if n > 0 {
+		t.Fatalf("%d correlation failures across %d keys", n, keys)
+	}
+
+	// Per-caller ordering on the deferred face: write then read the SAME key
+	// without waiting in between. Key routing must put both on one engine.
+	for i := 0; i < 500; i++ {
+		k := fmt.Sprintf("fdshards:order:{%d}", i)
+		setF := ap.Set(ctx, k, "second", 0)
+		getF := ap.Get(ctx, k)
+		if err := setF.Err(); err != nil {
+			t.Fatalf("ordered set: %v", err)
+		}
+		got, err := getF.Result()
+		if err != nil {
+			t.Fatalf("ordered get: %v", err)
+		}
+		if got != "second" {
+			t.Fatalf("ORDER VIOLATION at %s: got %q, want the value written just before it", k, got)
+		}
+	}
+	t.Logf("ok: %d keys correlated, 500 same-key write-then-read ordered, %d engines", keys, shards)
+}
+
+// The two-sited rule: NumShards>1 under FullDuplex must actually ENGAGE full
+// duplex, not fall back to half-duplex shards. Config() reports the EFFECTIVE
+// state, so a fallback surfaces as FullDuplex:false.
+func TestFDShardsKeepsFullDuplexOn(t *testing.T) {
+	ctx := context.Background()
+	cl := redis.NewClient(&redis.Options{Addr: apTestAddr()})
+	defer cl.Close()
+	if err := cl.Ping(ctx).Err(); err != nil {
+		t.Skipf("no local redis: %v", err)
+	}
+	ap, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
+		FullDuplex:           true,
+		NumShards:            8,
+		MaxConcurrentBatches: 1,
+	})
+	if err != nil {
+		t.Fatalf("construct: %v", err)
+	}
+	defer ap.Close()
+	if got := ap.Config(); !got.FullDuplex {
+		t.Fatal("FullDuplex reported false with NumShards=8: the engine fell back " +
+			"to half-duplex shards, which are unordered")
+	}
+}
+
+// Full duplex that cannot engage (no pipeline pool) falls back to round-robin
+// half-duplex shards, which do not keep submit order. The NumShards>1 ordering
+// check must then apply: the FullDuplex exemption follows the effective state,
+// not the requested one. Construction only, so no server is needed.
+func TestFDShardsFallbackStillNeedsUnordered(t *testing.T) {
+	cl := redis.NewClient(&redis.Options{
+		Addr:             apTestAddr(),
+		PipelinePoolSize: -1,
+	})
+	defer cl.Close()
+	ap, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
+		FullDuplex:           true,
+		NumShards:            2,
+		MaxConcurrentBatches: 1,
+	})
+	if err == nil {
+		ap.Close()
+		t.Fatal("FullDuplex without a pipeline pool fell back to round-robin shards " +
+			"without the Unordered opt-in")
+	}
+	if !strings.Contains(err.Error(), "Unordered") {
+		t.Fatalf("error should name the Unordered opt-in, got: %v", err)
+	}
+}
+
+// More engines than the pipeline pool can hold must fail at construction rather
+// than leave the surplus spilling to the main pool for the client's lifetime.
+func TestFDShardsRejectsMoreEnginesThanPool(t *testing.T) {
+	ctx := context.Background()
+	cl := redis.NewClient(&redis.Options{
+		Addr:             apTestAddr(),
+		PipelinePoolSize: 4,
+	})
+	defer cl.Close()
+	if err := cl.Ping(ctx).Err(); err != nil {
+		t.Skipf("no local redis: %v", err)
+	}
+	_, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
+		FullDuplex:           true,
+		NumShards:            8,
+		MaxConcurrentBatches: 1,
+	})
+	if err == nil {
+		t.Fatal("expected NumShards=8 with PipelinePoolSize=4 to be rejected")
+	}
+	if !strings.Contains(err.Error(), "pipeline pool") {
+		t.Fatalf("error should name the pipeline pool, got: %v", err)
+	}
+}

command.go

@@ -383,16 +383,23 @@ func cmdFirstKeyPosWithInfo(cmd Cmder, info *CommandInfo) int {
 		// XREADGROUP GROUP group consumer [...] STREAMS key... id...; the
 		// scan starts after the group and consumer names.
 		return cmdArgAfterToken(cmd, 4, "streams")
-	case "object", "xinfo":
-		// OBJECT|XINFO subcommand key; HELP has no key.
+	case "object", "xinfo", "xgroup":
+		// OBJECT|XINFO|XGROUP subcommand key; HELP has no key.
 		if len(cmd.Args()) > 2 {
 			return 2
 		}
 		return 0
+	case "himport":
+		// HIMPORT SET key fieldset ...; PREPARE takes a fieldset name, not a key.
+		if strings.EqualFold(cmd.stringArg(1), "set") {
+			return 2
+		}
+		return 0
 	case "bitop":
 		// BITOP op destkey key...
 		return 2
-	case "lmpop", "zmpop", "sintercard", "zintercard":
+	case "lmpop", "zmpop", "sintercard", "zintercard", "zunion", "zinter", "zdiff",
+		"sdiffcard", "sunioncard", "ts.nrange", "ts.nrevrange":
 		// numkeys key...
 		return 2
 	case "blmpop", "bzmpop":

command_keypos_raw_test.go

@@ -21,11 +21,26 @@ func TestRawCommandsRouteByTheirKey(t *testing.T) {
 		{"xreadgroup", "group", "streams", "streams", "count", 1, "noack", "streams", K, ">"},
 		{"object", "encoding", K},
 		{"xinfo", "stream", K},
+		{"xgroup", "create", K, "g", "$"},
+		{"XGROUP", "CREATE", K, "g", "$", "MKSTREAM"},
+		{"xgroup", "setid", K, "g", "0"},
+		{"xgroup", "destroy", K, "g"},
+		{"xgroup", "createconsumer", K, "g", "c"},
+		{"xgroup", "delconsumer", K, "g", "c"},
 		{"bitop", "and", K, "b"},
 		{"lmpop", 1, K, "left"},
 		{"zmpop", 1, K, "min"},
 		{"sintercard", 1, K},
 		{"zintercard", 2, K, "b"},
+		{"zunion", 1, K},
+		{"zinter", 2, K, "b", "withscores"},
+		{"zdiff", 2, K, "b"},
+		{"sdiffcard", 1, K, "limit", 0},
+		{"sunioncard", 2, K, "b"},
+		{"ts.nrange", 1, K, "-", "+"},
+		{"TS.NREVRANGE", 2, K, "b", "-", "+"},
+		{"himport", "set", K, "fs", "v"},
+		{"HIMPORT", "SET", K, "fs", "v"},
 		{"blmpop", 0, 1, K, "left"},
 		{"bzmpop", 0, 1, K, "min"},
 		{"migrate", "h", 1, K, 0, 1000},
@@ -46,9 +61,35 @@ func TestRawCommandsRouteByTheirKey(t *testing.T) {
 	}
 
 	// Subcommand help has no key.
-	for _, args := range [][]interface{}{{"object", "help"}, {"xinfo", "help"}} {
+	for _, args := range [][]interface{}{{"object", "help"}, {"xinfo", "help"}, {"xgroup", "help"}} {
 		if pos := cmdFirstKeyPosWithInfo(NewCmd(ctx, args...), nil); pos != 0 {
 			t.Errorf("%v: resolver picks position %d, want 0 (keyless)", args, pos)
 		}
 	}
 }
+
+// A keyed scan is routed by its key. A hash tag in MATCH must not move it:
+// the key decides the node, so routing by the pattern sent a cluster scan to
+// the wrong node (MOVED), a Ring scan to the wrong shard (no results), and a
+// multi-engine full-duplex scan past earlier writes to the key. Only the
+// keyless SCAN routes by the pattern's tag.
+func TestKeyedScansRouteByTheirKey(t *testing.T) {
+	ctx := context.Background()
+	var got Cmder
+	c := cmdable(func(_ context.Context, cmd Cmder) error { got = cmd; return nil })
+	for name, run := range map[string]func(){
+		"hscan":          func() { c.HScan(ctx, "k", 0, "{other}*", 0) },
+		"hscan novalues": func() { c.HScanNoValues(ctx, "k", 0, "{other}*", 0) },
+		"sscan":          func() { c.SScan(ctx, "k", 0, "{other}*", 0) },
+		"zscan":          func() { c.ZScan(ctx, "k", 0, "{other}*", 0) },
+	} {
+		run()
+		if pos := cmdFirstKeyPosWithInfo(got, nil); pos != 1 {
+			t.Errorf("%s with a tagged MATCH: routed by position %d (%v), want 1 (the key)", name, pos, got.stringArg(pos))
+		}
+	}
+	c.Scan(ctx, 0, "{tag}*", 0)
+	if pos := cmdFirstKeyPosWithInfo(got, nil); pos != 3 {
+		t.Errorf("keyless scan with a tagged MATCH: routed by position %d, want 3 (the pattern)", pos)
+	}
+}

hash_commands.go

@@ -3,8 +3,6 @@ package redis
 import (
 	"context"
 	"time"
-
-	"github.com/redis/go-redis/v9/internal/hashtag"
 )
 
 type HashCmdable interface {
@@ -206,9 +204,6 @@ func (c cmdable) HScan(ctx context.Context, key string, cursor uint64, match str
 		args = append(args, "count", count)
 	}
 	cmd := NewScanCmd(ctx, c, args...)
-	if hashtag.Present(match) {
-		cmd.SetFirstKeyPos(4)
-	}
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -228,9 +223,6 @@ func (c cmdable) HScanNoValues(ctx context.Context, key string, cursor uint64, m
 	}
 	args = append(args, "novalues")
 	cmd := NewScanCmd(ctx, c, args...)
-	if hashtag.Present(match) {
-		cmd.SetFirstKeyPos(4)
-	}
 	_ = c(ctx, cmd)
 	return cmd
 }

set_commands.go

@@ -1,10 +1,6 @@
 package redis
 
-import (
-	"context"
-
-	"github.com/redis/go-redis/v9/internal/hashtag"
-)
+import "context"
 
 // SetCmdable is an interface for Redis set commands.
 // Sets are unordered collections of unique strings.
@@ -416,9 +412,6 @@ func (c cmdable) SScan(ctx context.Context, key string, cursor uint64, match str
 		args = append(args, "count", count)
 	}
 	cmd := NewScanCmd(ctx, c, args...)
-	if hashtag.Present(match) {
-		cmd.SetFirstKeyPos(4)
-	}
 	_ = c(ctx, cmd)
 	return cmd
 }

sortedset_commands.go

@@ -5,8 +5,6 @@ import (
 	"errors"
 	"strings"
 	"time"
-
-	"github.com/redis/go-redis/v9/internal/hashtag"
 )
 
 type SortedSetCmdable interface {
@@ -770,9 +768,6 @@ func (c cmdable) ZScan(ctx context.Context, key string, cursor uint64, match str
 		args = append(args, "count", count)
 	}
 	cmd := NewScanCmd(ctx, c, args...)
-	if hashtag.Present(match) {
-		cmd.SetFirstKeyPos(4)
-	}
 	_ = c(ctx, cmd)
 	return cmd
 }