go-redis/osscluster_test.go

package redis_test

import (
	"bytes"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"net"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	. "github.com/bsm/ginkgo/v2"
	. "github.com/bsm/gomega"

	"github.com/redis/go-redis/v9"
	"github.com/redis/go-redis/v9/internal/hashtag"
	"github.com/redis/go-redis/v9/internal/routing"
)

type clusterScenario struct {
	ports   []string
	nodeIDs []string
	clients map[string]*redis.Client
}

func (s *clusterScenario) slots() []int {
	return []int{0, 5461, 10923, 16384}
}

func (s *clusterScenario) masters() []*redis.Client {
	result := make([]*redis.Client, 3)
	for pos, port := range s.ports[:3] {
		result[pos] = s.clients[port]
	}
	return result
}

func (s *clusterScenario) slaves() []*redis.Client {
	result := make([]*redis.Client, 3)
	for pos, port := range s.ports[3:] {
		result[pos] = s.clients[port]
	}
	return result
}

func (s *clusterScenario) addrs() []string {
	addrs := make([]string, len(s.ports))
	for i, port := range s.ports {
		addrs[i] = net.JoinHostPort("127.0.0.1", port)
	}
	return addrs
}

func (s *clusterScenario) newClusterClientUnstable(opt *redis.ClusterOptions) *redis.ClusterClient {
	opt.Addrs = s.addrs()
	return redis.NewClusterClient(opt)
}

func (s *clusterScenario) newClusterClient(
	ctx context.Context, opt *redis.ClusterOptions,
) *redis.ClusterClient {
	client := s.newClusterClientUnstable(opt)
	err := eventually(func() error {
		if opt.ClusterSlots != nil {
			return nil
		}

		state, err := client.LoadState(ctx)
		if err != nil {
			return err
		}

		if !state.IsConsistent(ctx) {
			return fmt.Errorf("cluster state is not consistent")
		}

		return nil
	}, 30*time.Second)
	if err != nil {
		panic(err)
	}

	return client
}

func (s *clusterScenario) Close() error {
	ctx := context.TODO()
	for _, master := range s.masters() {
		if master == nil {
			continue
		}
		err := master.FlushAll(ctx).Err()
		if err != nil {
			return err
		}

		// since 7.2 forget calls should be propagated, calling only master
		// nodes should be sufficient.
		for _, nID := range s.nodeIDs {
			master.ClusterForget(ctx, nID)
		}
	}

	return nil
}

func configureClusterTopology(ctx context.Context, scenario *clusterScenario) error {
	allowErrs := []string{
		"ERR Slot 0 is already busy",
		"ERR Slot 5461 is already busy",
		"ERR Slot 10923 is already busy",
		"ERR Slot 16384 is already busy",
	}

	err := collectNodeInformation(ctx, scenario)
	if err != nil {
		return err
	}

	// Meet cluster nodes.
	for _, client := range scenario.clients {
		err := client.ClusterMeet(ctx, "127.0.0.1", scenario.ports[0]).Err()
		if err != nil {
			return err
		}
	}

	slots := scenario.slots()
	for pos, master := range scenario.masters() {
		err := master.ClusterAddSlotsRange(ctx, slots[pos], slots[pos+1]-1).Err()
		if err != nil && slices.Contains(allowErrs, err.Error()) == false {
			return err
		}
	}

	// Bootstrap slaves.
	for idx, slave := range scenario.slaves() {
		masterID := scenario.nodeIDs[idx]

		// Wait until master is available
		err := eventually(func() error {
			s := slave.ClusterNodes(ctx).Val()
			wanted := masterID
			if !strings.Contains(s, wanted) {
				return fmt.Errorf("%q does not contain %q", s, wanted)
			}
			return nil
		}, 10*time.Second)
		if err != nil {
			return err
		}

		err = slave.ClusterReplicate(ctx, masterID).Err()
		if err != nil {
			return err
		}
	}

	// Wait until all nodes have consistent info.
	wanted := []redis.ClusterSlot{{
		Start: 0,
		End:   5460,
		Nodes: []redis.ClusterNode{{
			ID:   "",
			Addr: "127.0.0.1:16600",
		}, {
			ID:   "",
			Addr: "127.0.0.1:16603",
		}},
	}, {
		Start: 5461,
		End:   10922,
		Nodes: []redis.ClusterNode{{
			ID:   "",
			Addr: "127.0.0.1:16601",
		}, {
			ID:   "",
			Addr: "127.0.0.1:16604",
		}},
	}, {
		Start: 10923,
		End:   16383,
		Nodes: []redis.ClusterNode{{
			ID:   "",
			Addr: "127.0.0.1:16602",
		}, {
			ID:   "",
			Addr: "127.0.0.1:16605",
		}},
	}}

	for _, client := range scenario.clients {
		err := eventually(func() error {
			res, err := client.ClusterSlots(ctx).Result()
			if err != nil {
				return err
			}
			return assertSlotsEqual(res, wanted)
		}, 90*time.Second)
		if err != nil {
			return err
		}
	}

	return nil
}

func collectNodeInformation(ctx context.Context, scenario *clusterScenario) error {
	for pos, port := range scenario.ports {
		client := redis.NewClient(&redis.Options{
			Addr: ":" + port,
		})

		myID, err := client.ClusterMyID(ctx).Result()
		if err != nil {
			return err
		}

		scenario.clients[port] = client
		scenario.nodeIDs[pos] = myID
	}
	return nil
}

func assertSlotsEqual(slots, wanted []redis.ClusterSlot) error {
outerLoop:
	for _, s2 := range wanted {
		for _, s1 := range slots {
			if slotEqual(s1, s2) {
				continue outerLoop
			}
		}
		return fmt.Errorf("%v not found in %v", s2, slots)
	}
	return nil
}

func slotEqual(s1, s2 redis.ClusterSlot) bool {
	if s1.Start != s2.Start {
		return false
	}
	if s1.End != s2.End {
		return false
	}
	if len(s1.Nodes) != len(s2.Nodes) {
		return false
	}
	for i, n1 := range s1.Nodes {
		if n1.Addr != s2.Nodes[i].Addr {
			return false
		}
	}
	return true
}

// txWriteRecorder captures, per node addr, each Write payload that carries a
// MULTI/EXEC transaction. txWriteCount uses it to prove whether the client
// retried a transaction (e.g. after a transient write/read failure or an ASK).
type txWriteRecorder struct {
	mu     sync.Mutex
	writes map[string][][]byte // addr -> tx write payloads
}

func newTxWriteRecorder() *txWriteRecorder {
	return &txWriteRecorder{writes: map[string][][]byte{}}
}

func (r *txWriteRecorder) record(addr string, b []byte) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.writes[addr] = append(r.writes[addr], bytes.Clone(b))
}

// txWriteCount returns the number of tx writes (MULTI-containing batches) sent
// to addr. Used to prove whether the client retried a transaction.
func (r *txWriteRecorder) txWriteCount(addr string) int {
	r.mu.Lock()
	defer r.mu.Unlock()
	n := 0
	for _, b := range r.writes[addr] {
		if countRespCmd(b, "MULTI") > 0 {
			n++
		}
	}
	return n
}

// countRespCmd counts occurrences of a RESP command name (e.g. "MULTI") within a
// raw Write payload by matching the bulk-string framing "$<len>\r\n<NAME>\r\n".
// Matching is case-insensitive: go-redis writes tx commands in lower case
// ("multi"/"exec") but other commands may differ.
func countRespCmd(b []byte, cmd string) int {
	needle := []byte("$" + strconv.Itoa(len(cmd)) + "\r\n" + strings.ToUpper(cmd) + "\r\n")
	return bytes.Count(bytes.ToUpper(b), needle)
}

// txFaultConn wraps a real redis connection. Non-tx traffic is forwarded
// unchanged so the cluster client can still load topology; tx writes (those
// containing MULTI) are always recorded. When failOnce is non-nil
// (transient-retry test), the first MULTI write across all sharing conns
// fails with io.EOF and subsequent ones are forwarded, so the client must retry the
// whole tx and then succeed.
type txFaultConn struct {
	net.Conn
	addr     string
	rec      *txWriteRecorder
	failOnce *atomic.Int32
	// failAddr, when set, restricts failOnce to writes whose addr matches; an
	// empty failAddr (the default) fires on the first MULTI write to any addr.
	failAddr string
	// failRead makes reads return io.EOF once a MULTI write has been seen on
	// this conn, so the tx batch is committed server-side but the client cannot
	// read the reply. Used to prove a post-commit read failure is not retried.
	failRead bool
	sawMulti atomic.Bool
}

func (c *txFaultConn) Write(b []byte) (int, error) {
	if countRespCmd(b, "MULTI") > 0 {
		c.sawMulti.Store(true)
		c.rec.record(c.addr, b)
		if c.failOnce != nil && (c.failAddr == "" || c.addr == c.failAddr) &&
			c.failOnce.CompareAndSwap(0, 1) {
			return 0, io.EOF
		}
	}
	return c.Conn.Write(b)
}

func (c *txFaultConn) Read(b []byte) (int, error) {
	if c.failRead && c.sawMulti.Load() {
		return 0, io.EOF
	}
	return c.Conn.Read(b)
}

// slotMasterPort returns the port of the master owning `slot` in the test
// cluster's fixed distribution (0-5460 -> 16600, 5461-10922 -> 16601,
// 10923-16383 -> 16602), matching clusterScenario.slots().
func slotMasterPort(slot int) string {
	switch {
	case slot < 5461:
		return cluster.ports[0]
	case slot < 10923:
		return cluster.ports[1]
	default:
		return cluster.ports[2]
	}
}

// slotOtherMasterPort returns the port of a master that does NOT own `slot`,
// matching the migration target that setupAskMigration picks (the first master
// port != the slot's owner).
func slotOtherMasterPort(slot int) string {
	src := slotMasterPort(slot)
	for _, p := range cluster.ports[:3] {
		if p != src {
			return p
		}
	}
	return ""
}

// resetSlot restores `slot` to a clean, fully-owned state on its master and
// deletes `keys` from all masters. It is idempotent and best-effort, used to
// recover from a prior run that left the slot migrating/importing.
func resetSlot(ctx context.Context, slot int, keys ...string) {
	srcPort := slotMasterPort(slot)
	srcCli := redis.NewClient(&redis.Options{Addr: "127.0.0.1:" + srcPort})
	srcID, _ := srcCli.ClusterMyID(ctx).Result()
	_ = srcCli.Close()
	for _, p := range cluster.ports[:3] {
		c := redis.NewClient(&redis.Options{Addr: "127.0.0.1:" + p})
		_ = c.Do(ctx, "cluster", "setslot", slot, "node", srcID).Err()
		for _, k := range keys {
			_ = c.Del(ctx, k).Err()
		}
		_ = c.Close()
	}
}

// setupAskMigration puts `slot` into MIGRATING on its owner (source) and IMPORTING
// on a different master (target), leaving the slot's keys absent from both. The
// source then returns -ASK for any absent key in the slot. The returned cleanup
// restores slot ownership on all masters.
func setupAskMigration(ctx context.Context, slot int) func() {
	resetSlot(ctx, slot)
	srcPort := slotMasterPort(slot)
	tgtPort := ""
	for _, p := range cluster.ports[:3] {
		if p != srcPort {
			tgtPort = p
			break
		}
	}
	srcCli := redis.NewClient(&redis.Options{Addr: "127.0.0.1:" + srcPort})
	tgtCli := redis.NewClient(&redis.Options{Addr: "127.0.0.1:" + tgtPort})

	srcID, err := srcCli.ClusterMyID(ctx).Result()
	Expect(err).NotTo(HaveOccurred())
	dstID, err := tgtCli.ClusterMyID(ctx).Result()
	Expect(err).NotTo(HaveOccurred())

	Expect(srcCli.Do(ctx, "cluster", "setslot", slot, "migrating", dstID).Err()).
		NotTo(HaveOccurred())
	Expect(tgtCli.Do(ctx, "cluster", "setslot", slot, "importing", srcID).Err()).
		NotTo(HaveOccurred())

	return func() {
		for _, p := range cluster.ports[:3] {
			c := redis.NewClient(&redis.Options{Addr: "127.0.0.1:" + p})
			_ = c.Do(ctx, "cluster", "setslot", slot, "node", srcID).Err()
			_ = c.Close()
		}
		_ = srcCli.Close()
		_ = tgtCli.Close()
	}
}

// ------------------------------------------------------------------------------

