fix(autopipeline): reroute cluster FD recovery (#4111)

* fix(autopipeline): reroute cluster FD recovery

A cluster full-duplex node child replayed a dropped session's tail only
on a fresh connection to the same node, and failed its carry and queue
once lease retries ran out. If topology change had closed or replaced
that master, every accepted command on it failed instead of reaching
the slot's new owner (codex P2 on #4002, listed as a known limitation).

Add an internal owner seam, AutoPipelineOptions.fdReroute: an engine
with an owner runs that work through the owner, in order, instead.
- connection error: the replayable tail (same budget partition and
  sent-NoRetry split as the replay) goes through the owner;
- lease retries spent: the carry, then the queue, go through the owner
  instead of failing.
The cluster FD router sets it to the ClusterClient pipeline, which
routes by slot against live topology. Standalone FD is unchanged.

* fix(autopipeline): keep retry budget on FD reroute

fdReroute now takes the remaining retry bound. rerouteReqs goes through
flushCarryBudgeted, so each group passes its remaining budget and a
command never runs more than budget+1 times; the cluster child uses
cc.processPipelineRetries with that bound in place of MaxRedirects.

- The Close-time flush (flushReqs) also goes through the owner, so a
  child closed by topology GC does not flush through its dropped node.
- runReroute returns a per-command transport error even when the owner
  returns an earlier Redis error (redis.Nil), so the flush still aborts.
- flushReqsVia reports per-command metrics before completing, as the
  reader does. This also adds them to the standalone Close-time flush.

* test(autopipeline): fold reroute tests into FD tests

* docs(autopipeline): note nested hooks on FD recovery

.claude/specs/full-duplex.md

@@ -70,6 +70,27 @@ aggregating correctly; only single-node commands route through a child.
   tail on the first socket error. The router injects `cc.opt.MaxRedirects` as
   `clusterRetryBudget`, which `fdEngine.retryBudget()` substitutes in when
   `redirectAware`.
+- **Connection-error and lease-failure recovery (`fdReroute`)**: the router sets
+  each child's `AutoPipelineOptions.fdReroute` to `cc.processPipelineRetries`. When a
+  child's session drops, the replayable tail (after the budget partition and the
+  sent-`NoRetry` split, same set and order as the same-connection replay) runs
+  through the ClusterClient pipeline instead, which routes each command by its slot
+  against live topology. When lease retries are spent, the carry and then the queue
+  go the same way instead of failing. So a node that topology GC closed or replaced
+  does not strand its accepted commands. The Close-time flush (`flushReqs`) also
+  goes through the owner, so a child closed by topology GC does not flush through
+  its dropped node. `fdEngine.rerouteReqs` reuses `flushCarryBudgeted`: each group
+  passes its remaining budget (`fdCarryRemainingRetries`) as the owner's retry
+  bound, so a command never runs more than budget+1 times in total. `runReroute`
+  returns a per-command transport error even when the owner returns an earlier
+  Redis error (`redis.Nil`), so the abort-on-transport-failure rule still holds.
+  `flushReqsVia` reports per-command metrics before it completes a command, as the
+  reader does. Standalone FD (no owner) keeps the same-connection replay and
+  fail-fast paths.
+- **Hooks on recovery**: recovery attempts (the reroute and the MOVED/ASK
+  reprocess) run nested inside the command's `hostHook`, so node pipeline/process
+  hooks see them as nested spans. Making internal recovery hookless is a separate
+  hook-semantics change.
 - **Fallback to `Process`**: routing (`childFor`/`getOrCreateChild`) resolves the
   owning node from live cluster state on every submit. A keyless command, an
   unresolved slot, a not-yet-loaded topology, or a node whose client turns out
@@ -179,13 +200,3 @@ These are documented gaps, tracked as follow-ups, not fixed in the current PR.
   it introduced deadlocked against the off-pipe retry's executor guard (codex P1 on
   #4002). Composing the parent chain into the node engine's single host is the follow-up.
 