var _ = Describe("ClusterClient", func() {
	var failover bool
	var opt *redis.ClusterOptions
	var client *redis.ClusterClient

	assertClusterClient := func() {
		It("do", func() {
			val, err := client.Do(ctx, "ping").Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(val).To(Equal("PONG"))
		})

		It("should GET/SET/DEL", func() {
			err := client.Get(ctx, "A").Err()
			Expect(err).To(Equal(redis.Nil))

			err = client.Set(ctx, "A", "VALUE", 0).Err()
			Expect(err).NotTo(HaveOccurred())

			Eventually(func() string {
				return client.Get(ctx, "A").Val()
			}, 30*time.Second).Should(Equal("VALUE"))

			cnt, err := client.Del(ctx, "A").Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(cnt).To(Equal(int64(1)))
		})

		It("should follow redirects for GET", func() {
			err := client.Set(ctx, "A", "VALUE", 0).Err()
			Expect(err).NotTo(HaveOccurred())

			if !failover {
				Eventually(func() int64 {
					nodes, err := client.Nodes(ctx, "A")
					if err != nil {
						return 0
					}
					return nodes[1].Client.DBSize(ctx).Val()
				}, 30*time.Second).Should(Equal(int64(1)))

				Eventually(func() error {
					return client.SwapNodes(ctx, "A")
				}, 30*time.Second).ShouldNot(HaveOccurred())
			}

			v, err := client.Get(ctx, "A").Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(v).To(Equal("VALUE"))
		})

		It("should follow redirects for SET", func() {
			if !failover {
				Eventually(func() error {
					return client.SwapNodes(ctx, "A")
				}, 30*time.Second).ShouldNot(HaveOccurred())
			}

			err := client.Set(ctx, "A", "VALUE", 0).Err()
			Expect(err).NotTo(HaveOccurred())

			// With RouteByLatency / RouteRandomly the GET may be served by a
			// replica that has not yet replicated the SET. Poll until the
			// value propagates instead of failing on the first attempt.
			Eventually(func() string {
				return client.Get(ctx, "A").Val()
			}, 30*time.Second).Should(Equal("VALUE"))
		})

		It("should distribute keys", func() {
			for i := 0; i < 100; i++ {
				err := client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0).Err()
				Expect(err).NotTo(HaveOccurred())
			}

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				defer GinkgoRecover()
				Eventually(func() string {
					return master.Info(ctx, "keyspace").Val()
				}, 30*time.Second).Should(Or(
					ContainSubstring("keys=32"),
					ContainSubstring("keys=36"),
					ContainSubstring("keys=32"),
				))
				return nil
			})

			Expect(err).NotTo(HaveOccurred())
		})

		It("should distribute keys when using EVAL", func() {
			script := redis.NewScript(`
				local r = redis.call('SET', KEYS[1], ARGV[1])
				return r
			`)

			var key string
			for i := 0; i < 100; i++ {
				key = fmt.Sprintf("key%d", i)
				err := script.Run(ctx, client, []string{key}, "value").Err()
				Expect(err).NotTo(HaveOccurred())
			}

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				defer GinkgoRecover()
				Eventually(func() string {
					return master.Info(ctx, "keyspace").Val()
				}, 30*time.Second).Should(Or(
					ContainSubstring("keys=32"),
					ContainSubstring("keys=36"),
					ContainSubstring("keys=32"),
				))
				return nil
			})

			Expect(err).NotTo(HaveOccurred())
		})

		It("should distribute scripts when using Script Load", func() {
			client.ScriptFlush(ctx)

			script := redis.NewScript(`return 'Unique script'`)

			script.Load(ctx, client)

			err := client.ForEachShard(ctx, func(ctx context.Context, shard *redis.Client) error {
				defer GinkgoRecover()

				val, _ := script.Exists(ctx, shard).Result()
				Expect(val[0]).To(Equal(true))
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
		})

		It("should check all shards when using Script Exists", func() {
			client.ScriptFlush(ctx)

			script := redis.NewScript(`return 'First script'`)
			lostScriptSrc := `return 'Lost script'`
			lostScript := redis.NewScript(lostScriptSrc)

			script.Load(ctx, client)
			// Load one master so SCRIPT EXISTS must aggregate false across shards.
			var once sync.Once
			var loadErr error
			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				once.Do(func() {
					loadErr = master.ScriptLoad(ctx, lostScriptSrc).Err()
				})
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(loadErr).NotTo(HaveOccurred())

			val, _ := client.ScriptExists(ctx, script.Hash(), lostScript.Hash()).Result()

			Expect(val).To(Equal([]bool{true, false}))
		})

		It("should flush scripts from all shards when using ScriptFlush", func() {
			script := redis.NewScript(`return 'Unnecessary script'`)
			script.Load(ctx, client)

			val, _ := client.ScriptExists(ctx, script.Hash()).Result()
			Expect(val).To(Equal([]bool{true}))

			client.ScriptFlush(ctx)

			val, _ = client.ScriptExists(ctx, script.Hash()).Result()
			Expect(val).To(Equal([]bool{false}))
		})

		It("should support Watch", func() {
			var incr func(string) error

			// Transactionally increments key using GET and SET commands.
			incr = func(key string) error {
				err := client.Watch(ctx, func(tx *redis.Tx) error {
					n, err := tx.Get(ctx, key).Int64()
					if err != nil && err != redis.Nil {
						return err
					}

					_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
						pipe.Set(ctx, key, strconv.FormatInt(n+1, 10), 0)
						return nil
					})
					return err
				}, key)
				if err == redis.TxFailedErr {
					return incr(key)
				}
				return err
			}

			var wg sync.WaitGroup
			for i := 0; i < 100; i++ {
				wg.Add(1)
				go func() {
					defer GinkgoRecover()
					defer wg.Done()

					err := incr("key")
					Expect(err).NotTo(HaveOccurred())
				}()
			}
			wg.Wait()

			Eventually(func() string {
				return client.Get(ctx, "key").Val()
			}, 30*time.Second).Should(Equal("100"))
		})

		Describe("pipelining", func() {
			var pipe *redis.Pipeline

			assertPipeline := func(keys []string) {
				It("should follow redirects", func() {
					if !failover {
						for _, key := range keys {
							Eventually(func() error {
								return client.SwapNodes(ctx, key)
							}, 30*time.Second).ShouldNot(HaveOccurred())
						}
					}

					for i, key := range keys {
						pipe.Set(ctx, key, key+"_value", 0)
						pipe.Expire(ctx, key, time.Duration(i+1)*time.Hour)
					}
					cmds, err := pipe.Exec(ctx)
					Expect(err).NotTo(HaveOccurred())
					Expect(cmds).To(HaveLen(14))

					// Check that all keys are set.
					for _, key := range keys {
						Eventually(func() string {
							return client.Get(ctx, key).Val()
						}, 30*time.Second).Should(Equal(key + "_value"))
					}

					if !failover {
						for _, key := range keys {
							Eventually(func() error {
								return client.SwapNodes(ctx, key)
							}, 30*time.Second).ShouldNot(HaveOccurred())
						}
					}

					for _, key := range keys {
						pipe.Get(ctx, key)
						pipe.TTL(ctx, key)
					}
					cmds, err = pipe.Exec(ctx)
					Expect(err).NotTo(HaveOccurred())
					Expect(cmds).To(HaveLen(14))

					for i, key := range keys {
						get := cmds[i*2].(*redis.StringCmd)
						Expect(get.Val()).To(Equal(key + "_value"))

						ttl := cmds[(i*2)+1].(*redis.DurationCmd)
						dur := time.Duration(i+1) * time.Hour
						Expect(ttl.Val()).To(BeNumerically("~", dur, 30*time.Second))
					}
				})

				It("should work with missing keys", func() {
					pipe.Set(ctx, "A{s}", "A_value", 0)
					pipe.Set(ctx, "C{s}", "C_value", 0)
					_, err := pipe.Exec(ctx)
					Expect(err).NotTo(HaveOccurred())

					a := pipe.Get(ctx, "A{s}")
					b := pipe.Get(ctx, "B{s}")
					c := pipe.Get(ctx, "C{s}")
					cmds, err := pipe.Exec(ctx)
					Expect(err).To(Equal(redis.Nil))
					Expect(cmds).To(HaveLen(3))

					Expect(a.Err()).NotTo(HaveOccurred())
					Expect(a.Val()).To(Equal("A_value"))

					Expect(b.Err()).To(Equal(redis.Nil))
					Expect(b.Val()).To(Equal(""))

					Expect(c.Err()).NotTo(HaveOccurred())
					Expect(c.Val()).To(Equal("C_value"))
				})
			}

			Describe("with Pipeline", func() {
				BeforeEach(func() {
					pipe = client.Pipeline().(*redis.Pipeline)
				})

				AfterEach(func() {})

				keys := []string{"A", "B", "C", "D", "E", "F", "G"}
				assertPipeline(keys)

				It("should not fail node with context.Canceled error", func() {
					ctx, cancel := context.WithCancel(context.Background())
					cancel()
					pipe.Set(ctx, "A", "A_value", 0)
					_, err := pipe.Exec(ctx)

					Expect(err).To(HaveOccurred())
					Expect(errors.Is(err, context.Canceled)).To(BeTrue())

					clientNodes, _ := client.Nodes(ctx, "A")

					for _, node := range clientNodes {
						Expect(node.Failing()).To(BeFalse())
					}
				})

				It("should not fail node with context.DeadlineExceeded error", func() {
					ctx, cancel := context.WithTimeout(context.Background(), 1*time.Nanosecond)
					defer cancel()

					pipe.Set(ctx, "A", "A_value", 0)
					_, err := pipe.Exec(ctx)

					Expect(err).To(HaveOccurred())
					Expect(errors.Is(err, context.DeadlineExceeded)).To(BeTrue())

					clientNodes, _ := client.Nodes(ctx, "A")

					for _, node := range clientNodes {
						Expect(node.Failing()).To(BeFalse())
					}
				})
			})

			Describe("with TxPipeline", func() {
				BeforeEach(func() {
					pipe = client.TxPipeline().(*redis.Pipeline)
				})

				AfterEach(func() {})

				// TxPipeline doesn't support cross slot commands.
				// Use hashtag to force all keys to the same slot.
				keys := []string{"A{s}", "B{s}", "C{s}", "D{s}", "E{s}", "F{s}", "G{s}"}
				assertPipeline(keys)

				// make sure CrossSlot error is returned
				It("returns CrossSlot error", func() {
					pipe.Set(ctx, "A{s}", "A_value", 0)
					pipe.Set(ctx, "B{t}", "B_value", 0)
					Expect(hashtag.Slot("A{s}")).NotTo(Equal(hashtag.Slot("B{t}")))
					_, err := pipe.Exec(ctx)
					Expect(err).To(MatchError(redis.ErrCrossSlot))
				})

				It("works normally with keyless commands and no CrossSlot error", func() {
					pipe.Set(ctx, "A{s}", "A_value", 0)
					pipe.Ping(ctx)
					pipe.Set(ctx, "B{s}", "B_value", 0)
					pipe.Ping(ctx)
					_, err := pipe.Exec(ctx)
					Expect(err).To(Not(HaveOccurred()))
				})

				It("reads same-slot streams in transactions and follows MOVED", func() {
					keys := []string{"stream-a{s}", "stream-b{s}"}
					for _, key := range keys {
						Expect(client.XAdd(ctx, &redis.XAddArgs{
							Stream: key, ID: "1-0", Values: map[string]interface{}{"field": "value"},
						}).Err()).NotTo(HaveOccurred())
						Expect(client.XGroupCreate(ctx, key, "streams", "0").Err()).NotTo(HaveOccurred())
					}
					for _, redirect := range []bool{false, true} {
						id := ">"
						if redirect {
							id = "0" // Read the pending entries again after redirecting.
						}
						if redirect && !failover {
							Eventually(func() error { return client.SwapNodes(ctx, keys[0]) }, 30*time.Second).Should(Succeed())
						}
						read := pipe.XRead(ctx, &redis.XReadArgs{Streams: []string{keys[0], keys[1], "0", "0"}, Block: -1})
						group := pipe.XReadGroup(ctx, &redis.XReadGroupArgs{
							Group: "streams", Consumer: "streams", Count: 1,
							Streams: []string{keys[0], keys[1], id, id}, Block: -1,
						})
						_, err := pipe.Exec(ctx)
						Expect(err).NotTo(HaveOccurred())
						Expect(read.Val()).To(HaveLen(2))
						for i, stream := range read.Val() {
							Expect(stream.Stream).To(Equal(keys[i]))
							Expect(stream.Messages).To(HaveLen(1))
						}
						Expect(group.Val()).To(HaveLen(2))
						for _, stream := range group.Val() {
							Expect(stream.Messages).To(HaveLen(1))
						}
					}
				})

				// doesn't fail when no commands are queued
				It("returns no error when there are no commands", func() {
					_, err := pipe.Exec(ctx)
					Expect(err).NotTo(HaveOccurred())
				})
			})
		})

		It("should support PubSub", func() {
			pubsub := client.Subscribe(ctx, "mychannel")
			defer pubsub.Close()

			Eventually(func() error {
				_, err := client.Publish(ctx, "mychannel", "hello").Result()
				if err != nil {
					return err
				}

				msg, err := pubsub.ReceiveTimeout(ctx, time.Second)
				if err != nil {
					return err
				}

				_, ok := msg.(*redis.Message)
				if !ok {
					return fmt.Errorf("got %T, wanted *redis.Message", msg)
				}

				return nil
			}, 30*time.Second).ShouldNot(HaveOccurred())
		})

		It("should support sharded PubSub", func() {
			pubsub := client.SSubscribe(ctx, "mychannel")
			defer pubsub.Close()

			Eventually(func() error {
				_, err := client.SPublish(ctx, "mychannel", "hello").Result()
				if err != nil {
					return err
				}

				msg, err := pubsub.ReceiveTimeout(ctx, time.Second)
				if err != nil {
					return err
				}

				_, ok := msg.(*redis.Message)
				if !ok {
					return fmt.Errorf("got %T, wanted *redis.Message", msg)
				}

				return nil
			}, 30*time.Second).ShouldNot(HaveOccurred())
		})

		It("should support PubSub.Ping without channels", func() {
			pubsub := client.Subscribe(ctx)
			defer pubsub.Close()

			err := pubsub.Ping(ctx)
			Expect(err).NotTo(HaveOccurred())
		})
	}

	Describe("ClusterClient PROTO 2", func() {
		BeforeEach(func() {
			opt = redisClusterOptions()
			opt.Protocol = 2
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(client.Close()).NotTo(HaveOccurred())
		})

		It("should CLUSTER PROTO 2", func() {
			_ = client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
				val, err := c.Do(ctx, "HELLO").Result()
				Expect(err).NotTo(HaveOccurred())
				Expect(val).Should(ContainElements("proto", int64(2)))
				return nil
			})
		})
	})

	Describe("ClusterClient", func() {
		BeforeEach(func() {
			opt = redisClusterOptions()
			opt.ClientName = "cluster_hi"
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(client.Close()).NotTo(HaveOccurred())
		})

		It("should return pool stats", func() {
			stats := client.PoolStats()
			Expect(stats).To(BeAssignableToTypeOf(&redis.PoolStats{}))
		})

		It("should sum pool wait statistics across all nodes", func() {
			// Set a unity pool size to force a wait on any concurrently dispatched requests.
			opt := redisClusterOptions()
			opt.PoolSize = 1
			opt.MinIdleConns = 0
			opt.PoolTimeout = 5 * time.Second
			client := cluster.newClusterClient(ctx, opt)

			state, err := client.LoadState(ctx)
			Expect(err).NotTo(HaveOccurred())
			Expect(state.Masters).NotTo(BeEmpty())

			// Deliberately contend connection acquisition on all per-node connection pools.
			for _, master := range state.Masters {
				nodePool := master.Client.Pool()
				cn, err := nodePool.Get(ctx)
				Expect(err).NotTo(HaveOccurred())
				time.AfterFunc(100*time.Millisecond, func() {
					nodePool.Put(ctx, cn)
				})

				// Get will block until a connection is available; this synchronizes on the Put
				// of the connection obtained from the initial acquisition
				cn2, err := nodePool.Get(ctx)
				Expect(err).NotTo(HaveOccurred())
				nodePool.Put(ctx, cn2)
			}

			var expectedWaits uint32
			var expectedWaitDuration int64

			for _, master := range state.Masters {
				s := master.Client.Pool().Stats()
				expectedWaits += s.WaitCount
				expectedWaitDuration += s.WaitDurationNs
			}

			for _, slave := range state.Slaves {
				s := slave.Client.Pool().Stats()
				expectedWaits += s.WaitCount
				expectedWaitDuration += s.WaitDurationNs
			}

			Expect(expectedWaits).To(BeNumerically(">=", uint32(len(state.Masters))))
			Expect(expectedWaitDuration).To(BeNumerically(">", int64(0)))

			stats := client.PoolStats()
			Expect(stats.WaitCount).To(Equal(expectedWaits))
			Expect(stats.WaitDurationNs).To(Equal(expectedWaitDuration))
		})

		It("should return an error when there are no attempts left", func() {
			opt := redisClusterOptions()
			opt.MaxRedirects = -1
			client := cluster.newClusterClient(ctx, opt)

			Eventually(func() error {
				return client.SwapNodes(ctx, "A")
			}, 30*time.Second).ShouldNot(HaveOccurred())

			err := client.Get(ctx, "A").Err()
			Expect(err).To(HaveOccurred())
			Expect(err.Error()).To(ContainSubstring("MOVED"))

			Expect(client.Close()).NotTo(HaveOccurred())
		})

		It("should determine hash slots correctly for generic commands", func() {
			opt := redisClusterOptions()
			opt.MaxRedirects = -1
			client := cluster.newClusterClient(ctx, opt)

			err := client.Do(ctx, "GET", "A").Err()
			Expect(err).To(Equal(redis.Nil))

			err = client.Do(ctx, []byte("GET"), []byte("A")).Err()
			Expect(err).To(Equal(redis.Nil))

			// Re-swap and re-issue both GETs inside a single Eventually so a
			// background topology reload that reverts the swap before our
			// assertions just causes another iteration instead of a flake.
			Eventually(func() string {
				if err := client.SwapNodes(ctx, "A"); err != nil {
					return err.Error()
				}
				strErr := client.Do(ctx, "GET", "A").Err()
				bytesErr := client.Do(ctx, []byte("GET"), []byte("A")).Err()
				if strErr == nil || bytesErr == nil {
					return ""
				}
				if !strings.Contains(strErr.Error(), "MOVED") {
					return strErr.Error()
				}
				return bytesErr.Error()
			}, 30*time.Second).Should(ContainSubstring("MOVED"))

			Expect(client.Close()).NotTo(HaveOccurred())
		})

		It("should follow node redirection immediately", func() {
			// Configure retry backoffs far in excess of the expected duration of redirection
			opt := redisClusterOptions()
			opt.MinRetryBackoff = 10 * time.Minute
			opt.MaxRetryBackoff = 20 * time.Minute
			client := cluster.newClusterClient(ctx, opt)

			Eventually(func() error {
				return client.SwapNodes(ctx, "A")
			}, 30*time.Second).ShouldNot(HaveOccurred())

			// Note that this context sets a deadline more aggressive than the lowest possible bound
			// of the retry backoff; this verifies that redirection completes immediately.
			redirCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
			defer cancel()

			err := client.Set(redirCtx, "A", "VALUE", 0).Err()
			Expect(err).NotTo(HaveOccurred())

			v, err := client.Get(redirCtx, "A").Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(v).To(Equal("VALUE"))

			Expect(client.Close()).NotTo(HaveOccurred())
		})

		// A TxPipeline whose slot is initially routed at a replica gets -MOVED on
		// the write; the client must follow the redirect and run the whole tx on
		// the real master, with the written values visible afterwards.
		It("should follow MOVED redirect for TxPipeline", func() {
			// {s} hashtag forces a single-slot, single-node transaction.
			_, err := client.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "A{s}", "1", 0)
				pipe.Set(ctx, "B{s}", "2", 0)
				return nil
			})
			Expect(err).NotTo(HaveOccurred())

			// Route the slot at a replica so the write tx is MOVED back to the real master.
			Eventually(func() error {
				return client.SwapNodes(ctx, "A{s}")
			}, 30*time.Second).ShouldNot(HaveOccurred())

			_, err = client.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "A{s}", "11", 0)
				pipe.Set(ctx, "B{s}", "22", 0)
				return nil
			})
			Expect(err).NotTo(HaveOccurred())

			// The MOVED triggers a topology reload that un-swaps the state, so
			// reads eventually land on the real master with the new values.
			Eventually(func() string {
				return client.Get(ctx, "A{s}").Val()
			}, 30*time.Second).Should(Equal("11"))
			Eventually(func() string {
				return client.Get(ctx, "B{s}").Val()
			}, 30*time.Second).Should(Equal("22"))
		})

		// When a queued command fails with a non-redirect error (here GET with
		// wrong arity), Redis aborts the tx with -EXECABORT. The surfaced error
		// should name the triggering command's error ("wrong number of
		// arguments"), not a generic parse error.
		It("should surface the triggering error on TxPipeline EXECABORT instead of 'expected'", func() {
			_, err := client.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "A{s}", "1", 0)
				pipe.Do(ctx, "GET", "A{s}", "extra")
				pipe.Set(ctx, "B{s}", "2", 0)
				return nil
			})
			Expect(err).To(HaveOccurred())
			Expect(err.Error()).NotTo(ContainSubstring("expected '*'"))
			Expect(err.Error()).To(ContainSubstring("wrong number of arguments"))
		})

		// With redirects disabled (MaxRedirects = -1), a TxPipeline that hits
		// -MOVED must surface a MOVED error so the caller can tell it was a
		// redirect, not a generic parse error.
		It("should return MOVED (not 'expected') for TxPipeline when redirects are exhausted", func() {
			opt := redisClusterOptions()
			opt.MaxRedirects = -1
			redirectClient := cluster.newClusterClient(ctx, opt)
			defer func() { Expect(redirectClient.Close()).NotTo(HaveOccurred()) }()

			Eventually(func() error {
				return redirectClient.SwapNodes(ctx, "A{s}")
			}, 30*time.Second).ShouldNot(HaveOccurred())

			_, err := redirectClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "A{s}", "1", 0)
				pipe.Set(ctx, "B{s}", "2", 0)
				return nil
			})
			Expect(err).To(HaveOccurred())
			Expect(err.Error()).To(ContainSubstring("MOVED"))
			Expect(err.Error()).NotTo(ContainSubstring("expected '*'"))
		})

		// A transient write failure during a tx must be retried as a whole
		// transaction and ultimately succeed, with both values actually written.
		// The first tx write is failed with io.EOF and the retry is let through.
		It("retries the whole tx after a transient write failure and succeeds", func() {
			rec := newTxWriteRecorder()
			var failOnce atomic.Int32 // first MULTI write fails, subsequent ones forward
			opt := redisClusterOptions()
			opt.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
				var d net.Dialer
				cn, err := d.DialContext(ctx, network, addr)
				if err != nil {
					return nil, err
				}
				return &txFaultConn{Conn: cn, addr: addr, rec: rec, failOnce: &failOnce}, nil
			}
			retryClient := cluster.newClusterClient(ctx, opt)
			defer func() { Expect(retryClient.Close()).NotTo(HaveOccurred()) }()

			_, err := retryClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "retry1{s}", "1", 0)
				pipe.Set(ctx, "retry2{s}", "2", 0)
				return nil
			})
			Expect(err).NotTo(HaveOccurred(),
				"a transient write failure should be retried as a whole tx and succeed")

			// The retried tx must have executed: both values are present on the
			// slot master.
			Expect(retryClient.Get(ctx, "retry1{s}").Val()).To(Equal("1"))
			Expect(retryClient.Get(ctx, "retry2{s}").Val()).To(Equal("2"))

			// The slot master received more than one tx write, so the first
			// attempt really failed and the tx was retried.
			srcAddr := "127.0.0.1:" + slotMasterPort(hashtag.Slot("retry1{s}"))
			Expect(rec.txWriteCount(srcAddr)).To(BeNumerically(">=", 2),
				"the first tx write failed and the client should have retried the whole tx")
		})

		// During an ASK redirect, a transient connection failure to the
		// importing target must be retried with the ASKING flag preserved, so the
		// tx still executes on the target.
		It("preserves ASKING when retrying a conn failure on the ASK target", func() {
			key := "askretry{s}"
			slot := hashtag.Slot(key)
			defer resetSlot(ctx, slot, key)
			cleanup := setupAskMigration(ctx, slot)
			defer cleanup()

			tgtAddr := "127.0.0.1:" + slotOtherMasterPort(slot)
			rec := newTxWriteRecorder()
			var failOnce atomic.Int32 // first MULTI write to the target fails, then forwards
			opt := redisClusterOptions()
			opt.MaxRedirects = 2 // source-ASK (0) + target-connfail (1) + target-success (2)
			opt.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
				var d net.Dialer
				cn, err := d.DialContext(ctx, network, addr)
				if err != nil {
					return nil, err
				}
				return &txFaultConn{Conn: cn, addr: addr, rec: rec, failOnce: &failOnce, failAddr: tgtAddr}, nil
			}
			askClient := cluster.newClusterClient(ctx, opt)
			defer func() { Expect(askClient.Close()).NotTo(HaveOccurred()) }()

			_, err := askClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, key, "v", 0)
				return nil
			})
			Expect(err).NotTo(HaveOccurred(),
				"a transient write failure on the ASK target should be retried with ASKING preserved and succeed")

			// The tx executed on the target (with ASKING); a Get follows the
			// ASK back to the target and reads the value.
			Expect(askClient.Get(ctx, key).Val()).To(Equal("v"))

			// The target received >=2 tx writes: the failed first attempt and
			// the successful retry, both with a leading ASKING.
			Expect(rec.txWriteCount(tgtAddr)).To(BeNumerically(">=", 2))
		})

		// A read failure after the tx batch was written must not be retried: the
		// server may have committed the transaction, and re-sending the batch
		// would double-execute non-idempotent commands. The batch should be sent
		// exactly once.
		It("does not retry a read failure after the tx batch is written", func() {
			rec := newTxWriteRecorder()
			opt := redisClusterOptions()
			opt.MaxRedirects = 3
			opt.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
				var d net.Dialer
				cn, err := d.DialContext(ctx, network, addr)
				if err != nil {
					return nil, err
				}
				// failRead returns EOF from every read once a MULTI write has
				// been seen on the conn: the batch is committed server-side but
				// the client cannot read the reply.
				return &txFaultConn{Conn: cn, addr: addr, rec: rec, failRead: true}, nil
			}
			readClient := cluster.newClusterClient(ctx, opt)
			defer func() { Expect(readClient.Close()).NotTo(HaveOccurred()) }()

			_, err := readClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "rdretry1{s}", "1", 0)
				pipe.Set(ctx, "rdretry2{s}", "2", 0)
				return nil
			})
			Expect(err).To(HaveOccurred(),
				"a read failure after the batch was written should surface an error")

			// The batch was written exactly once: the read failure is fatal, so
			// the tx is not re-sent (no double-execution).
			srcAddr := "127.0.0.1:" + slotMasterPort(hashtag.Slot("rdretry1{s}"))
			Expect(rec.txWriteCount(srcAddr)).To(Equal(1),
				"a post-commit read failure must not re-send the tx batch")
		})

		It("should call fn for every master node", func() {
			for i := range 10 {
				Expect(client.Set(ctx, strconv.Itoa(i), "", 0).Err()).NotTo(HaveOccurred())
			}

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			size, err := client.DBSize(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(size).To(Equal(int64(0)))
		})

		It("should CLUSTER SLOTS", func() {
			res, err := client.ClusterSlots(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(HaveLen(3))

			wanted := []redis.ClusterSlot{{
				Start: 0,
				End:   5460,
				Nodes: []redis.ClusterNode{{
					ID:   "",
					Addr: "127.0.0.1:16600",
				}, {
					ID:   "",
					Addr: "127.0.0.1:16603",
				}},
			}, {
				Start: 5461,
				End:   10922,
				Nodes: []redis.ClusterNode{{
					ID:   "",
					Addr: "127.0.0.1:16601",
				}, {
					ID:   "",
					Addr: "127.0.0.1:16604",
				}},
			}, {
				Start: 10923,
				End:   16383,
				Nodes: []redis.ClusterNode{{
					ID:   "",
					Addr: "127.0.0.1:16602",
				}, {
					ID:   "",
					Addr: "127.0.0.1:16605",
				}},
			}}
			Expect(assertSlotsEqual(res, wanted)).NotTo(HaveOccurred())
		})

		It("should CLUSTER SHARDS", func() {
			res, err := client.ClusterShards(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(res).NotTo(BeEmpty())

			// Iterate over the ClusterShard results and validate the fields.
			for _, shard := range res {
				Expect(shard.Slots).NotTo(BeEmpty())
				for _, slotRange := range shard.Slots {
					Expect(slotRange.Start).To(BeNumerically(">=", 0))
					Expect(slotRange.End).To(BeNumerically(">=", slotRange.Start))
				}

				Expect(shard.Nodes).NotTo(BeEmpty())
				for _, node := range shard.Nodes {
					Expect(node.ID).NotTo(BeEmpty())
					Expect(node.Endpoint).NotTo(BeEmpty())
					Expect(node.IP).NotTo(BeEmpty())
					Expect(node.Port).To(BeNumerically(">", 0))

					validRoles := []string{"master", "slave", "replica"}
					Expect(validRoles).To(ContainElement(node.Role))

					Expect(node.ReplicationOffset).To(BeNumerically(">=", 0))

					validHealthStatuses := []string{"online", "fail", "failed", "loading"}
					Expect(validHealthStatuses).To(ContainElement(node.Health))
				}
			}
		})

		It("should CLUSTER LINKS", func() {
			res, err := client.ClusterLinks(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(res).NotTo(BeEmpty())

			// Iterate over the ClusterLink results and validate the map keys.
			for _, link := range res {

				Expect(link.Direction).NotTo(BeEmpty())
				Expect([]string{"from", "to"}).To(ContainElement(link.Direction))
				Expect(link.Node).NotTo(BeEmpty())
				Expect(link.CreateTime).To(BeNumerically(">", 0))

				Expect(link.Events).NotTo(BeEmpty())
				validEventChars := []rune{'r', 'w'}
				for _, eventChar := range link.Events {
					Expect(validEventChars).To(ContainElement(eventChar))
				}

				Expect(link.SendBufferAllocated).To(BeNumerically(">=", 0))
				Expect(link.SendBufferUsed).To(BeNumerically(">=", 0))
			}
		})

		It("should cluster client setname", func() {
			err := client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
				return c.Ping(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			_ = client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
				val, err := c.ClientList(ctx).Result()
				Expect(err).NotTo(HaveOccurred())
				Expect(val).Should(ContainSubstring("name=cluster_hi"))
				return nil
			})
		})

		It("should CLUSTER PROTO 3", func() {
			_ = client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
				val, err := c.Do(ctx, "HELLO").Result()
				Expect(err).NotTo(HaveOccurred())
				Expect(val).Should(HaveKeyWithValue("proto", int64(3)))
				return nil
			})
		})

		It("should CLUSTER MYSHARDID", func() {
			shardID, err := client.ClusterMyShardID(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(shardID).ToNot(BeEmpty())
		})

		It("should CLUSTER NODES", func() {
			res, err := client.ClusterNodes(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(len(res)).To(BeNumerically(">", 400))
		})

		It("should CLUSTER INFO", func() {
			res, err := client.ClusterInfo(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(ContainSubstring("cluster_known_nodes:6"))
		})

		It("should CLUSTER KEYSLOT", func() {
			hashSlot, err := client.ClusterKeySlot(ctx, "somekey").Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(hashSlot).To(Equal(int64(hashtag.Slot("somekey"))))
		})

		It("should CLUSTER GETKEYSINSLOT", func() {
			keys, err := client.ClusterGetKeysInSlot(ctx, hashtag.Slot("somekey"), 1).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(len(keys)).To(Equal(0))
		})

		It("should CLUSTER COUNT-FAILURE-REPORTS", func() {
			n, err := client.ClusterCountFailureReports(ctx, cluster.nodeIDs[0]).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(n).To(Equal(int64(0)))
		})

		It("should CLUSTER COUNTKEYSINSLOT", func() {
			n, err := client.ClusterCountKeysInSlot(ctx, 10).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(n).To(Equal(int64(0)))
		})

		It("should CLUSTER SAVECONFIG", func() {
			res, err := client.ClusterSaveConfig(ctx).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(Equal("OK"))
		})

		It("should CLUSTER SLAVES", func() {
			nodesList, err := client.ClusterSlaves(ctx, cluster.nodeIDs[0]).Result()
			Expect(err).NotTo(HaveOccurred())
			Expect(nodesList).Should(ContainElement(ContainSubstring("slave")))
			Expect(nodesList).Should(HaveLen(1))
		})

		It("should RANDOMKEY", func() {
			const nkeys = 100

			for i := 0; i < nkeys; i++ {
				err := client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0).Err()
				Expect(err).NotTo(HaveOccurred())
			}

			var keys []string
			addKey := func(key string) {
				for _, k := range keys {
					if k == key {
						return
					}
				}
				keys = append(keys, key)
			}

			for i := 0; i < nkeys*10; i++ {
				key, err := client.RandomKey(ctx).Result()
				Expect(err).NotTo(HaveOccurred())
				addKey(key)
			}

			Expect(len(keys)).To(BeNumerically("~", nkeys, nkeys/10))
		})

		It("should support Process hook", func() {
			testCtx, cancel := context.WithCancel(ctx)
			defer cancel()

			err := client.Ping(ctx).Err()
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
				return node.Ping(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			var mu sync.Mutex
			var stack []string

			clusterHook := &hook{
				processHook: func(hook redis.ProcessHook) redis.ProcessHook {
					return func(ctx context.Context, cmd redis.Cmder) error {
						select {
						case <-testCtx.Done():
							return hook(ctx, cmd)
						default:
						}

						Expect(cmd.String()).To(Equal("ping: "))
						mu.Lock()
						stack = append(stack, "cluster.BeforeProcess")
						mu.Unlock()

						err := hook(ctx, cmd)

						Expect(cmd.String()).To(Equal("ping: PONG"))
						mu.Lock()
						stack = append(stack, "cluster.AfterProcess")
						mu.Unlock()

						return err
					}
				},
			}
			client.AddHook(clusterHook)

			nodeHook := &hook{
				processHook: func(hook redis.ProcessHook) redis.ProcessHook {
					return func(ctx context.Context, cmd redis.Cmder) error {
						select {
						case <-testCtx.Done():
							return hook(ctx, cmd)
						default:
						}

						Expect(cmd.String()).To(Equal("ping: "))
						mu.Lock()
						stack = append(stack, "shard.BeforeProcess")
						mu.Unlock()

						err := hook(ctx, cmd)

						Expect(cmd.String()).To(Equal("ping: PONG"))
						mu.Lock()
						stack = append(stack, "shard.AfterProcess")
						mu.Unlock()

						return err
					}
				},
			}

			_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
				node.AddHook(nodeHook)
				return nil
			})

			err = client.Ping(ctx).Err()
			Expect(err).NotTo(HaveOccurred())

			mu.Lock()
			finalStack := make([]string, len(stack))
			copy(finalStack, stack)
			mu.Unlock()

			Expect(finalStack).To(ContainElements([]string{
				"cluster.BeforeProcess",
				"shard.BeforeProcess",
				"shard.AfterProcess",
				"cluster.AfterProcess",
			}))
		})

		It("should support Pipeline hook", func() {
			err := client.Ping(ctx).Err()
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
				return node.Ping(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			var mu sync.Mutex
			var stack []string

			client.AddHook(&hook{
				processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
					return func(ctx context.Context, cmds []redis.Cmder) error {
						Expect(cmds).To(HaveLen(1))
						cmdStr := cmds[0].String()

						// Handle SET command (should succeed)
						if cmdStr == "set pipeline_test_key pipeline_test_value: " {
							mu.Lock()
							stack = append(stack, "cluster.BeforeProcessPipeline")
							mu.Unlock()

							err := hook(ctx, cmds)

							Expect(cmds).To(HaveLen(1))
							Expect(cmds[0].String()).To(Equal("set pipeline_test_key pipeline_test_value: OK"))
							mu.Lock()
							stack = append(stack, "cluster.AfterProcessPipeline")
							mu.Unlock()

							return err
						}

						// For other commands (like ping), just pass through without expectations
						// since they might fail before reaching this point
						return hook(ctx, cmds)
					}
				},
			})

			_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
				node.AddHook(&hook{
					processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
						return func(ctx context.Context, cmds []redis.Cmder) error {
							defer GinkgoRecover()
							// skip the connection initialization: the node's
							// pipeline-pool connection initializes lazily, and
							// its handshake pipeline runs through this hook chain
							if len(cmds) == 0 || cmds[0].Name() == "hello" || cmds[0].Name() == "client" {
								return hook(ctx, cmds)
							}
							Expect(cmds).To(HaveLen(1))
							cmdStr := cmds[0].String()

							// Handle SET command (should succeed)
							if cmdStr == "set pipeline_test_key pipeline_test_value: " {
								mu.Lock()
								stack = append(stack, "shard.BeforeProcessPipeline")
								mu.Unlock()

								err := hook(ctx, cmds)

								Expect(cmds).To(HaveLen(1))
								Expect(cmds[0].String()).To(Equal("set pipeline_test_key pipeline_test_value: OK"))
								mu.Lock()
								stack = append(stack, "shard.AfterProcessPipeline")
								mu.Unlock()

								return err
							}

							// For other commands (like ping), just pass through without expectations
							return hook(ctx, cmds)
						}
					},
				})
				return nil
			})

			_, err = client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Set(ctx, "pipeline_test_key", "pipeline_test_value", 0)
				return nil
			})
			Expect(err).NotTo(HaveOccurred())

			mu.Lock()
			finalStack := make([]string, len(stack))
			copy(finalStack, stack)
			mu.Unlock()

			Expect(finalStack).To(Equal([]string{
				"cluster.BeforeProcessPipeline",
				"shard.BeforeProcessPipeline",
				"shard.AfterProcessPipeline",
				"cluster.AfterProcessPipeline",
			}))
		})

		It("should support TxPipeline hook", func() {
			err := client.Ping(ctx).Err()
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
				return node.Ping(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			var mu sync.Mutex
			var stack []string

			client.AddHook(&hook{
				processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
					return func(ctx context.Context, cmds []redis.Cmder) error {
						Expect(cmds).To(HaveLen(3))
						Expect(cmds[1].String()).To(Equal("ping: "))
						mu.Lock()
						stack = append(stack, "cluster.BeforeProcessPipeline")
						mu.Unlock()

						err := hook(ctx, cmds)

						Expect(cmds).To(HaveLen(3))
						Expect(cmds[1].String()).To(Equal("ping: PONG"))
						mu.Lock()
						stack = append(stack, "cluster.AfterProcessPipeline")
						mu.Unlock()

						return err
					}
				},
			})

			_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
				node.AddHook(&hook{
					processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
						return func(ctx context.Context, cmds []redis.Cmder) error {
							defer GinkgoRecover()
							// skip the connection initialization: the node's
							// pipeline-pool connection initializes lazily, and
							// its handshake pipeline runs through this hook chain
							if len(cmds) == 0 || cmds[0].Name() == "hello" || cmds[0].Name() == "client" {
								return hook(ctx, cmds)
							}
							Expect(cmds).To(HaveLen(3))
							Expect(cmds[1].String()).To(Equal("ping: "))
							mu.Lock()
							stack = append(stack, "shard.BeforeProcessPipeline")
							mu.Unlock()

							err := hook(ctx, cmds)

							Expect(cmds).To(HaveLen(3))
							Expect(cmds[1].String()).To(Equal("ping: PONG"))
							mu.Lock()
							stack = append(stack, "shard.AfterProcessPipeline")
							mu.Unlock()

							return err
						}
					},
				})
				return nil
			})

			_, err = client.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Ping(ctx)
				return nil
			})
			Expect(err).NotTo(HaveOccurred())

			mu.Lock()
			finalStack := make([]string, len(stack))
			copy(finalStack, stack)
			mu.Unlock()

			Expect(finalStack).To(Equal([]string{
				"cluster.BeforeProcessPipeline",
				"shard.BeforeProcessPipeline",
				"shard.AfterProcessPipeline",
				"cluster.AfterProcessPipeline",
			}))
		})

		It("should return correct replica for key", func() {
			client, err := client.SlaveForKey(ctx, "test")
			Expect(err).ToNot(HaveOccurred())
			info := client.Info(ctx, "server")
			Expect(info.Val()).Should(ContainSubstring("tcp_port:16604"))
		})

		It("should return correct master for key", func() {
			client, err := client.MasterForKey(ctx, "test")
			Expect(err).ToNot(HaveOccurred())
			info := client.Info(ctx, "server")
			Expect(info.Val()).Should(ContainSubstring("tcp_port:16601"))
		})

		It("should not retry Watch callback errors wrapping io.EOF", func() {
			opt := redisClusterOptions()
			opt.MaxRedirects = 1
			opt.MinRetryBackoff = -1
			opt.MaxRetryBackoff = -1

			client := cluster.newClusterClient(ctx, opt)
			defer func() {
				Expect(client.Close()).NotTo(HaveOccurred())
			}()

			calls := 0
			err := client.Watch(ctx, func(tx *redis.Tx) error {
				calls++
				return fmt.Errorf("external call failed: %w", io.EOF)
			}, "watch-callback-net-error")

			Expect(errors.Is(err, io.EOF)).To(BeTrue())
			Expect(calls).To(Equal(1))
		})

		assertClusterClient()
	})

	Describe("ClusterClient with RouteByLatency", func() {
		BeforeEach(func() {
			opt = redisClusterOptions()
			opt.RouteByLatency = true
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				Eventually(func() int64 {
					return client.DBSize(ctx).Val()
				}, 30*time.Second).Should(Equal(int64(0)))
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			err := client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				return slave.ReadWrite(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.Close()
			Expect(err).NotTo(HaveOccurred())
		})

		assertClusterClient()
	})

	Describe("ClusterClient with RouteByLatencyTolerance", func() {
		BeforeEach(func() {
			opt = redisClusterOptions()
			opt.RouteByLatency = true
			// Far wider than any intra-cluster spread, so every healthy node for a slot is a
			// candidate and the per-slot rotation is the thing under test.
			opt.RouteByLatencyTolerance = time.Second
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				Eventually(func() int64 {
					return client.DBSize(ctx).Val()
				}, 30*time.Second).Should(Equal(int64(0)))
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			err := client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				return slave.ReadWrite(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.Close()
			Expect(err).NotTo(HaveOccurred())
		})

		It("should read keys from distinct shards", func() {
			// Discovered rather than hardcoded: keyN does not spread over the three masters
			// (key2 and key3 both hash into 0-5460), and hardcoded keys would silently stop
			// covering multiple shards if the slot ranges ever moved.
			keys := make([]string, 0, 3)
			slots := map[int64]string{}
			masters := map[string]string{}

			for i := 0; len(masters) < 3 && i < 100; i++ {
				key := fmt.Sprintf("tolerance-key-%d", i)

				nodes, err := client.Nodes(ctx, key)
				Expect(err).NotTo(HaveOccurred())
				Expect(nodes).NotTo(BeEmpty())

				master := nodes[0].Client.Options().Addr
				if _, seen := masters[master]; seen {
					continue
				}
				masters[master] = key

				slot := client.ClusterKeySlot(ctx, key).Val()
				Expect(slots).NotTo(HaveKey(slot))
				slots[slot] = key
				keys = append(keys, key)
			}
			Expect(masters).To(HaveLen(3), "could not find keys on three distinct masters: %v", masters)

			for _, key := range keys {
				Expect(client.Set(ctx, key, "VALUE-"+key, 0).Err()).NotTo(HaveOccurred())
			}

			// Reads go to whichever in-band node the rotation picks, including replicas that
			// may not have caught up yet, so poll rather than failing on the first attempt.
			for _, key := range keys {
				Eventually(func() string {
					return client.Get(ctx, key).Val()
				}, 30*time.Second).Should(Equal("VALUE-" + key))
			}

			// Repeated reads must keep resolving to a usable node for every slot. This is what
			// regressed when the rotation cursor was shared across slots.
			for i := 0; i < 50; i++ {
				for _, key := range keys {
					Eventually(func() string {
						return client.Get(ctx, key).Val()
					}, 30*time.Second).Should(Equal("VALUE-" + key))
				}
			}
		})

		assertClusterClient()
	})

	Describe("ClusterClient with ClusterSlots", func() {
		BeforeEach(func() {
			failover = true

			opt = redisClusterOptions()
			opt.ClusterSlots = func(ctx context.Context) ([]redis.ClusterSlot, error) {
				slots := []redis.ClusterSlot{{
					Start: 0,
					End:   5460,
					Nodes: []redis.ClusterNode{{
						Addr: ":" + ringShard1Port,
					}},
				}, {
					Start: 5461,
					End:   10922,
					Nodes: []redis.ClusterNode{{
						Addr: ":" + ringShard2Port,
					}},
				}, {
					Start: 10923,
					End:   16383,
					Nodes: []redis.ClusterNode{{
						Addr: ":" + ringShard3Port,
					}},
				}}
				return slots, nil
			}
			client = cluster.newClusterClient(ctx, opt)
			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				Eventually(func() int64 {
					return client.DBSize(ctx).Val()
				}, 30*time.Second).Should(Equal(int64(0)))
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			failover = false

			err := client.Close()
			Expect(err).NotTo(HaveOccurred())
		})

		assertClusterClient()
	})

	Describe("ClusterClient with RouteRandomly and ClusterSlots", func() {
		BeforeEach(func() {
			failover = true

			opt = redisClusterOptions()
			opt.RouteRandomly = true
			opt.ClusterSlots = func(ctx context.Context) ([]redis.ClusterSlot, error) {
				slots := []redis.ClusterSlot{{
					Start: 0,
					End:   5460,
					Nodes: []redis.ClusterNode{{
						Addr: ":" + ringShard1Port,
					}},
				}, {
					Start: 5461,
					End:   10922,
					Nodes: []redis.ClusterNode{{
						Addr: ":" + ringShard2Port,
					}},
				}, {
					Start: 10923,
					End:   16383,
					Nodes: []redis.ClusterNode{{
						Addr: ":" + ringShard3Port,
					}},
				}}
				return slots, nil
			}
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				Eventually(func() int64 {
					return client.DBSize(ctx).Val()
				}, 30*time.Second).Should(Equal(int64(0)))
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			failover = false

			err := client.Close()
			Expect(err).NotTo(HaveOccurred())
		})

		assertClusterClient()
	})

	Describe("ClusterClient with ClusterSlots with multiple nodes per slot", func() {
		BeforeEach(func() {
			failover = true

			opt = redisClusterOptions()
			opt.ReadOnly = true
			opt.ClusterSlots = func(ctx context.Context) ([]redis.ClusterSlot, error) {
				slots := []redis.ClusterSlot{{
					Start: 0,
					End:   5460,
					Nodes: []redis.ClusterNode{{
						Addr: ":16600",
					}, {
						Addr: ":16603",
					}},
				}, {
					Start: 5461,
					End:   10922,
					Nodes: []redis.ClusterNode{{
						Addr: ":16601",
					}, {
						Addr: ":16604",
					}},
				}, {
					Start: 10923,
					End:   16383,
					Nodes: []redis.ClusterNode{{
						Addr: ":16602",
					}, {
						Addr: ":16605",
					}},
				}}
				return slots, nil
			}
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.FlushDB(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
				Eventually(func() int64 {
					return client.DBSize(ctx).Val()
				}, 30*time.Second).Should(Equal(int64(0)))
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
		})

		AfterEach(func() {
			failover = false

			err := client.Close()
			Expect(err).NotTo(HaveOccurred())
		})

		assertClusterClient()
	})
})

// With cluster-preferred-endpoint-type unknown-endpoint the servers send a
// null host in CLUSTER SLOTS and an empty host in MOVED and ASK.
// The MOVED check passes only because the test cluster is on loopback: the
// client dials an empty-host target ":port" on the local machine, so a
// redirect from a remote host is not covered. Taking the host of the node
// that sent the redirect is left for a follow-up.
var _ = Describe("ClusterClient with unknown-endpoint", func() {
	const param = "cluster-preferred-endpoint-type"

	var client *redis.ClusterClient
	var saved map[string]string

	BeforeEach(func() {
		client, saved = nil, nil
		// Redis 8.0 and 8.2 cluster nodes segfault in a module timer
		// shortly after this setting is applied, which takes down the
		// shared test cluster for every later spec.
		SkipBeforeRedisVersion("8.4", "cluster nodes crash with unknown-endpoint")
		saved = map[string]string{}
		for port, node := range cluster.clients {
			old, err := node.ConfigGet(ctx, param).Result()
			Expect(err).NotTo(HaveOccurred())
			saved[port] = old[param]
			Expect(node.ConfigSet(ctx, param, "unknown-endpoint").Err()).NotTo(HaveOccurred())
		}
		client = cluster.newClusterClientUnstable(redisClusterOptions())
	})

	AfterEach(func() {
		// Restore the setting even if BeforeEach failed before the client
		// was built; AfterEach also runs after the version skip.
		for port, value := range saved {
			Expect(cluster.clients[port].ConfigSet(ctx, param, value).Err()).NotTo(HaveOccurred())
		}
		if client == nil {
			return
		}
		_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
			return master.FlushDB(ctx).Err()
		})
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should load the topology and follow a MOVED redirect on loopback", func() {
		slots, err := cluster.masters()[0].ClusterSlots(ctx).Result()
		Expect(err).NotTo(HaveOccurred())
		for _, slot := range slots {
			for _, node := range slot.Nodes {
				Expect(node.Addr).To(HavePrefix(":"))
			}
		}

		state, err := client.LoadState(ctx)
		Expect(err).NotTo(HaveOccurred())
		for _, master := range state.Masters {
			Expect(master.Client.Options().Addr).To(HavePrefix("127.0.0.1:"))
		}

		// A (slot 6373) and C (slot 14503) belong to different masters.
		for _, key := range []string{"A", "C"} {
			Expect(client.Set(ctx, key, "VALUE", 0).Err()).NotTo(HaveOccurred())
			Expect(client.Get(ctx, key).Val()).To(Equal("VALUE"))
		}

		// Route the slot of A at its replica, which answers with
		// -MOVED <slot> :<port>; see the note above the Describe.
		Expect(client.SwapNodes(ctx, "A")).NotTo(HaveOccurred())
		Expect(client.Set(ctx, "A", "MOVED", 0).Err()).NotTo(HaveOccurred())
		Expect(client.Get(ctx, "A").Val()).To(Equal("MOVED"))
	})
})