-- **Cluster FD connection-error carries stay on the origin node.** When a node child's
-  FD session hits a CONNECTION error, `fd.run` replays the unacked tail on a fresh
-  connection from that same node client's fixed-address pool (bounded by the cluster's
-  `MaxRedirects`). A reply-level redirect (MOVED/ASK) already re-resolves through
-  `cc.process`, but a connection error does not: if topology GC has closed or replaced
-  that master, the accepted commands exhaust their budget with `ErrClosed`/a dial error
-  instead of routing to the replacement. Re-resolving recovered connection-error carries
-  through the parent cluster path (preserving the sent/`NoRetry` guard) needs the
-  `fdConnErr` branch of the recovery loop restructured, so it is a follow-up (codex P2
-  on #4002).

autopipeline.go

@@ -283,6 +283,19 @@ type AutoPipelineOptions struct {
 	// retries live in MaxRedirects). Never set by users (unexported).
 	clusterRetryBudget int
 
+	// fdReroute is set internally by an owner of full-duplex children (the
+	// cluster full-duplex router). Commands the child's engine cannot complete on
+	// its own connection run through it, in order, instead of being replayed on
+	// that connection or failed: the replayable tail after a connection error,
+	// and the carry plus the queue once lease retries are spent. A cluster child
+	// reroutes through the ClusterClient, so a node that went away does not
+	// strand its accepted commands. maxRetries bounds the owner's retry loop
+	// (0 = one execution): the commands' remaining retry budget, so a command
+	// already issued on the child's connection does not get a fresh one. Shutdown
+	// flushes run through it too. Never set by users (unexported). See
+	// fdEngine.reroute.
+	fdReroute func(ctx context.Context, cmds []Cmder, maxRetries int) error
+
 	// NumShards is the number of independent queue+flusher shards the
 	// autopipeliner runs. 0 (the default) means auto: a single shard, which
 	// funnels every caller into one queue so batches stay deep — measured
@@ -2451,6 +2464,8 @@ func (ap *AutoPipeliner) Config() AutoPipelineOptions {
 	// MaxRedirects, used as the child engine's recovery budget); it has no meaning
 	// on a config a caller copies into a standalone client, so strip it too.
 	cfg.clusterRetryBudget = 0
+	// fdReroute holds the owner's closure, like clusterReprocess: strip it.
+	cfg.fdReroute = nil
 	return cfg
 }
 

autopipeline_cluster_fd.go

@@ -90,6 +90,13 @@ func newClusterFDRouter(parent *AutoPipeliner, cc *ClusterClient, cfg *AutoPipel
 	// command on the first socket error. Give the child the cluster's MaxRedirects
 	// so a transient node blip replays the unacked tail on a fresh connection.
 	child.clusterRetryBudget = cc.opt.MaxRedirects
+	// Connection-error and lease-failure recovery: re-issue the child's
+	// replayable work through the ClusterClient pipeline, which routes each
+	// command by its slot against live topology, instead of only on a fresh
+	// connection to this node. A node removed or replaced by topology change
+	// would otherwise exhaust the budget and fail every accepted command on it.
+	// The engine passes each group's remaining budget as the redirect bound.
+	child.fdReroute = cc.processPipelineRetries
 	return &clusterFDRouter{
 		parent:   parent,
 		cc:       cc,

autopipeline_fullduplex.go

@@ -704,6 +704,15 @@ type fdEngine struct {
 	// cluster mode (redirectAware): the ClusterClient's MaxRedirects, injected by the
 	// router. See retryBudget().
 	clusterRetryBudget int
+	// reroute, when set by the engine's owner (AutoPipelineOptions.fdReroute),
+	// runs commands this engine cannot complete on its own connection through the
+	// owner's path, in order: the replayable tail after a connection error (in
+	// place of the same-connection replay) and the carry plus the queue once
+	// lease retries are spent (in place of failing them), and the Close-time
+	// flush (in place of this engine's own client pipeline). maxRetries is the
+	// commands' remaining retry budget. nil keeps the standalone behavior. Set
+	// once in newFDEngine.
+	reroute func(ctx context.Context, cmds []Cmder, maxRetries int) error
 }
 
 // retryBudget bounds the engine's OWN connection-failure recovery: the lease retry
@@ -820,6 +829,7 @@ func newFDEngine(ap *AutoPipeliner, client *Client) *fdEngine {
 	} else {
 		fd.reprocess = client.processStartingAt
 	}
+	fd.reroute = ap.config.fdReroute
 	// Keep one warm spare in the dedicated pipeline pool. The engine holds one
 	// pipeline connection for its whole session, so without an idle spare the
 	// replacement after a connection error must dial first, while every unacked
@@ -1360,8 +1370,17 @@ func (fd *fdEngine) run() {
 				fd.sleepBackoff(leaseAttempts)
 				continue // carry unchanged; re-lease
 			}
-			fd.failReqs(carry, aerr)
-			fd.failQueue(aerr)
+			if fd.reroute != nil {
+				// This connection's endpoint cannot be leased, but the owner may
+				// still serve the commands (e.g. through the slot's new master): run
+				// the carry, then the queue, through it in order instead of failing
+				// them. The carry holds no sent NoRetry command (the split already
+				// failed those).
+				fd.rerouteReqs(bg, append(carry, fd.q.drainAll(nil)...))
+			} else {
+				fd.failReqs(carry, aerr)
+				fd.failQueue(aerr)
+			}
 			carry = nil
 			leaseAttempts++
 			fd.sleepBackoff(leaseAttempts)
@@ -1447,6 +1466,13 @@ func (fd *fdEngine) run() {
 							carry[i].batch.fdAttempts = carry[i].attempts
 						}
 					}
+					if fd.reroute != nil {
+						// Re-issue through the owner rather than on a fresh connection to
+						// this same endpoint, which may be gone (a cluster node replaced
+						// by topology change). Same set, same order as the replay.
+						fd.rerouteReqs(bg, carry)
+						carry = nil
+					}
 					continue
 				}
 			}
@@ -3136,7 +3162,52 @@ func (fd *fdEngine) flushCarryBudgeted(bg context.Context, carry []fdReq) error
 // fails the remainder so an ordered flush stops rather than running later chunks
 // out of order. Every accepted command still settles (it is failed, not left
 // hanging), honoring accepted⇒completes. See TestFDShutdownFlushAbortsAfterRecoveredPanic.
-func (fd *fdEngine) flushReqs(bg context.Context, reqs []fdReq, maxRetries int) (retErr error) {
+func (fd *fdEngine) flushReqs(bg context.Context, reqs []fdReq, maxRetries int) error {
+	// Test seam: nil in production.
+	if fd.runPipeline != nil {
+		return fd.flushReqsVia(bg, reqs, maxRetries, fd.runPipeline)
+	}
+	// An owner takes the Close-time flush too: this engine closes when its
+	// endpoint goes away (a cluster node dropped by topology GC), and this
+	// engine's own client cannot follow the commands to the new owner.
+	if fd.reroute != nil {
+		return fd.flushReqsVia(bg, reqs, maxRetries, fd.runReroute)
+	}
+	return fd.flushReqsVia(bg, reqs, maxRetries, fd.client.processPipelineRetries)
+}
+
+// rerouteReqs runs reqs, in order, through the owner's path (fd.reroute) and
+// completes them, with flushReqs' chunking and abort-on-transport-failure. The
+// commands may have been sent once on this engine's connection: the same
+// at-most-once rule as the same-connection replay applies (callers pass only
+// the replayable set, never a sent NoRetry command). reqs are POST-BUMP, like
+// the Close-time carry, so flushCarryBudgeted gives each group its remaining
+// budget (fdCarryRemainingRetries) and flushReqs sends it to the owner.
+func (fd *fdEngine) rerouteReqs(bg context.Context, reqs []fdReq) {
+	_ = fd.flushCarryBudgeted(bg, reqs)
+}
+
+// runReroute runs cmds through the owner and returns the error flushReqsVia
+// uses to decide whether to abort. An owner pipeline may return the first
+// command error (e.g. redis.Nil from ClusterClient.processPipeline) while a
+// later command failed in transport; return that transport error instead, so
+// later chunks do not run after an uncertain predecessor.
+func (fd *fdEngine) runReroute(ctx context.Context, cmds []Cmder, maxRetries int) error {
+	err := fd.reroute(ctx, cmds, maxRetries)
+	if err != nil && pipelineErrShouldStamp(err) {
+		return err
+	}
+	for _, cmd := range cmds {
+		// rawErr: engine goroutine, before complete(); never await here.
+		if e := cmd.rawErr(); e != nil && pipelineErrShouldStamp(e) {
+			return e
+		}
+	}
+	return err
+}
+
+// flushReqsVia is flushReqs with an explicit pipeline executor.
+func (fd *fdEngine) flushReqsVia(bg context.Context, reqs []fdReq, maxRetries int, run func(context.Context, []Cmder, int) error) (retErr error) {
 	if len(reqs) == 0 {
 		return nil
 	}
@@ -3155,11 +3226,6 @@ func (fd *fdEngine) flushReqs(bg context.Context, reqs []fdReq, maxRetries int)
 			fd.failReqs(reqs[i:], retErr)
 		}
 	}()
-	// Test seam: nil in production, so this is exactly client.processPipelineRetries.
-	run := fd.runPipeline
-	if run == nil {
-		run = fd.client.processPipelineRetries
-	}
 	// Same MaxBatchSize/MaxBatchBytes chunking as normal FD writes: reqs can hold a
 	// large window, and one unchunked pipeline would ignore MaxBatchBytes and burst
 	// the connection.
@@ -3200,8 +3266,14 @@ func (fd *fdEngine) flushReqs(bg context.Context, reqs []fdReq, maxRetries int)
 			}
 		}
 		cmds := make([]Cmder, end-i)