var _ = Describe("ClusterClient without nodes", func() {
	var client *redis.ClusterClient

	BeforeEach(func() {
		client = redis.NewClusterClient(&redis.ClusterOptions{})
	})

	AfterEach(func() {
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should return an error for Ping", func() {
		err := client.Ping(ctx).Err()
		Expect(err).To(MatchError("redis: cluster has no nodes"))
	})

	It("should return an error for pipeline", func() {
		_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.Ping(ctx)
			return nil
		})
		Expect(err).To(MatchError("redis: cluster has no nodes"))
	})
})

var _ = Describe("ClusterClient without valid nodes", func() {
	var client *redis.ClusterClient

	BeforeEach(func() {
		client = redis.NewClusterClient(&redis.ClusterOptions{
			Addrs: []string{redisAddr},
		})
	})

	AfterEach(func() {
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should return an error when cluster support is disabled", func() {
		err := client.Ping(ctx).Err()
		Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
	})

	It("should return an error for pipeline when cluster support is disabled", func() {
		_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.Ping(ctx)
			return nil
		})
		Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
	})
})

var _ = Describe("ClusterClient with unavailable Cluster", func() {
	var client *redis.ClusterClient

	BeforeEach(func() {
		opt := redisClusterOptions()
		opt.ReadTimeout = 250 * time.Millisecond
		opt.WriteTimeout = 250 * time.Millisecond
		opt.MaxRedirects = 1
		client = cluster.newClusterClientUnstable(opt)
		Expect(client.Ping(ctx).Err()).NotTo(HaveOccurred())

		for _, node := range cluster.clients {
			err := node.ClientPause(ctx, 5*time.Second).Err()
			Expect(err).NotTo(HaveOccurred())
		}
	})

	AfterEach(func() {
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should recover when Cluster recovers", func() {
		err := client.Ping(ctx).Err()
		Expect(err).To(HaveOccurred())

		Eventually(func() error {
			return client.Ping(ctx).Err()
		}, "30s").ShouldNot(HaveOccurred())
	})
})

var _ = Describe("ClusterClient timeout", func() {
	var client *redis.ClusterClient

	AfterEach(func() {
		_ = client.Close()
	})

	testTimeout := func() {
		It("should timeout Ping", func() {
			err := client.Ping(ctx).Err()
			Expect(err).To(HaveOccurred())
			Expect(err.(net.Error).Timeout()).To(BeTrue())
		})

		It("should timeout Pipeline", func() {
			_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.Ping(ctx)
				return nil
			})
			Expect(err).To(HaveOccurred())
			Expect(err.(net.Error).Timeout()).To(BeTrue())
		})

		It("should timeout Tx", func() {
			err := client.Watch(ctx, func(tx *redis.Tx) error {
				return tx.Ping(ctx).Err()
			}, "foo")
			Expect(err).To(HaveOccurred())
			Expect(err.(net.Error).Timeout()).To(BeTrue())
		})

		It("should timeout Tx Pipeline", func() {
			err := client.Watch(ctx, func(tx *redis.Tx) error {
				_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
					pipe.Ping(ctx)
					return nil
				})
				return err
			}, "foo")
			Expect(err).To(HaveOccurred())
			Expect(err.(net.Error).Timeout()).To(BeTrue())
		})
	}

	const pause = 5 * time.Second

	Context("read/write timeout", func() {
		BeforeEach(func() {
			opt := redisClusterOptions()
			client = cluster.newClusterClient(ctx, opt)

			err := client.ForEachShard(ctx, func(ctx context.Context, client *redis.Client) error {
				err := client.ClientPause(ctx, pause).Err()

				opt := client.Options()
				opt.ReadTimeout = time.Nanosecond
				opt.WriteTimeout = time.Nanosecond

				return err
			})
			Expect(err).NotTo(HaveOccurred())

			// Overwrite timeouts after the client is initialized.
			opt.ReadTimeout = time.Nanosecond
			opt.WriteTimeout = time.Nanosecond
			opt.MaxRedirects = 0
		})

		AfterEach(func() {
			_ = client.ForEachShard(ctx, func(ctx context.Context, client *redis.Client) error {
				defer GinkgoRecover()

				opt := client.Options()
				opt.ReadTimeout = time.Second
				opt.WriteTimeout = time.Second

				Eventually(func() error {
					return client.Ping(ctx).Err()
				}, 2*pause).ShouldNot(HaveOccurred())
				return nil
			})

			err := client.Close()
			Expect(err).NotTo(HaveOccurred())
		})

		testTimeout()
	})
})

var _ = Describe("Command Tips tests", func() {
	var client *redis.ClusterClient

	BeforeEach(func() {
		opt := redisClusterOptions()
		client = cluster.newClusterClient(ctx, opt)
		client.SetCommandInfoResolver(client.NewDynamicResolver())
	})

	AfterEach(func() {
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	for _, metadataPolicies := range []bool{false, true} {
		mode := "default policies"
		if metadataPolicies {
			mode = "metadata policies"
		}
		Describe("routing compatibility with "+mode, func() {
			const firstKey, secondKey = "routing-review:{a}:one", "routing-review:{b}:two"
			const serializedKey = "routing-review:{a}:serialized"
			configure := func(c *redis.ClusterClient) {
				// The command-tips suite installs a dynamic resolver. Reset it
				// explicitly so each case exercises the intended public mode.
				if metadataPolicies {
					c.SetCommandInfoResolver(redis.NewCommandMetadataResolver())
				} else {
					c.SetCommandInfoResolver(redis.NewDefaultCommandPolicyResolver())
				}
			}
			BeforeEach(func() {
				configure(client)
				Expect(client.Set(ctx, firstKey, "one", 0).Err()).NotTo(HaveOccurred())
				Expect(client.Set(ctx, secondKey, "two", 0).Err()).NotTo(HaveOccurred())
			})
			AfterEach(func() {
				for _, key := range []string{firstKey, secondKey, serializedKey} {
					Expect(client.Del(ctx, key).Err()).NotTo(HaveOccurred())
				}
			})

			It("keeps ordinary direct commands available and makes cross-slot fan-out opt-in", func() {
				Expect(client.Ping(ctx).Val()).To(Equal("PONG"))
				info, err := client.Info(ctx, "server").Result()
				Expect(err).NotTo(HaveOccurred())
				Expect(info).To(ContainSubstring("redis_version:"))
				_, _, err = client.Scan(ctx, 0, "routing-review:*", 10).Result()
				Expect(err).NotTo(HaveOccurred())
				values, err := client.MGet(ctx, secondKey, firstKey).Result()
				if metadataPolicies {
					Expect(err).NotTo(HaveOccurred())
					Expect(values).To(Equal([]interface{}{"two", "one"}))
				} else {
					Expect(err).To(HaveOccurred())
					Expect(err.Error()).To(ContainSubstring("CROSSSLOT"))
				}
			})

			It("runs a mixed pipeline despite one cross-slot command", func() {
				pipe := client.Pipeline()
				first := pipe.Get(ctx, firstKey)
				ping := pipe.Ping(ctx)
				size := pipe.DBSize(ctx)
				script := pipe.ScriptLoad(ctx, "return 1")
				info := pipe.Info(ctx, "server")
				crossSlot := pipe.MGet(ctx, firstKey, secondKey)
				last := pipe.Get(ctx, secondKey)
				cmds, err := pipe.Exec(ctx)
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("CROSSSLOT"))
				for _, cmd := range cmds {
					if cmd == crossSlot {
						Expect(cmd.Err()).To(HaveOccurred())
					} else {
						Expect(cmd.Err()).NotTo(HaveOccurred())
					}
				}
				Expect(first.Val()).To(Equal("one"))
				Expect(last.Val()).To(Equal("two"))
				Expect(ping.Val()).To(Equal("PONG"))
				Expect(size.Val()).To(BeNumerically(">=", 0))
				Expect(script.Val()).To(HaveLen(40))
				Expect(info.Val()).To(ContainSubstring("redis_version:"))
			})

			It("runs administrative commands on the transaction connection", func() {
				pipe := client.TxPipeline()
				value := pipe.Get(ctx, firstKey)
				ping := pipe.Ping(ctx)
				size := pipe.DBSize(ctx)
				script := pipe.ScriptLoad(ctx, "return 1")
				info := pipe.Info(ctx, "server")
				cmds, err := pipe.Exec(ctx)
				Expect(err).NotTo(HaveOccurred())
				for _, cmd := range cmds {
					Expect(cmd.Err()).NotTo(HaveOccurred())
				}
				Expect(value.Val()).To(Equal("one"))
				Expect(ping.Val()).To(Equal("PONG"))
				Expect(size.Val()).To(BeNumerically(">=", 1))
				Expect(script.Val()).To(HaveLen(40))
				Expect(info.Val()).To(ContainSubstring("redis_version:"))
			})

			It("runs SORT transactions with incomplete key metadata", func() {
				Expect(client.Del(ctx, firstKey).Err()).NotTo(HaveOccurred())
				Expect(client.RPush(ctx, firstKey, "3", "1", "2").Err()).NotTo(HaveOccurred())
				pipe := client.TxPipeline()
				sorted := pipe.Sort(ctx, firstKey, &redis.Sort{})
				stored := pipe.SortStore(ctx, firstKey, serializedKey, &redis.Sort{})
				_, err := pipe.Exec(ctx)
				Expect(err).NotTo(HaveOccurred())
				Expect(sorted.Val()).To(Equal([]string{"1", "2", "3"}))
				Expect(stored.Val()).To(Equal(int64(3)))
				Expect(client.LRange(ctx, serializedKey, 0, -1).Val()).To(Equal([]string{"1", "2", "3"}))
			})

			It("accepts BinaryMarshaler values when MSET needs only one slot", func() {
				Expect(client.MSet(ctx, firstKey, numberStruct{Number: 7}).Err()).NotTo(HaveOccurred())
				Expect(client.Get(ctx, firstKey).Val()).To(Equal(`{"Number":7}`))
				Expect(client.MSet(ctx,
					firstKey, numberStruct{Number: 8}, serializedKey, numberStruct{Number: 9},
				).Err()).NotTo(HaveOccurred())
				values, err := client.MGet(ctx, firstKey, serializedKey).Result()
				Expect(err).NotTo(HaveOccurred())
				Expect(values).To(Equal([]interface{}{`{"Number":8}`, `{"Number":9}`}))
			})

			It("enforces fan-out ordering across AutoPipeline modes", func() {
				for _, fullDuplex := range []bool{false, true} {
					opt := *client.Options()
					apClient := redis.NewClusterClient(&opt)
					configure(apClient)
					defer apClient.Close()
					cfg := &redis.AutoPipelineOptions{FullDuplex: fullDuplex}
					blocking, err := apClient.AutoPipelineWithOptions(cfg)
					Expect(err).NotTo(HaveOccurred())
					async, err := apClient.AsyncAutoPipelineWithOptions(cfg)
					Expect(err).NotTo(HaveOccurred())
					for i, ap := range []*redis.AutoPipeliner{blocking, async} {
						By(fmt.Sprintf("full duplex=%v, async=%v", fullDuplex, i == 1))
						ping := ap.Ping(ctx)
						values := ap.MGet(ctx, secondKey, firstKey)
						sameSlot := ap.MGet(ctx, firstKey, firstKey)
						value := ap.Get(ctx, firstKey)
						size := ap.DBSize(ctx)
						if metadataPolicies && i == 1 {
							Expect(ping.Err()).To(MatchError(ContainSubstring("ordered AsyncAutoPipeline")))
						} else {
							Expect(ping.Err()).NotTo(HaveOccurred())
							Expect(ping.Val()).To(Equal("PONG"))
						}
						Expect(sameSlot.Err()).NotTo(HaveOccurred())
						Expect(sameSlot.Val()).To(Equal([]interface{}{"one", "one"}))
						Expect(value.Err()).NotTo(HaveOccurred())
						Expect(value.Val()).To(Equal("one"))
						Expect(size.Err()).NotTo(HaveOccurred())
						Expect(size.Val()).To(BeNumerically(">=", 0))
						if metadataPolicies {
							Expect(size.Val()).To(BeNumerically(">=", 2))
							if i == 1 {
								Expect(values.Err()).To(MatchError(ContainSubstring("ordered AsyncAutoPipeline")))
							} else {
								Expect(values.Err()).NotTo(HaveOccurred())
								Expect(values.Val()).To(Equal([]interface{}{"two", "one"}))
							}
						} else {
							Expect(values.Err()).To(HaveOccurred())
							Expect(values.Err().Error()).To(ContainSubstring("CROSSSLOT"))
						}
					}
				}
			})
		})
	}

	It("should verify COMMAND tips match router policy types", func() {
		SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")
		expectedPolicies := map[string]struct {
			RequestPolicy  string
			ResponsePolicy string
		}{
			"touch": {
				RequestPolicy:  "multi_shard",
				ResponsePolicy: "agg_sum",
			},
			"flushall": {
				RequestPolicy:  "all_shards",
				ResponsePolicy: "all_succeeded",
			},
		}

		cmds, err := client.Command(ctx).Result()
		Expect(err).NotTo(HaveOccurred())

		for cmdName, expected := range expectedPolicies {
			actualCmd := cmds[cmdName]

			Expect(actualCmd.CommandPolicy).NotTo(BeNil())

			// Verify request_policy from COMMAND matches router policy
			actualRequestPolicy := actualCmd.CommandPolicy.Request.String()
			Expect(actualRequestPolicy).To(Equal(expected.RequestPolicy))

			// Verify response_policy from COMMAND matches router policy
			actualResponsePolicy := actualCmd.CommandPolicy.Response.String()
			Expect(actualResponsePolicy).To(Equal(expected.ResponsePolicy))
		}
	})

	Describe("Explicit Routing Policy Tests", func() {
		It("should test explicit routing policy for TOUCH", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Verify TOUCH command has multi_shard policy
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			touchCmd := cmds["touch"]

			Expect(touchCmd.CommandPolicy).NotTo(BeNil())
			Expect(touchCmd.CommandPolicy.Request.String()).To(Equal("multi_shard"))
			Expect(touchCmd.CommandPolicy.Response.String()).To(Equal("agg_sum"))

			keys := []string{"key1", "key2", "key3", "key4", "key5"}
			for _, key := range keys {
				err := client.Set(ctx, key, "value", 0).Err()
				Expect(err).NotTo(HaveOccurred())
			}

			result := client.Touch(ctx, keys...)
			Expect(result.Err()).NotTo(HaveOccurred())
			Expect(result.Val()).To(Equal(int64(len(keys))))
		})

		It("should test explicit routing policy for FLUSHALL", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Verify FLUSHALL command has all_shards policy
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			flushallCmd := cmds["flushall"]

			Expect(flushallCmd.CommandPolicy).NotTo(BeNil())
			Expect(flushallCmd.CommandPolicy.Request.String()).To(Equal("all_shards"))
			Expect(flushallCmd.CommandPolicy.Response.String()).To(Equal("all_succeeded"))

			testKeys := []string{"test1", "test2", "test3"}
			for _, key := range testKeys {
				err := client.Set(ctx, key, "value", 0).Err()
				Expect(err).NotTo(HaveOccurred())
			}

			err = client.FlushAll(ctx).Err()
			Expect(err).NotTo(HaveOccurred())

			for _, key := range testKeys {
				exists := client.Exists(ctx, key)
				Expect(exists.Val()).To(Equal(int64(0)))
			}
		})

		It("should test explicit routing policy for PING", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Verify PING command has all_shards policy
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			pingCmd := cmds["ping"]
			Expect(pingCmd.CommandPolicy).NotTo(BeNil())
			Expect(pingCmd.CommandPolicy.Request.String()).To(Equal("all_shards"))
			Expect(pingCmd.CommandPolicy.Response.String()).To(Equal("all_succeeded"))

			result := client.Ping(ctx)
			Expect(result.Err()).NotTo(HaveOccurred())
			Expect(result.Val()).To(Equal("PONG"))
		})

		It("should test explicit routing policy for DBSIZE", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Verify DBSIZE command has all_shards policy with agg_sum response
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			dbsizeCmd := cmds["dbsize"]
			Expect(dbsizeCmd.CommandPolicy).NotTo(BeNil())
			Expect(dbsizeCmd.CommandPolicy.Request.String()).To(Equal("all_shards"))
			Expect(dbsizeCmd.CommandPolicy.Response.String()).To(Equal("agg_sum"))

			testKeys := []string{"dbsize_test1", "dbsize_test2", "dbsize_test3"}
			for _, key := range testKeys {
				err := client.Set(ctx, key, "value", 0).Err()
				Expect(err).NotTo(HaveOccurred())
			}

			size := client.DBSize(ctx)
			Expect(size.Err()).NotTo(HaveOccurred())
			Expect(size.Val()).To(BeNumerically(">=", int64(len(testKeys))))
		})
	})

	Describe("DDL Commands Routing Policy Tests", func() {
		BeforeEach(func() {
			info := client.Info(ctx, "modules")
			Expect(info.Err()).NotTo(HaveOccurred())
			if !strings.Contains(info.Val(), "search") {
				Skip("Search module not available")
			}
		})

		It("should test DDL commands routing policy for FT.CREATE", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Verify FT.CREATE command routing policy
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			ftCreateCmd, exists := cmds["ft.create"]
			if !exists || ftCreateCmd.CommandPolicy == nil {
				Skip("FT.CREATE command or tips not available")
			}

			// DDL commands should NOT be broadcasted - they should go to coordinator only
			Expect(ftCreateCmd.CommandPolicy).NotTo(BeNil())
			requestPolicy := ftCreateCmd.CommandPolicy.Request.String()
			Expect(requestPolicy).NotTo(Equal("all_shards"))
			Expect(requestPolicy).NotTo(Equal("all_nodes"))

			indexName := "test_index_create"
			client.FTDropIndex(ctx, indexName)

			result := client.FTCreate(ctx, indexName,
				&redis.FTCreateOptions{
					OnHash: true,
					Prefix: []interface{}{"doc:"},
				},
				&redis.FieldSchema{
					FieldName: "title",
					FieldType: redis.SearchFieldTypeText,
				})
			Expect(result.Err()).NotTo(HaveOccurred())
			Expect(result.Val()).To(Equal("OK"))

			infoResult := client.FTInfo(ctx, indexName)
			Expect(infoResult.Err()).NotTo(HaveOccurred())
			Expect(infoResult.Val().IndexName).To(Equal(indexName))
			client.FTDropIndex(ctx, indexName)
		})

		It("should test DDL commands routing policy for FT.ALTER", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Verify FT.ALTER command routing policy
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			ftAlterCmd, exists := cmds["ft.alter"]
			if !exists || ftAlterCmd.CommandPolicy == nil {
				Skip("FT.ALTER command or tips not available")
			}

			Expect(ftAlterCmd.CommandPolicy).NotTo(BeNil())
			requestPolicy := ftAlterCmd.CommandPolicy.Request.String()
			Expect(requestPolicy).NotTo(Equal("all_shards"))
			Expect(requestPolicy).NotTo(Equal("all_nodes"))

			indexName := "test_index_alter"
			client.FTDropIndex(ctx, indexName)

			result := client.FTCreate(ctx, indexName,
				&redis.FTCreateOptions{
					OnHash: true,
					Prefix: []interface{}{"doc:"},
				},
				&redis.FieldSchema{
					FieldName: "title",
					FieldType: redis.SearchFieldTypeText,
				})
			Expect(result.Err()).NotTo(HaveOccurred())

			alterResult := client.FTAlter(ctx, indexName, false,
				[]interface{}{"description", redis.SearchFieldTypeText.String()})
			Expect(alterResult.Err()).NotTo(HaveOccurred())
			Expect(alterResult.Val()).To(Equal("OK"))
			client.FTDropIndex(ctx, indexName)
		})

		It("should route keyed commands to correct shard based on hash slot", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			type masterNode struct {
				client *redis.Client
				addr   string
			}
			var masterNodes []masterNode
			var mu sync.Mutex

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				addr := master.Options().Addr
				mu.Lock()
				masterNodes = append(masterNodes, masterNode{
					client: master,
					addr:   addr,
				})
				mu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(len(masterNodes)).To(BeNumerically(">", 1))

			// Single keyed command should go to exactly one shard - determined by hash slot
			testKey := "test_key_12345"
			testValue := "test_value"

			result := client.Set(ctx, testKey, testValue, 0)
			Expect(result.Err()).NotTo(HaveOccurred())
			Expect(result.Val()).To(Equal("OK"))

			time.Sleep(200 * time.Millisecond)

			var targetNodeAddr string
			foundNodes := 0

			for _, node := range masterNodes {
				getResult := node.client.Get(ctx, testKey)
				if getResult.Err() == nil && getResult.Val() == testValue {
					foundNodes++
					targetNodeAddr = node.addr
				} else {
				}
			}

			Expect(foundNodes).To(Equal(1))
			Expect(targetNodeAddr).NotTo(BeEmpty())

			// Multiple commands with same key should go to same shard
			finalValue := ""
			for i := 0; i < 5; i++ {
				finalValue = fmt.Sprintf("value_%d", i)
				result := client.Set(ctx, testKey, finalValue, 0)
				Expect(result.Err()).NotTo(HaveOccurred())
				Expect(result.Val()).To(Equal("OK"))
			}

			time.Sleep(200 * time.Millisecond)

			var currentTargetNode string
			foundNodesAfterUpdate := 0

			for _, node := range masterNodes {
				getResult := node.client.Get(ctx, testKey)
				if getResult.Err() == nil && getResult.Val() == finalValue {
					foundNodesAfterUpdate++
					currentTargetNode = node.addr
				} else {
				}
			}

			// All commands with same key should go to same shard
			Expect(foundNodesAfterUpdate).To(Equal(1))
			Expect(currentTargetNode).To(Equal(targetNodeAddr))
		})

		It("should aggregate responses according to explicit aggregation policies", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			type masterNode struct {
				client *redis.Client
				addr   string
			}
			var masterNodes []masterNode
			var mu sync.Mutex

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				addr := master.Options().Addr
				mu.Lock()
				masterNodes = append(masterNodes, masterNode{
					client: master,
					addr:   addr,
				})
				mu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(len(masterNodes)).To(BeNumerically(">", 1))

			// verify TOUCH command has agg_sum policy
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			touchCmd, exists := cmds["touch"]
			if !exists || touchCmd.CommandPolicy == nil {
				Skip("TOUCH command or tips not available")
			}

			Expect(touchCmd.CommandPolicy.Response.String()).To(Equal("agg_sum"))

			testKeys := []string{
				"touch_test_key_1111", // These keys should map to different hash slots
				"touch_test_key_2222",
				"touch_test_key_3333",
				"touch_test_key_4444",
				"touch_test_key_5555",
			}

			// Set keys on different shards
			keysPerShard := make(map[string][]string)
			for _, key := range testKeys {
				result := client.Set(ctx, key, "test_value", 0)
				Expect(result.Err()).NotTo(HaveOccurred())

				// Find which shard contains this key
				for _, node := range masterNodes {
					getResult := node.client.Get(ctx, key)
					if getResult.Err() == nil {
						keysPerShard[node.addr] = append(keysPerShard[node.addr], key)
						break
					}
				}
			}

			// Verify keys are distributed across multiple shards
			shardsWithKeys := len(keysPerShard)
			Expect(shardsWithKeys).To(BeNumerically(">", 1))

			// Execute TOUCH command on all keys - this should aggregate results using agg_sum
			touchResult := client.Touch(ctx, testKeys...)
			Expect(touchResult.Err()).NotTo(HaveOccurred())

			totalTouched := touchResult.Val()
			Expect(totalTouched).To(Equal(int64(len(testKeys))))

			totalKeysOnShards := 0
			for _, keys := range keysPerShard {
				totalKeysOnShards += len(keys)
			}

			Expect(totalKeysOnShards).To(Equal(len(testKeys)))

			// FLUSHALL command with all_succeeded aggregation policy
			flushallCmd, exists := cmds["flushall"]
			if !exists || flushallCmd.CommandPolicy == nil {
				Skip("FLUSHALL command or tips not available")
			}

			Expect(flushallCmd.CommandPolicy.Response.String()).To(Equal("all_succeeded"))

			for i := 0; i < len(masterNodes); i++ {
				testKey := fmt.Sprintf("flush_test_key_%d_%d", i, time.Now().UnixNano())
				result := client.Set(ctx, testKey, "test_data", 0)
				Expect(result.Err()).NotTo(HaveOccurred())
			}

			flushResult := client.FlushAll(ctx)
			Expect(flushResult.Err()).NotTo(HaveOccurred())
			Expect(flushResult.Val()).To(Equal("OK"))

			for _, node := range masterNodes {
				dbSizeResult := node.client.DBSize(ctx)
				Expect(dbSizeResult.Err()).NotTo(HaveOccurred())
				Expect(dbSizeResult.Val()).To(Equal(int64(0)))
			}

			// WAIT command aggregation policy - verify agg_min policy
			waitCmd, exists := cmds["wait"]
			if !exists || waitCmd.CommandPolicy == nil {
				Skip("WAIT command or tips not available")
			}

			Expect(waitCmd.CommandPolicy.Response.String()).To(Equal("agg_min"))

			// Set up some data to replicate
			testKey := "wait_test_key_1111"
			result := client.Set(ctx, testKey, "test_value", 0)
			Expect(result.Err()).NotTo(HaveOccurred())

			// Execute WAIT command - should aggregate using agg_min across all shards
			// WAIT waits for a given number of replicas to acknowledge writes
			// With agg_min policy, it returns the minimum number of replicas that acknowledged
			waitResult := client.Wait(ctx, 0, 1000) // Wait for 0 replicas with 1 second timeout
			Expect(waitResult.Err()).NotTo(HaveOccurred())

			// The result should be the minimum number of replicas across all shards
			// Since we're asking for 0 replicas, all shards should return 0, so min is 0
			minReplicas := waitResult.Val()
			Expect(minReplicas).To(BeNumerically(">=", 0))

			// SCRIPT EXISTS command aggregation policy - verify agg_logical_and policy
			scriptExistsCmd, exists := cmds["script exists"]
			if !exists || scriptExistsCmd.CommandPolicy == nil {
				Skip("SCRIPT EXISTS command or tips not available")
			}

			Expect(scriptExistsCmd.CommandPolicy.Response.String()).To(Equal("agg_logical_and"))

			// Load a script on all shards
			testScript := "return 'hello'"
			scriptLoadResult := client.ScriptLoad(ctx, testScript)
			Expect(scriptLoadResult.Err()).NotTo(HaveOccurred())
			scriptSHA := scriptLoadResult.Val()

			// Verify script exists on all shards using SCRIPT EXISTS
			// With agg_logical_and policy, it should return true only if script exists on ALL shards
			scriptExistsResult := client.ScriptExists(ctx, scriptSHA)
			Expect(scriptExistsResult.Err()).NotTo(HaveOccurred())

			existsResults := scriptExistsResult.Val()
			Expect(len(existsResults)).To(Equal(1))
			Expect(existsResults[0]).To(BeTrue()) // Script should exist on all shards

			// Test with a non-existent script SHA
			nonExistentSHA := "0000000000000000000000000000000000000000"
			scriptExistsResult2 := client.ScriptExists(ctx, nonExistentSHA)
			Expect(scriptExistsResult2.Err()).NotTo(HaveOccurred())

			existsResults2 := scriptExistsResult2.Val()
			Expect(len(existsResults2)).To(Equal(1))
			Expect(existsResults2[0]).To(BeFalse()) // Script should not exist on any shard

			// Test with mixed scenario - flush scripts from one shard manually
			// This is harder to test in practice since SCRIPT FLUSH affects all shards
			// So we'll just verify the basic functionality works
		})

		It("should verify command aggregation policies", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			commandPolicies := map[string]string{
				"touch":         "agg_sum",
				"flushall":      "all_succeeded",
				"pfcount":       "default(hashslot)",
				"exists":        "agg_sum",
				"script exists": "agg_logical_and",
				"wait":          "agg_min",
			}

			for cmdName, expectedPolicy := range commandPolicies {
				cmd, exists := cmds[cmdName]
				if !exists {
					continue
				}

				if cmd.CommandPolicy == nil {
					continue
				}

				actualPolicy := cmd.CommandPolicy.Response.String()
				Expect(actualPolicy).To(Equal(expectedPolicy))
			}
		})

		It("should properly aggregate responses from keyless commands executed on multiple shards", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			type masterNode struct {
				client *redis.Client
				addr   string
			}
			var masterNodes []masterNode
			var mu sync.Mutex

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				addr := master.Options().Addr
				mu.Lock()
				masterNodes = append(masterNodes, masterNode{
					client: master,
					addr:   addr,
				})
				mu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(len(masterNodes)).To(BeNumerically(">", 1))

			// PING command with all_shards policy - should aggregate responses
			cmds, err := client.Command(ctx).Result()
			Expect(err).NotTo(HaveOccurred())

			pingCmd, exists := cmds["ping"]
			if exists && pingCmd.CommandPolicy != nil {
			}

			pingResult := client.Ping(ctx)
			Expect(pingResult.Err()).NotTo(HaveOccurred())
			Expect(pingResult.Val()).To(Equal("PONG"))

			// Verify PING was executed on all shards by checking individual nodes
			for _, node := range masterNodes {
				nodePingResult := node.client.Ping(ctx)
				Expect(nodePingResult.Err()).NotTo(HaveOccurred())
				Expect(nodePingResult.Val()).To(Equal("PONG"))
			}

			// Test 2: DBSIZE command aggregation across shards - verify agg_sum policy
			testKeys := []string{
				"dbsize_test_key_1111",
				"dbsize_test_key_2222",
				"dbsize_test_key_3333",
				"dbsize_test_key_4444",
			}

			for _, key := range testKeys {
				result := client.Set(ctx, key, "test_value", 0)
				Expect(result.Err()).NotTo(HaveOccurred())
			}

			dbSizeResult := client.DBSize(ctx)
			Expect(dbSizeResult.Err()).NotTo(HaveOccurred())

			totalSize := dbSizeResult.Val()
			Expect(totalSize).To(BeNumerically(">=", int64(len(testKeys))))

			// Verify aggregation by manually getting sizes from each shard
			totalManualSize := int64(0)

			for _, node := range masterNodes {
				nodeDbSizeResult := node.client.DBSize(ctx)
				Expect(nodeDbSizeResult.Err()).NotTo(HaveOccurred())

				nodeSize := nodeDbSizeResult.Val()
				totalManualSize += nodeSize
			}

			// Verify aggregation worked correctly
			Expect(totalSize).To(Equal(totalManualSize))
		})

		It("should properly aggregate responses from keyed commands executed on multiple shards", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")
			type masterNode struct {
				client *redis.Client
				addr   string
			}
			var masterNodes []masterNode
			var mu sync.Mutex

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				addr := master.Options().Addr
				mu.Lock()
				masterNodes = append(masterNodes, masterNode{
					client: master,
					addr:   addr,
				})
				mu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(len(masterNodes)).To(BeNumerically(">", 1))

			// MGET command with keys on different shards
			testData := map[string]string{
				"mget_test_key_1111": "value1",
				"mget_test_key_2222": "value2",
				"mget_test_key_3333": "value3",
				"mget_test_key_4444": "value4",
				"mget_test_key_5555": "value5",
			}

			keyLocations := make(map[string]string)
			for key, value := range testData {
				result := client.Set(ctx, key, value, 0)
				Expect(result.Err()).NotTo(HaveOccurred())

				for _, node := range masterNodes {
					getResult := node.client.Get(ctx, key)
					if getResult.Err() == nil && getResult.Val() == value {
						keyLocations[key] = node.addr
						break
					}
				}
			}

			shardsUsed := make(map[string]bool)
			for _, shardAddr := range keyLocations {
				shardsUsed[shardAddr] = true
			}
			Expect(len(shardsUsed)).To(BeNumerically(">", 1))

			keys := make([]string, 0, len(testData))
			expectedValues := make([]interface{}, 0, len(testData))
			for key, value := range testData {
				keys = append(keys, key)
				expectedValues = append(expectedValues, value)
			}

			mgetResult := client.MGet(ctx, keys...)
			Expect(mgetResult.Err()).NotTo(HaveOccurred())

			actualValues := mgetResult.Val()
			Expect(len(actualValues)).To(Equal(len(expectedValues)))
			for i, value := range actualValues {
				if value != nil {
					Expect(value).To(Equal(expectedValues[i]))
				} else {
					Expect(value).To(BeNil())
				}
			}

			// EXISTS command aggregation across multiple keys
			existsTestData := map[string]string{
				"exists_agg_key_1111": "value1",
				"exists_agg_key_2222": "value2",
				"exists_agg_key_3333": "value3",
			}

			existsKeys := make([]string, 0, len(existsTestData))
			for key, value := range existsTestData {
				result := client.Set(ctx, key, value, 0)
				Expect(result.Err()).NotTo(HaveOccurred())
				existsKeys = append(existsKeys, key)
			}

			// Add a non-existent key to the list
			nonExistentKey := "non_existent_key_9999"
			existsKeys = append(existsKeys, nonExistentKey)

			existsResult := client.Exists(ctx, existsKeys...)
			Expect(existsResult.Err()).NotTo(HaveOccurred())

			existsCount := existsResult.Val()
			Expect(existsCount).To(Equal(int64(len(existsTestData))))
		})

		It("should propagate coordinator errors to client without modification", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			type masterNode struct {
				client *redis.Client
				addr   string
			}
			var masterNodes []masterNode
			var mu sync.Mutex

			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				addr := master.Options().Addr
				mu.Lock()
				masterNodes = append(masterNodes, masterNode{
					client: master,
					addr:   addr,
				})
				mu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(len(masterNodes)).To(BeNumerically(">", 0))

			invalidSlotResult := client.ClusterAddSlotsRange(ctx, 99999, 100000)
			coordinatorErr := invalidSlotResult.Err()

			if coordinatorErr != nil {
				// Verify the error is a Redis error
				var redisErr redis.Error
				Expect(errors.As(coordinatorErr, &redisErr)).To(BeTrue())

				// Verify error message is preserved exactly as returned by coordinator
				errorMsg := coordinatorErr.Error()
				Expect(errorMsg).To(SatisfyAny(
					ContainSubstring("slot"),
					ContainSubstring("ERR"),
					ContainSubstring("Invalid"),
				))

				// Test that the same error occurs when calling coordinator directly
				coordinatorNode := masterNodes[0]
				directResult := coordinatorNode.client.ClusterAddSlotsRange(ctx, 99999, 100000)
				directErr := directResult.Err()

				if directErr != nil {
					Expect(coordinatorErr.Error()).To(Equal(directErr.Error()))
				}
			}

			// Try cluster forget with invalid node ID
			invalidNodeID := "invalid_node_id_12345"
			forgetResult := client.ClusterForget(ctx, invalidNodeID)
			forgetErr := forgetResult.Err()

			if forgetErr != nil {
				var redisErr redis.Error
				Expect(errors.As(forgetErr, &redisErr)).To(BeTrue())

				errorMsg := forgetErr.Error()
				Expect(errorMsg).To(SatisfyAny(
					ContainSubstring("Unknown node"),
					ContainSubstring("Invalid node"),
					ContainSubstring("ERR"),
				))

				coordinatorNode := masterNodes[0]
				directForgetResult := coordinatorNode.client.ClusterForget(ctx, invalidNodeID)
				directForgetErr := directForgetResult.Err()

				if directForgetErr != nil {
					Expect(forgetErr.Error()).To(Equal(directForgetErr.Error()))
				}
			}

			// Test error type preservation and format
			keySlotResult := client.ClusterKeySlot(ctx, "")
			keySlotErr := keySlotResult.Err()

			if keySlotErr != nil {
				var redisErr redis.Error
				Expect(errors.As(keySlotErr, &redisErr)).To(BeTrue())

				errorMsg := keySlotErr.Error()
				Expect(len(errorMsg)).To(BeNumerically(">", 0))
				Expect(errorMsg).NotTo(ContainSubstring("wrapped"))
				Expect(errorMsg).NotTo(ContainSubstring("context"))
			}

			// Verify error propagation consistency
			clusterInfoResult := client.ClusterInfo(ctx)
			clusterInfoErr := clusterInfoResult.Err()

			if clusterInfoErr != nil {
				var redisErr redis.Error
				Expect(errors.As(clusterInfoErr, &redisErr)).To(BeTrue())

				coordinatorNode := masterNodes[0]
				directInfoResult := coordinatorNode.client.ClusterInfo(ctx)
				directInfoErr := directInfoResult.Err()

				if directInfoErr != nil {
					Expect(clusterInfoErr.Error()).To(Equal(directInfoErr.Error()))
				}
			}

			// Verify no error modification in router
			invalidReplicateResult := client.ClusterReplicate(ctx, "00000000000000000000000000000000invalid00")
			invalidReplicateErr := invalidReplicateResult.Err()

			if invalidReplicateErr != nil {
				var redisErr redis.Error
				Expect(errors.As(invalidReplicateErr, &redisErr)).To(BeTrue())

				errorMsg := invalidReplicateErr.Error()
				Expect(errorMsg).NotTo(ContainSubstring("router"))
				Expect(errorMsg).NotTo(ContainSubstring("cluster client"))
				Expect(errorMsg).NotTo(ContainSubstring("failed to execute"))

				Expect(errorMsg).To(SatisfyAny(
					HavePrefix("ERR"),
					ContainSubstring("Invalid"),
					ContainSubstring("Unknown"),
				))
			}
		})

		Describe("Routing Policies Comprehensive Test Suite", func() {
			It("should test MGET command with multi-slot routing and key order preservation", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Set up test data across multiple shards
				testData := map[string]string{
					"mget_test_key_1111": "value1",
					"mget_test_key_2222": "value2",
					"mget_test_key_3333": "value3",
					"mget_test_key_4444": "value4",
					"mget_test_key_5555": "value5",
				}

				// Set all keys
				for key, value := range testData {
					Expect(client.Set(ctx, key, value, 0).Err()).NotTo(HaveOccurred())
				}

				// Verify keys are distributed across multiple shards
				slotMap := make(map[int]bool)
				for key := range testData {
					slot := hashtag.Slot(key)
					slotMap[slot] = true
				}
				Expect(len(slotMap)).To(BeNumerically(">", 1))

				// Test MGET with specific key order
				keys := []string{
					"mget_test_key_3333",
					"mget_test_key_1111",
					"mget_test_key_5555",
					"mget_test_key_2222",
					"mget_test_key_4444",
				}

				result := client.MGet(ctx, keys...)
				Expect(result.Err()).NotTo(HaveOccurred())

				// Verify values are returned in the same order as keys
				values := result.Val()
				Expect(len(values)).To(Equal(len(keys)))
				for i, key := range keys {
					Expect(values[i]).To(Equal(testData[key]))
				}
			})

			It("should test MGET with non-existent keys across multiple shards", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Set up some keys
				Expect(client.Set(ctx, "mget_exists_1111", "value1", 0).Err()).NotTo(HaveOccurred())
				Expect(client.Set(ctx, "mget_exists_3333", "value3", 0).Err()).NotTo(HaveOccurred())

				// MGET with mix of existing and non-existing keys
				keys := []string{
					"mget_exists_1111",
					"mget_nonexist_2222",
					"mget_exists_3333",
					"mget_nonexist_4444",
				}

				result := client.MGet(ctx, keys...)
				Expect(result.Err()).ToNot(HaveOccurred())

				// MGET returns nil for non-existent keys (not errors)
				// Values should be in the same order as requested keys
				values := result.Val()
				Expect(len(values)).To(Equal(4))
				Expect(values[0]).To(Equal("value1")) // existing key
				Expect(values[1]).To(BeNil())         // non-existent key returns nil
				Expect(values[2]).To(Equal("value3")) // existing key
				Expect(values[3]).To(BeNil())         // non-existent key returns nil
			})

			It("should test TOUCH command with multi-slot routing", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Set up keys across multiple shards
				keys := []string{
					"touch_test_key_1111",
					"touch_test_key_2222",
					"touch_test_key_3333",
					"touch_test_key_4444",
				}

				// Set all keys
				for _, key := range keys {
					Expect(client.Set(ctx, key, "value", 0).Err()).NotTo(HaveOccurred())
				}

				// Verify keys are on different shards
				slotMap := make(map[int]bool)
				for _, key := range keys {
					slot := hashtag.Slot(key)
					slotMap[slot] = true
				}
				Expect(len(slotMap)).To(BeNumerically(">", 1))

				// TOUCH should work across multiple shards
				result := client.Touch(ctx, keys...)
				Expect(result.Err()).NotTo(HaveOccurred())
				Expect(result.Val()).To(Equal(int64(len(keys))))
			})

			It("should test DEL command with multi-slot routing", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Set up keys across multiple shards
				keys := []string{
					"del_test_key_1111",
					"del_test_key_2222",
					"del_test_key_3333",
				}

				// Set all keys
				for _, key := range keys {
					Expect(client.Set(ctx, key, "value", 0).Err()).NotTo(HaveOccurred())
				}

				// Verify keys are on different shards
				slotMap := make(map[int]bool)
				for _, key := range keys {
					slot := hashtag.Slot(key)
					slotMap[slot] = true
				}
				Expect(len(slotMap)).To(BeNumerically(">", 1))

				// DEL should work across multiple shards
				result := client.Del(ctx, keys...)
				Expect(result.Err()).NotTo(HaveOccurred())
				Expect(result.Val()).To(Equal(int64(len(keys))))

				// Verify all keys were deleted
				for _, key := range keys {
					val := client.Get(ctx, key)
					Expect(val.Err()).To(Equal(redis.Nil))
				}
			})

			It("should test DBSIZE command with agg_sum aggregation across all shards", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Set keys across multiple shards
				keys := []string{
					"dbsize_test_1111",
					"dbsize_test_2222",
					"dbsize_test_3333",
					"dbsize_test_4444",
					"dbsize_test_5555",
				}

				// Clean up first
				client.Del(ctx, keys...)

				// Set all keys
				for _, key := range keys {
					Expect(client.Set(ctx, key, "value", 0).Err()).NotTo(HaveOccurred())
				}

				// DBSIZE should aggregate results from all shards
				result := client.DBSize(ctx)
				Expect(result.Err()).NotTo(HaveOccurred())
				Expect(result.Val()).To(BeNumerically(">=", int64(len(keys))))
			})

			It("should test PING command with all_shards routing", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// PING should be sent to all shards and return one successful response
				result := client.Ping(ctx)
				Expect(result.Err()).NotTo(HaveOccurred())
				Expect(result.Val()).To(Equal("PONG"))
			})

			It("should test MGET with single shard optimization", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Use hash tags to ensure all keys are on the same shard
				keys := []string{
					"{sameslot}key1",
					"{sameslot}key2",
					"{sameslot}key3",
				}

				// Verify all keys hash to the same slot
				slot := hashtag.Slot(keys[0])
				for _, key := range keys {
					Expect(hashtag.Slot(key)).To(Equal(slot))
				}

				// Set all keys
				for i, key := range keys {
					Expect(client.Set(ctx, key, fmt.Sprintf("value%d", i+1), 0).Err()).NotTo(HaveOccurred())
				}

				// MGET should work even with single shard
				result := client.MGet(ctx, keys...)
				Expect(result.Err()).NotTo(HaveOccurred())

				values := result.Val()
				Expect(len(values)).To(Equal(3))
				Expect(values[0]).To(Equal("value1"))
				Expect(values[1]).To(Equal("value2"))
				Expect(values[2]).To(Equal("value3"))
			})

			It("should test empty MGET command returns error", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// MGET with no keys should return an error
				result := client.MGet(ctx)
				Expect(result.Err()).To(HaveOccurred())
				Expect(result.Err().Error()).To(ContainSubstring("cannot determine all key arguments for multi-shard command mget"))
			})

			It("should test MGET integration with MSET across multiple shards", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Create test data
				testData := map[string]string{
					"integration_key_1111": "alpha",
					"integration_key_2222": "beta",
					"integration_key_3333": "gamma",
					"integration_key_4444": "delta",
					"integration_key_5555": "epsilon",
				}

				// Verify keys are on different shards
				slotMap := make(map[int]bool)
				for key := range testData {
					slot := hashtag.Slot(key)
					slotMap[slot] = true
				}
				Expect(len(slotMap)).To(BeNumerically(">", 1))

				// Use individual SET commands instead of MSET
				keys := make([]string, 0, len(testData))
				for key := range testData {
					keys = append(keys, key)
				}

				msetResult := client.MSet(ctx, testData)
				Expect(msetResult.Err()).NotTo(HaveOccurred())
				Expect(msetResult.Val()).To(Equal("OK"))

				// Execute MGET
				mgetResult := client.MGet(ctx, keys...)
				Expect(mgetResult.Err()).NotTo(HaveOccurred())

				// Verify all values match
				values := mgetResult.Val()
				Expect(len(values)).To(Equal(len(keys)))
				for i, key := range keys {
					Expect(values[i]).To(Equal(testData[key]))
				}
			})

			It("should test DisableRoutingPolicies option disables routing policies", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// The fixture explicitly enables metadata policies, so MGET works across slots.
				testData := map[string]string{
					"disable_routing_key_1111": "value1",
					"disable_routing_key_2222": "value2",
					"disable_routing_key_3333": "value3",
				}

				// Set keys
				for key, value := range testData {
					Expect(client.Set(ctx, key, value, 0).Err()).NotTo(HaveOccurred())
				}

				// Verify keys are on different shards
				slotMap := make(map[int]bool)
				for key := range testData {
					slot := hashtag.Slot(key)
					slotMap[slot] = true
				}
				Expect(len(slotMap)).To(BeNumerically(">", 1))

				keys := make([]string, 0, len(testData))
				for key := range testData {
					keys = append(keys, key)
				}

				// With routing policies enabled, MGET should work
				mgetResult := client.MGet(ctx, keys...)
				Expect(mgetResult.Err()).NotTo(HaveOccurred())
				Expect(len(mgetResult.Val())).To(Equal(len(keys)))

				// Test 2: With routing policies disabled, MGET should fail with CROSSSLOT error
				opt := redisClusterOptions()
				opt.DisableRoutingPolicies = true
				clientWithoutPolicies := cluster.newClusterClient(ctx, opt)
				defer clientWithoutPolicies.Close()

				// Try MGET with routing policies disabled - should fail with CROSSSLOT error
				mgetResultDisabled := clientWithoutPolicies.MGet(ctx, keys...)
				Expect(mgetResultDisabled.Err()).To(HaveOccurred())
				Expect(mgetResultDisabled.Err().Error()).To(ContainSubstring("CROSSSLOT"))
			})

			It("should test large MGET with many keys across all shards", func() {
				SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

				// Create many keys to ensure coverage across all shards
				numKeys := 100
				keys := make([]string, numKeys)
				values := make(map[string]string)

				for i := 0; i < numKeys; i++ {
					key := fmt.Sprintf("large_mget_key_%d", i)
					value := fmt.Sprintf("value_%d", i)
					keys[i] = key
					values[key] = value
					Expect(client.Set(ctx, key, value, 0).Err()).NotTo(HaveOccurred())
				}

				// Verify keys are distributed across multiple shards
				slotMap := make(map[int]bool)
				for _, key := range keys {
					slot := hashtag.Slot(key)
					slotMap[slot] = true
				}
				Expect(len(slotMap)).To(BeNumerically(">", 1))

				// Execute MGET
				result := client.MGet(ctx, keys...)
				Expect(result.Err()).NotTo(HaveOccurred())

				// Verify all values are correct
				resultValues := result.Val()
				Expect(len(resultValues)).To(Equal(numKeys))
				for i, key := range keys {
					Expect(resultValues[i]).To(Equal(values[key]))
				}
			})
		})

		It("should route keyless commands to arbitrary shards using round robin", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			var numMasters int
			var numMastersMu sync.Mutex
			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				numMastersMu.Lock()
				numMasters++
				numMastersMu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(numMasters).To(BeNumerically(">", 1))

			err = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.ConfigResetStat(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			// Helper function to get ECHO command counts from all nodes
			getEchoCounts := func() map[string]int {
				echoCounts := make(map[string]int)
				var echoCountsMu sync.Mutex
				err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
					info := master.Info(ctx, "server")
					Expect(info.Err()).NotTo(HaveOccurred())

					serverInfo := info.Val()
					portStart := strings.Index(serverInfo, "tcp_port:")
					portLine := serverInfo[portStart:]
					portEnd := strings.Index(portLine, "\r\n")
					if portEnd == -1 {
						portEnd = len(portLine)
					}
					port := strings.TrimPrefix(portLine[:portEnd], "tcp_port:")

					commandStats := master.Info(ctx, "commandstats")
					count := 0
					if commandStats.Err() == nil {
						stats := commandStats.Val()
						cmdStatKey := "cmdstat_echo:"
						if strings.Contains(stats, cmdStatKey) {
							statStart := strings.Index(stats, cmdStatKey)
							statLine := stats[statStart:]
							statEnd := strings.Index(statLine, "\r\n")
							if statEnd == -1 {
								statEnd = len(statLine)
							}
							statLine = statLine[:statEnd]

							callsStart := strings.Index(statLine, "calls=")
							if callsStart != -1 {
								callsStr := statLine[callsStart+6:]
								callsEnd := strings.Index(callsStr, ",")
								if callsEnd == -1 {
									callsEnd = strings.Index(callsStr, "\r")
									if callsEnd == -1 {
										callsEnd = len(callsStr)
									}
								}
								if callsCount, err := strconv.Atoi(callsStr[:callsEnd]); err == nil {
									count = callsCount
								}
							}
						}
					}

					echoCountsMu.Lock()
					echoCounts[port] = count
					echoCountsMu.Unlock()
					return nil
				})
				Expect(err).NotTo(HaveOccurred())
				return echoCounts
			}

			// Single ECHO command should go to exactly one shard
			result := client.Echo(ctx, "single_test")
			Expect(result.Err()).NotTo(HaveOccurred())
			Expect(result.Val()).To(Equal("single_test"))

			time.Sleep(200 * time.Millisecond)

			// Verify single command went to exactly one shard
			echoCounts := getEchoCounts()
			shardsWithEcho := 0
			for _, count := range echoCounts {
				if count > 0 {
					shardsWithEcho++
					Expect(count).To(Equal(1))
				}
			}
			Expect(shardsWithEcho).To(Equal(1))

			// Reset stats before multi-ECHO test to get clean counts
			err = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.ConfigResetStat(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			// Test Multiple ECHO commands should distribute across all shards using round robin
			// With round robin, sending numMasters commands should hit each shard exactly once
			numCommands := numMasters

			for i := 0; i < numCommands; i++ {
				result := client.Echo(ctx, fmt.Sprintf("multi_test_%d", i))
				Expect(result.Err()).NotTo(HaveOccurred())
				Expect(result.Val()).To(Equal(fmt.Sprintf("multi_test_%d", i)))
			}

			time.Sleep(200 * time.Millisecond)

			echoCounts = getEchoCounts()
			totalEchos := 0
			shardsWithEchos := 0
			for _, count := range echoCounts {
				if count > 0 {
					shardsWithEchos++
				}
				totalEchos += count
			}

			// All shards should have received commands with round robin distribution
			Expect(shardsWithEchos).To(Equal(numMasters))

			// Total should match what we sent
			Expect(totalEchos).To(Equal(numCommands))
		})
	})

	_ = Describe("ClusterClient ParseURL", func() {
		cases := []struct {
			test string
			url  string
			o    *redis.ClusterOptions // expected value
			err  error
		}{
			{
				test: "ParseRedisURL",
				url:  "redis://localhost:123",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}},
			}, {
				test: "ParseRedissURL",
				url:  "rediss://localhost:123",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, TLSConfig: &tls.Config{ServerName: "localhost", MinVersion: tls.VersionTLS12}},
			}, {
				test: "MissingRedisPort",
				url:  "redis://localhost",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:6379"}},
			}, {
				test: "MissingRedissPort",
				url:  "rediss://localhost",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:6379"}, TLSConfig: &tls.Config{ServerName: "localhost", MinVersion: tls.VersionTLS12}},
			}, {
				test: "MultipleRedisURLs",
				url:  "redis://localhost:123?addr=localhost:1234&addr=localhost:12345",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234", "localhost:12345"}},
			}, {
				test: "MultipleRedissURLs",
				url:  "rediss://localhost:123?addr=localhost:1234&addr=localhost:12345",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234", "localhost:12345"}, TLSConfig: &tls.Config{ServerName: "localhost", MinVersion: tls.VersionTLS12}},
			}, {
				test: "OnlyPassword",
				url:  "redis://:bar@localhost:123",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Password: "bar"},
			}, {
				test: "OnlyUser",
				url:  "redis://foo@localhost:123",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Username: "foo"},
			}, {
				test: "RedisUsernamePassword",
				url:  "redis://foo:bar@localhost:123",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Username: "foo", Password: "bar"},
			}, {
				test: "RedissUsernamePassword",
				url:  "rediss://foo:bar@localhost:123?addr=localhost:1234",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234"}, Username: "foo", Password: "bar", TLSConfig: &tls.Config{ServerName: "localhost", MinVersion: tls.VersionTLS12}},
			}, {
				test: "RedissSkipVerify",
				url:  "rediss://localhost:123?skip_verify=true",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, TLSConfig: &tls.Config{ServerName: "localhost", MinVersion: tls.VersionTLS12, InsecureSkipVerify: true}},
			}, {
				test: "RedissVerifyNodeHost",
				url:  "rediss://localhost:123?addr=127.0.0.1:1234&tls_verify_node_host=true",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123", "127.0.0.1:1234"}, TLSConfig: &tls.Config{MinVersion: tls.VersionTLS12}},
			}, {
				test: "RedissVerifyNodeHostFalse",
				url:  "rediss://localhost:123?tls_verify_node_host=false",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, TLSConfig: &tls.Config{ServerName: "localhost", MinVersion: tls.VersionTLS12}},
			}, {
				test: "QueryParameters",
				url:  "redis://localhost:123?read_timeout=2&pool_fifo=true&addr=localhost:1234",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234"}, ReadTimeout: 2 * time.Second, PoolFIFO: true},
			}, {
				test: "DisabledTimeout",
				url:  "redis://localhost:123?conn_max_idle_time=0",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: -1},
			}, {
				test: "DisabledTimeoutNeg",
				url:  "redis://localhost:123?conn_max_idle_time=-1",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: -1},
			}, {
				test: "UseDefault",
				url:  "redis://localhost:123?conn_max_idle_time=",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: 0},
			}, {
				test: "Protocol",
				url:  "redis://localhost:123?protocol=2",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Protocol: 2},
			}, {
				test: "ClientName",
				url:  "redis://localhost:123?client_name=cluster_hi",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ClientName: "cluster_hi"},
			}, {
				test: "UseDefaultMissing=",
				url:  "redis://localhost:123?conn_max_idle_time",
				o:    &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: 0},
			}, {
				test: "InvalidQueryAddr",
				url:  "rediss://foo:bar@localhost:123?addr=rediss://foo:barr@localhost:1234",
				err:  errors.New(`redis: unable to parse addr param: rediss://foo:barr@localhost:1234`),
			}, {
				test: "InvalidInt",
				url:  "redis://localhost?pool_size=five",
				err:  errors.New(`redis: invalid pool_size number: strconv.Atoi: parsing "five": invalid syntax`),
			}, {
				test: "InvalidBool",
				url:  "redis://localhost?pool_fifo=yes",
				err:  errors.New(`redis: invalid pool_fifo boolean: expected true/false/1/0 or an empty string, got "yes"`),
			}, {
				test: "InvalidSkipVerify",
				url:  "rediss://localhost?skip_verify=yes",
				err:  errors.New(`redis: invalid skip_verify boolean: expected true/false/1/0 or an empty string, got "yes"`),
			}, {
				test: "SkipVerifyWithoutTLS",
				url:  "redis://localhost?skip_verify=true",
				err:  errors.New("redis: unexpected option: skip_verify"),
			}, {
				test: "InvalidVerifyNodeHost",
				url:  "rediss://localhost?tls_verify_node_host=yes",
				err:  errors.New(`redis: invalid tls_verify_node_host boolean: expected true/false/1/0 or an empty string, got "yes"`),
			}, {
				test: "VerifyNodeHostWithoutTLS",
				url:  "redis://localhost?tls_verify_node_host=true",
				err:  errors.New("redis: unexpected option: tls_verify_node_host"),
			}, {
				test: "UnknownParam",
				url:  "redis://localhost?abc=123",
				err:  errors.New("redis: unexpected option: abc"),
			}, {
				test: "InvalidScheme",
				url:  "https://google.com",
				err:  errors.New("redis: invalid URL scheme: https"),
			},
		}

		It("should match ParseClusterURL", func() {
			for i := range cases {
				tc := cases[i]
				actual, err := redis.ParseClusterURL(tc.url)
				if tc.err != nil {
					Expect(err).Should(MatchError(tc.err))
				} else {
					Expect(err).NotTo(HaveOccurred())
				}

				if err == nil {
					Expect(tc.o).NotTo(BeNil())

					Expect(tc.o.Addrs).To(Equal(actual.Addrs))
					Expect(tc.o.TLSConfig).To(Equal(actual.TLSConfig))
					Expect(tc.o.Username).To(Equal(actual.Username))
					Expect(tc.o.Password).To(Equal(actual.Password))
					Expect(tc.o.MaxRetries).To(Equal(actual.MaxRetries))
					Expect(tc.o.MinRetryBackoff).To(Equal(actual.MinRetryBackoff))
					Expect(tc.o.MaxRetryBackoff).To(Equal(actual.MaxRetryBackoff))
					Expect(tc.o.DialTimeout).To(Equal(actual.DialTimeout))
					Expect(tc.o.ReadTimeout).To(Equal(actual.ReadTimeout))
					Expect(tc.o.WriteTimeout).To(Equal(actual.WriteTimeout))
					Expect(tc.o.PoolFIFO).To(Equal(actual.PoolFIFO))
					Expect(tc.o.PoolSize).To(Equal(actual.PoolSize))
					Expect(tc.o.MinIdleConns).To(Equal(actual.MinIdleConns))
					Expect(tc.o.ConnMaxLifetime).To(Equal(actual.ConnMaxLifetime))
					Expect(tc.o.ConnMaxIdleTime).To(Equal(actual.ConnMaxIdleTime))
					Expect(tc.o.PoolTimeout).To(Equal(actual.PoolTimeout))
				}
			}
		})

		It("should distribute keyless commands randomly across shards using random shard picker", func() {
			SkipBeforeRedisVersion("7.9", "The tips are included from Redis 8")

			// Create a cluster client with random shard picker
			opt := redisClusterOptions()
			opt.ShardPicker = &routing.RandomPicker{}
			randomClient := cluster.newClusterClient(ctx, opt)
			defer randomClient.Close()

			Eventually(func() error {
				return randomClient.Ping(ctx).Err()
			}, 30*time.Second).ShouldNot(HaveOccurred())

			var numMasters int
			var numMastersMu sync.Mutex
			err := randomClient.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				numMastersMu.Lock()
				numMasters++
				numMastersMu.Unlock()
				return nil
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(numMasters).To(BeNumerically(">", 1))

			// Reset command statistics on all masters
			err = randomClient.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
				return master.ConfigResetStat(ctx).Err()
			})
			Expect(err).NotTo(HaveOccurred())

			// Helper function to get ECHO command counts from all nodes
			getEchoCounts := func() map[string]int {
				echoCounts := make(map[string]int)
				var echoCountsMu sync.Mutex
				err := randomClient.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
					addr := master.Options().Addr
					port := addr[strings.LastIndex(addr, ":")+1:]

					info, err := master.Info(ctx, "commandstats").Result()
					if err != nil {
						return err
					}

					count := 0
					if strings.Contains(info, "cmdstat_echo:") {
						lines := strings.Split(info, "\n")
						for _, line := range lines {
							if strings.HasPrefix(line, "cmdstat_echo:") {
								parts := strings.Split(line, ",")
								if len(parts) > 0 {
									callsPart := strings.Split(parts[0], "=")
									if len(callsPart) > 1 {
										if parsedCount, parseErr := strconv.Atoi(callsPart[1]); parseErr == nil {
											count = parsedCount
										}
									}
								}
								break
							}
						}
					}

					echoCountsMu.Lock()
					echoCounts[port] = count
					echoCountsMu.Unlock()
					return nil
				})
				Expect(err).NotTo(HaveOccurred())
				return echoCounts
			}

			// Execute multiple ECHO commands and measure distribution
			numCommands := 100
			for i := 0; i < numCommands; i++ {
				result := randomClient.Echo(ctx, fmt.Sprintf("random_test_%d", i))
				Expect(result.Err()).NotTo(HaveOccurred())
			}

			echoCounts := getEchoCounts()

			totalEchos := 0
			shardsWithEchos := 0

			for _, count := range echoCounts {
				if count > 0 {
					shardsWithEchos++
				}
				totalEchos += count
			}

			Expect(totalEchos).To(Equal(numCommands))
			Expect(shardsWithEchos).To(BeNumerically(">=", 2))
		})
	})
})

var _ = Describe("ClusterClient FailingTimeoutSeconds", func() {
	var client *redis.ClusterClient

	AfterEach(func() {
		if client != nil {
			_ = client.Close()
		}
	})

	It("should use default failing timeout of 15 seconds", func() {
		opt := redisClusterOptions()
		client = cluster.newClusterClient(ctx, opt)

		// Default should be 15 seconds
		Expect(opt.FailingTimeoutSeconds).To(Equal(15))
	})

	It("should use custom failing timeout", func() {
		opt := redisClusterOptions()
		opt.FailingTimeoutSeconds = 30
		client = cluster.newClusterClient(ctx, opt)

		// Should use custom value
		Expect(opt.FailingTimeoutSeconds).To(Equal(30))
	})

	It("should parse failing_timeout_seconds from URL", func() {
		url := "redis://localhost:16600?failing_timeout_seconds=25"
		opt, err := redis.ParseClusterURL(url)
		Expect(err).NotTo(HaveOccurred())
		Expect(opt.FailingTimeoutSeconds).To(Equal(25))
	})

	It("should handle node failing timeout correctly", func() {
		// Use a larger window than the original 2s. Failing() compares
		// truncated unix seconds (failing < timeout), so a 1s sleep against
		// a 2s timeout has < 1s of margin and flips false on slow CI runners
		// where time.Sleep overshoots.
		opt := redisClusterOptions()
		opt.FailingTimeoutSeconds = 5
		client = cluster.newClusterClient(ctx, opt)

		// Get a node and mark it as failing
		nodes, err := client.Nodes(ctx, "A")
		Expect(err).NotTo(HaveOccurred())
		Expect(len(nodes)).To(BeNumerically(">", 0))

		node := nodes[0]

		// Initially not failing
		Expect(node.Failing()).To(BeFalse())

		// Mark as failing
		node.MarkAsFailing()
		Expect(node.Failing()).To(BeTrue())

		// Should still be failing well before the timeout expires.
		time.Sleep(2 * time.Second)
		Expect(node.Failing()).To(BeTrue())

		// Should not be failing after timeout expires (total 7s > 5s timeout,
		// with > 1s margin on each side).
		time.Sleep(5 * time.Second)
		Expect(node.Failing()).To(BeFalse())
	})

	It("should handle zero timeout by using default", func() {
		opt := redisClusterOptions()
		opt.FailingTimeoutSeconds = 0 // Should use default
		client = cluster.newClusterClient(ctx, opt)

		// After initialization, should be set to default
		Expect(opt.FailingTimeoutSeconds).To(Equal(15))
	})
})