+		now := time.Now()
 		for j := i; j < end; j++ {
 			cmds[j-i] = reqs[j].cmd
+			// Anchor the duration metric of a command never written on the FD
+			// connection (e.g. queued during a lease failure) at this run.
+			if reqs[j].writtenOff == 0 {
+				reqs[j].writtenOff = max(int64(now.Sub(fd.epoch)), 1)
+			}
 			// The pooled pipeline below runs this chunk again and records its
 			// pipeline metric. Stamped here, where the chunk runs: one more
 			// issue (so fdPipelineExec does not re-run it), and flushed. A batch
@@ -3226,6 +3298,16 @@ func (fd *fdEngine) flushReqs(bg context.Context, reqs []fdReq, maxRetries int)
 		}
 		err := run(fctx, cmds, maxRetries) // per-command results/errors set inside
 		for j := i; j < end; j++ {
+			// Per-command metrics, as the reader reports them inline: this path
+			// bypasses the reader. A pipelined command is measured with its
+			// batch (fdPipelineMetrics).
+			if !reqs[j].pipelined {
+				octx := reqs[j].ctx
+				if octx == nil {
+					octx = bg
+				}
+				fd.reportReplyMetrics(octx, reqs[j], reqs[j].cmd.rawErr(), nil)
+			}
 			reqs[j].complete()
 		}
 		i = end

autopipeline_fullduplex_test.go

@@ -6911,6 +6911,347 @@ func BenchmarkFDQueueDeepDrain(b *testing.B) {
 	}
 }
 
+// rerouteTo returns an fdReroute that runs the commands on target and counts
+// them.
+func rerouteTo(target *Client, n *atomic.Int64) func(context.Context, []Cmder, int) error {
+	return func(ctx context.Context, cmds []Cmder, maxRetries int) error {
+		n.Add(int64(len(cmds)))
+		return target.processPipelineRetries(ctx, cmds, maxRetries)
+	}
+}
+
+// fdRerouteTestEngine is an engine with only what the flush paths need, and
+// a reroute that records each call's retry bound and serves the commands with
+// serve.
+func fdRerouteTestEngine(budget int, serve func(cmds []Cmder) error) (*fdEngine, *[]int) {
+	var bounds []int
+	fd := &fdEngine{
+		ap:                 &AutoPipeliner{config: &AutoPipelineOptions{}},
+		client:             &Client{baseClient: &baseClient{opt: &Options{}}},
+		epoch:              time.Now(),
+		maxBatch:           8,
+		q:                  newFDQueue(8),
+		redirectAware:      true,
+		clusterRetryBudget: budget,
+	}
+	fd.reroute = func(_ context.Context, cmds []Cmder, maxRetries int) error {
+		bounds = append(bounds, maxRetries)
+		return serve(cmds)
+	}
+	return fd, &bounds
+}
+
+func fdServeOK(cmds []Cmder) error {
+	for _, c := range cmds {
+		c.SetErr(nil)
+	}
+	return nil
+}
+
+// A rerouted command keeps its remaining retry budget: the owner gets the
+// budget left after the attempts already made on the FD connection, not a
+// fresh one (codex P1 on #4111). Red before: the bound was dropped, and the
+// cluster owner ran its full MaxRedirects loop.
+func TestFDRerouteKeepsRemainingRetryBudget(t *testing.T) {
+	ctx := context.Background()
+	fd, bounds := fdRerouteTestEngine(3, fdServeOK)
+	// POST-BUMP carry: attempts=2 is one execution done on the FD connection.
+	// attempts=1 is a queued command never sent.
+	reqs := []fdReq{
+		{cmd: NewStatusCmd(ctx, "set", "a", "1"), batch: newAPBatch(), attempts: 2},
+		{cmd: NewStatusCmd(ctx, "set", "b", "1"), batch: newAPBatch(), attempts: 2},
+		{cmd: NewStatusCmd(ctx, "set", "c", "1"), batch: newAPBatch(), attempts: 1},
+	}
+	fd.rerouteReqs(ctx, reqs)
+	if got, want := fmt.Sprint(*bounds), "[2 3]"; got != want {
+		t.Fatalf("reroute retry bounds = %s, want %s (budget 3 minus executions done)", got, want)
+	}
+}
+
+// The Close-time flush of an engine with an owner goes through the owner, not
+// through the engine's own client (codex P2 on #4111: a child closed by
+// topology GC flushed through its dropped node).
+func TestFDShutdownFlushReroutes(t *testing.T) {
+	ctx := context.Background()
+	fd, bounds := fdRerouteTestEngine(3, fdServeOK)
+	// Nothing listens here: a flush through the engine's own client fails.
+	fd.client = NewClient(&Options{Addr: "127.0.0.1:1", MaxRetries: -1, DialTimeout: 50 * time.Millisecond})
+	defer fd.client.Close()
+
+	carry := NewStatusCmd(ctx, "set", "a", "1")
+	queued := NewStatusCmd(ctx, "set", "b", "1")
+	if fd.q.push(fdReq{cmd: queued, batch: newAPBatch(), attempts: 1}) != fdPushOK {
+		t.Fatal("queue push failed")
+	}
+	fd.shutdownFlush(ctx, []fdReq{{cmd: carry, batch: newAPBatch(), attempts: 2}})
+	if got, want := fmt.Sprint(*bounds), "[2 3]"; got != want {
+		t.Fatalf("reroute retry bounds = %s, want %s", got, want)
+	}
+	for _, c := range []*StatusCmd{carry, queued} {
+		if err := c.rawErr(); err != nil {
+			t.Fatalf("%v: err = %v, want it served through the reroute", c.Args(), err)
+		}
+	}
+}
+
+// An owner pipeline that returns an earlier Redis error (redis.Nil) while a
+// later command failed in transport still aborts the rest of the flush
+// (codex P2 on #4111).
+func TestFDRerouteDetectsMaskedTransportError(t *testing.T) {
+	ctx := context.Background()
+	calls := 0
+	fd, _ := fdRerouteTestEngine(3, func(cmds []Cmder) error {
+		calls++
+		cmds[0].SetErr(Nil)
+		for _, c := range cmds[1:] {
+			c.SetErr(io.EOF)
+		}
+		return Nil // first command error, as ClusterClient.processPipeline returns
+	})
+	fd.maxBatch = 2 // two chunks
+	reqs := make([]fdReq, 4)
+	for i := range reqs {
+		reqs[i] = fdReq{cmd: NewStringCmd(ctx, "get", "k"+itoa(i)), batch: newAPBatch(), attempts: 1}
+	}
+	fd.rerouteReqs(ctx, reqs)
+	if calls != 1 {
+		t.Fatalf("owner ran %d chunks, want 1: the second ran after a transport failure", calls)
+	}
+	for i := 2; i < len(reqs); i++ {
+		if err := reqs[i].cmd.rawErr(); !errors.Is(err, io.EOF) {
+			t.Fatalf("reqs[%d] err = %v, want io.EOF (failed by the abort)", i, err)
+		}
+	}
+}
+
+// Rerouted commands report the per-command duration metric the reader would
+// have reported (codex P2 on #4111).
+func TestFDRerouteRecordsCommandMetrics(t *testing.T) {
+	rec := &fdOtelRecorder{}
+	otel.SetGlobalRecorder(rec)
+	defer otel.SetGlobalRecorder(nil)
+
+	ctx := context.Background()
+	fd, _ := fdRerouteTestEngine(3, fdServeOK)
+	reqs := []fdReq{
+		{cmd: NewStatusCmd(ctx, "set", "a", "1"), batch: newAPBatch(), attempts: 2},
+		{cmd: NewStatusCmd(ctx, "set", "b", "1"), batch: newAPBatch(), attempts: 1},
+	}
+	fd.rerouteReqs(ctx, reqs)
+	if n := rec.opDurations.Load(); n != int64(len(reqs)) {
+		t.Fatalf("operation durations recorded = %d, want %d", n, len(reqs))
+	}
+}
+
+// With an owner reroute, a persistent lease failure runs the accepted commands
+// through the owner instead of failing them.
+func TestFDLeaseFailureReroutesBacklog(t *testing.T) {
+	ctx := context.Background()
+	target := NewClient(&Options{Addr: ":6379"})
+	defer target.Close()
+	if err := target.Ping(ctx).Err(); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+
+	c := NewClient(&Options{
+		Addr:             "127.0.0.1:1", // nothing listening: every lease fails
+		Protocol:         3,
+		PipelinePoolSize: 2,
+		PoolSize:         2,
+		MaxRetries:       2,
+		DialTimeout:      150 * time.Millisecond,
+		MinRetryBackoff:  time.Millisecond,
+		MaxRetryBackoff:  5 * time.Millisecond,
+	})
+	defer c.Close()
+	var rerouted atomic.Int64
+	ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{
+		FullDuplex: true,
+		fdReroute:  rerouteTo(target, &rerouted),
+	})
+	if err != nil {
+		t.Fatalf("AsyncAutoPipeline: %v", err)
+	}
+	defer ap.Close()
+	if ap.fd == nil || ap.fd.reroute == nil {
+		t.Fatal("full-duplex engine with a reroute not active")
+	}
+
+	key := "fd:reroute:lease"
+	done := make(chan error, 1)
+	go func() { done <- ap.Set(ctx, key, "v", 0).Err() }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Set = %v, want it served through the reroute", err)
+		}
+	case <-time.After(6 * time.Second):
+		t.Fatal("command hung on a persistent lease failure")
+	}
+	if rerouted.Load() == 0 {
+		t.Fatal("reroute not used")
+	}
+	if got, err := target.Get(ctx, key).Result(); err != nil || got != "v" {
+		t.Fatalf("GET on the reroute target = %q, %v", got, err)
+	}
+}
+
+// With an owner reroute, the replayable tail after a connection error runs
+// through the owner, in order: no worker sees a failed or misaligned reply.
+func TestFDConnErrReroutesTail(t *testing.T) {
+	ctx := context.Background()
+	target := NewClient(&Options{Addr: ":6379"})
+	defer target.Close()
+	if err := target.Ping(ctx).Err(); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+	c := fdTestClient(":6379")
+	defer c.Close()
+	var rerouted atomic.Int64
+	ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{
+		FullDuplex: true,
+		fdReroute:  rerouteTo(target, &rerouted),
+	})
+	if err != nil {
+		t.Fatalf("AsyncAutoPipeline: %v", err)
+	}
+	defer ap.Close()
+
+	killer := NewClient(&Options{Addr: ":6379"})
+	defer killer.Close()
+
+	const workers = 16
+	var stop atomic.Bool
+	done := make(chan error, workers)
+	for w := 0; w < workers; w++ {
+		go func(w int) {
+			key := "fd:reroute:kill:" + itoa(w)
+			for n := 0; !stop.Load(); n++ {
+				val := itoa(w) + ":" + itoa(n)
+				if err := ap.Set(ctx, key, val, 0).Err(); err != nil {
+					done <- fmt.Errorf("worker %d set #%d: %w", w, n, err)
+					return
+				}
+				got, err := ap.Get(ctx, key).Result()
+				if err != nil {
+					done <- fmt.Errorf("worker %d get #%d: %w", w, n, err)
+					return
+				}
+				if got != val {
+					done <- fmt.Errorf("worker %d get #%d: got %q want %q", w, n, got, val)
+					return
+				}
+			}
+			done <- nil
+		}(w)
+	}
+
+	time.Sleep(120 * time.Millisecond)
+	for i := 0; i < 5; i++ {
+		_ = killer.Do(ctx, "CLIENT", "KILL", "TYPE", "normal", "SKIPME", "yes").Err()
+		time.Sleep(90 * time.Millisecond)
+	}
+	time.Sleep(120 * time.Millisecond)
+	stop.Store(true)
+
+	deadline := time.After(15 * time.Second)
+	for w := 0; w < workers; w++ {
+		select {
+		case err := <-done:
+			if err != nil {
+				t.Fatal(err)
+			}
+		case <-deadline:
+			t.Fatal("timeout: a worker hung after a connection kill")
+		}
+	}
+	if rerouted.Load() == 0 {
+		t.Fatal("no command went through the reroute across 5 connection kills")
+	}
+}
+
+// The cluster full-duplex router gives its node children a reroute through
+// the ClusterClient, and Config never hands it back.
+func TestClusterFDRouterChildReroutes(t *testing.T) {
+	cc := NewClusterClient(&ClusterOptions{Addrs: []string{"127.0.0.1:1"}})
+	defer cc.Close()
+	r := newClusterFDRouter(nil, cc, &AutoPipelineOptions{FullDuplex: true}, true)
+	if r.childCfg.fdReroute == nil {
+		t.Fatal("cluster FD child config has no reroute")
+	}
+
+	ap := &AutoPipeliner{config: r.childCfg}
+	if ap.Config().fdReroute != nil {
+		t.Fatal("Config() exposes the internal reroute")
+	}
+}
+
+// Cluster full-duplex under connection kills on every master: each node child's
+// replayable tail reroutes through the ClusterClient, in order, so no worker
+// sees a failed or misaligned reply.
+func TestClusterFDConnKillReroutesTail(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()
+	if ap.clusterFD == nil {
+		t.Fatal("cluster full-duplex router not active")
+	}
+
+	const workers = 24
+	var stop atomic.Bool
+	done := make(chan error, workers)
+	for w := 0; w < workers; w++ {
+		go func(w int) {
+			key := fmt.Sprintf("cfd:reroute:{%d}", w) // spread over all masters
+			for n := 0; !stop.Load(); n++ {
+				val := itoa(w) + ":" + itoa(n)
+				if err := ap.Set(ctx, key, val, 0).Err(); err != nil {
+					done <- fmt.Errorf("worker %d set #%d: %w", w, n, err)
+					return
+				}
+				got, err := ap.Get(ctx, key).Result()
+				if err != nil {
+					done <- fmt.Errorf("worker %d get #%d: %w", w, n, err)
+					return
+				}
+				if got != val {
+					done <- fmt.Errorf("worker %d get #%d: got %q want %q", w, n, got, val)
+					return
+				}
+			}
+			done <- nil
+		}(w)
+	}
+
+	time.Sleep(150 * time.Millisecond)
+	for i := 0; i < 4; i++ {
+		_ = cc.ForEachMaster(ctx, func(ctx context.Context, m *Client) error {
+			return m.Do(ctx, "CLIENT", "KILL", "TYPE", "normal", "SKIPME", "yes").Err()
+		})
+		time.Sleep(100 * time.Millisecond)
+	}
+	time.Sleep(150 * time.Millisecond)
+	stop.Store(true)
+
+	deadline := time.After(20 * time.Second)
+	for w := 0; w < workers; w++ {
+		select {
+		case err := <-done:
+			if err != nil {
+				t.Fatal(err)
+			}
+		case <-deadline:
+			t.Fatal("timeout: a worker hung after a connection kill")
+		}
+	}
+}
+
 // FullDuplexPipes chooses the number of full-duplex pipes (engines). NumShards>1
 // is still honored as the 9.23.0 spelling when FullDuplexPipes is 0, and
 // Config() reports the effective pipe count either way. Construction only:

osscluster.go

@@ -2759,6 +2759,15 @@ func (c *ClusterClient) pipelineDecision(
 }
 
 func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error {
+	return c.processPipelineRetries(ctx, cmds, c.opt.MaxRedirects)
+}
+
+// processPipelineRetries is processPipeline with an explicit bound on the
+// redirect/retry loop in place of MaxRedirects: the pipeline runs at most
+// maxRedirects+1 times. A full-duplex node child reroutes through it with the
+// commands' remaining budget, so a command already issued on the child's
+// connection is not given a fresh MaxRedirects+1 executions.
+func (c *ClusterClient) processPipelineRetries(ctx context.Context, cmds []Cmder, maxRedirects int) error {
 	// Only call time.Now() if pipeline operation duration callback is set to avoid overhead
 	var operationStart time.Time
 	pipelineOpDurationCallback := otel.GetPipelineOperationDurationCallback()
@@ -2793,7 +2802,7 @@ func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error
 	}
 
 	var lastErr error
-	for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
+	for attempt := 0; attempt <= maxRedirects; attempt++ {
 		totalAttempts++
 		if attempt > 0 {
 			if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {