cluster.go

package valkey

import (
	"context"
	"errors"
	"io"
	"math/rand"
	"net"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/valkey-io/valkey-go/internal/cmds"
	"github.com/valkey-io/valkey-go/internal/util"
)

// ErrNoSlot indicates that there is no valkey node owns the key slot.
var ErrNoSlot = errors.New("the slot has no valkey node")
var ErrReplicaOnlyConflict = errors.New("ReplicaOnly conflicts with SendToReplicas option")

type retry struct {
	cIndexes []int
	commands []Completed
	aIndexes []int
	cAskings []Completed
}

func (r *retry) Capacity() int {
	return cap(r.commands)
}

func (r *retry) ResetLen(n int) {
	r.cIndexes = r.cIndexes[:n]
	r.commands = r.commands[:n]
	r.aIndexes = r.aIndexes[:0]
	r.cAskings = r.cAskings[:0]
}

var retryp = util.NewPool(func(capacity int) *retry {
	return &retry{
		cIndexes: make([]int, 0, capacity),
		commands: make([]Completed, 0, capacity),
	}
})

type retrycache struct {
	cIndexes []int
	commands []CacheableTTL
	aIndexes []int
	cAskings []CacheableTTL
}

func (r *retrycache) Capacity() int {
	return cap(r.commands)
}

func (r *retrycache) ResetLen(n int) {
	r.cIndexes = r.cIndexes[:n]
	r.commands = r.commands[:n]
	r.aIndexes = r.aIndexes[:0]
	r.cAskings = r.cAskings[:0]
}

var retrycachep = util.NewPool(func(capacity int) *retrycache {
	return &retrycache{
		cIndexes: make([]int, 0, capacity),
		commands: make([]CacheableTTL, 0, capacity),
	}
})

type clusterClient struct {
	pslots [16384]conn
	rslots []conn
	opt    *ClientOption
	rOpt   *ClientOption
	conns  map[string]connrole
	connFn connFn
	sc     call
	mu     sync.RWMutex
	stop   uint32
	cmd    Builder
	retry  bool
	aws    bool
}

// NOTE: connrole and conn must be initialized at the same time
type connrole struct {
	conn    conn
	replica bool
}

func newClusterClient(opt *ClientOption, connFn connFn) (*clusterClient, error) {
	client := &clusterClient{
		cmd:    cmds.NewBuilder(cmds.InitSlot),
		connFn: connFn,
		opt:    opt,
		conns:  make(map[string]connrole),
		retry:  !opt.DisableRetry,
		aws:    len(opt.InitAddress) == 1 && strings.Contains(opt.InitAddress[0], "amazonaws.com"),
	}

	if opt.ReplicaOnly && opt.SendToReplicas != nil {
		return nil, ErrReplicaOnlyConflict
	}

	if opt.SendToReplicas != nil {
		rOpt := *opt
		rOpt.ReplicaOnly = true
		client.rOpt = &rOpt
	}

	client.connFn = func(dst string, opt *ClientOption) conn {
		cc := connFn(dst, opt)
		cc.SetOnCloseHook(func(err error) {
			client.lazyRefresh()
		})
		return cc
	}

	if err := client.init(); err != nil {
		return nil, err
	}

	if err := client.refresh(context.Background()); err != nil {
		return client, err
	}

	return client, nil
}

func (c *clusterClient) init() error {
	if len(c.opt.InitAddress) == 0 {
		return ErrNoAddr
	}
	results := make(chan error, len(c.opt.InitAddress))
	for _, addr := range c.opt.InitAddress {
		cc := c.connFn(addr, c.opt)
		go func(addr string, cc conn) {
			if err := cc.Dial(); err == nil {
				c.mu.Lock()
				if _, ok := c.conns[addr]; ok {
					go cc.Close() // abort the new connection instead of closing the old one which may already been used
				} else {
					c.conns[addr] = connrole{
						conn: cc,
					}
				}
				c.mu.Unlock()
				results <- nil
			} else {
				results <- err
			}
		}(addr, cc)
	}
	es := make([]error, cap(results))
	for i := 0; i < cap(results); i++ {
		if err := <-results; err == nil {
			return nil
		} else {
			es[i] = err
		}
	}
	return es[0]
}

func (c *clusterClient) refresh(ctx context.Context) (err error) {
	return c.sc.Do(ctx, c._refresh)
}

func (c *clusterClient) lazyRefresh() {
	c.sc.LazyDo(time.Second, c._refresh)
}

type clusterslots struct {
	addr  string
	reply ValkeyResult
	ver   int
}

func (s clusterslots) parse(tls bool) map[string]group {
	if s.ver < 7 {
		return parseSlots(s.reply.val, s.addr)
	}
	return parseShards(s.reply.val, s.addr, tls)
}

func getClusterSlots(c conn) clusterslots {
	v := c.Version()
	if v < 7 {
		return clusterslots{reply: c.Do(context.Background(), cmds.SlotCmd), addr: c.Addr(), ver: v}
	}
	return clusterslots{reply: c.Do(context.Background(), cmds.ShardsCmd), addr: c.Addr(), ver: v}
}

func (c *clusterClient) _refresh() (err error) {
	c.mu.RLock()
	results := make(chan clusterslots, len(c.conns))
	pending := make([]conn, 0, len(c.conns))
	if c.aws {
		pending = append(pending, c.conns[c.opt.InitAddress[0]].conn)
	} else {
		for _, cc := range c.conns {
			pending = append(pending, cc.conn)
		}
	}
	c.mu.RUnlock()

	var result clusterslots
	for i := 0; i < cap(results); i++ {
		if i&3 == 0 { // batch CLUSTER SLOTS/CLUSTER SHARDS for every 4 connections
			for j := i; j < i+4 && j < len(pending); j++ {
				go func(c conn) {
					results <- getClusterSlots(c)
				}(pending[j])
			}
		}
		result = <-results
		err = result.reply.Error()
		if len(result.reply.val.values) != 0 {
			break
		}
	}
	if err != nil {
		return err
	}
	pending = nil

	groups := result.parse(c.opt.TLSConfig != nil)
	conns := make(map[string]connrole, len(groups))
	for master, g := range groups {
		conns[master] = connrole{conn: c.connFn(master, c.opt), replica: false}
		for _, addr := range g.nodes[1:] {
			if c.rOpt != nil {
				conns[addr] = connrole{conn: c.connFn(addr, c.rOpt), replica: true}
			} else {
				conns[addr] = connrole{conn: c.connFn(addr, c.opt), replica: true}
			}
		}
	}
	// make sure InitAddress always be present
	for _, addr := range c.opt.InitAddress {
		if _, ok := conns[addr]; !ok {
			conns[addr] = connrole{
				conn: c.connFn(addr, c.opt),
			}
		}
	}

	var removes []conn

	c.mu.RLock()
	for addr, cc := range c.conns {
		fresh, ok := conns[addr]
		if ok && (cc.replica == fresh.replica || c.rOpt == nil) {
			conns[addr] = connrole{
				conn:    cc.conn,
				replica: fresh.replica,
			}
		} else {
			removes = append(removes, cc.conn)
		}
	}
	c.mu.RUnlock()

	pslots := [16384]conn{}
	var rslots []conn
	for master, g := range groups {
		switch {
		case c.opt.ReplicaOnly && len(g.nodes) > 1:
			nodesCount := len(g.nodes)
			for _, slot := range g.slots {
				for i := slot[0]; i <= slot[1]; i++ {
					pslots[i] = conns[g.nodes[1+rand.Intn(nodesCount-1)]].conn
				}
			}
		case c.rOpt != nil: // implies c.opt.SendToReplicas != nil
			if len(rslots) == 0 { // lazy init
				rslots = make([]conn, 16384)
			}
			if len(g.nodes) > 1 {
				for _, slot := range g.slots {
					for i := slot[0]; i <= slot[1]; i++ {
						pslots[i] = conns[master].conn
						rslots[i] = conns[g.nodes[1+rand.Intn(len(g.nodes)-1)]].conn
					}
				}
			} else {
				for _, slot := range g.slots {
					for i := slot[0]; i <= slot[1]; i++ {
						pslots[i] = conns[master].conn
						rslots[i] = conns[master].conn
					}
				}
			}
		default:
			for _, slot := range g.slots {
				for i := slot[0]; i <= slot[1]; i++ {
					pslots[i] = conns[master].conn
				}
			}
		}
	}

	c.mu.Lock()
	c.pslots = pslots
	c.rslots = rslots
	c.conns = conns
	c.mu.Unlock()

	if len(removes) > 0 {
		go func(removes []conn) {
			time.Sleep(time.Second * 5)
			for _, cc := range removes {
				cc.Close()
			}
		}(removes)
	}

	return nil
}

func (c *clusterClient) single() (conn conn) {
	return c._pick(cmds.InitSlot, false)
}

func (c *clusterClient) nodes() []string {
	c.mu.RLock()
	nodes := make([]string, 0, len(c.conns))
	for addr := range c.conns {
		nodes = append(nodes, addr)
	}
	c.mu.RUnlock()
	return nodes
}

type group struct {
	nodes []string
	slots [][2]int64
}

func parseEndpoint(fallback, endpoint string, port int64) string {
	switch endpoint {
	case "":
		endpoint, _, _ = net.SplitHostPort(fallback)
	case "?":
		return ""
	}
	return net.JoinHostPort(endpoint, strconv.FormatInt(port, 10))
}

// parseSlots - map valkey slots for each valkey nodes/addresses
// defaultAddr is needed in case the node does not know its own IP
func parseSlots(slots ValkeyMessage, defaultAddr string) map[string]group {
	groups := make(map[string]group, len(slots.values))
	for _, v := range slots.values {
		master := parseEndpoint(defaultAddr, v.values[2].values[0].string, v.values[2].values[1].integer)
		if master == "" {
			continue
		}
		g, ok := groups[master]
		if !ok {
			g.slots = make([][2]int64, 0)
			g.nodes = make([]string, 0, len(v.values)-2)
			for i := 2; i < len(v.values); i++ {
				if dst := parseEndpoint(defaultAddr, v.values[i].values[0].string, v.values[i].values[1].integer); dst != "" {
					g.nodes = append(g.nodes, dst)
				}
			}
		}
		g.slots = append(g.slots, [2]int64{v.values[0].integer, v.values[1].integer})
		groups[master] = g
	}
	return groups
}

// parseShards - map valkey shards for each valkey nodes/addresses
// defaultAddr is needed in case the node does not know its own IP
func parseShards(shards ValkeyMessage, defaultAddr string, tls bool) map[string]group {
	groups := make(map[string]group, len(shards.values))
	for _, v := range shards.values {
		m := -1
		shard, _ := v.AsMap()
		slots := shard["slots"].values
		nodes := shard["nodes"].values
		g := group{
			nodes: make([]string, 0, len(nodes)),
			slots: make([][2]int64, len(slots)/2),
		}
		for i := range g.slots {
			g.slots[i][0], _ = slots[i*2].AsInt64()
			g.slots[i][1], _ = slots[i*2+1].AsInt64()
		}
		for _, n := range nodes {
			dict, _ := n.AsMap()
			port := dict["port"].integer
			if tls && dict["tls-port"].integer > 0 {
				port = dict["tls-port"].integer
			}
			if dst := parseEndpoint(defaultAddr, dict["endpoint"].string, port); dst != "" {
				if dict["role"].string == "master" {
					m = len(g.nodes)
				}
				g.nodes = append(g.nodes, dst)
			}
		}
		if m >= 0 {
			g.nodes[0], g.nodes[m] = g.nodes[m], g.nodes[0]
			groups[g.nodes[0]] = g
		}
	}
	return groups
}

func (c *clusterClient) _pick(slot uint16, toReplica bool) (p conn) {
	c.mu.RLock()
	if slot == cmds.InitSlot {
		for _, cc := range c.conns {
			if cc.replica {
				continue
			}
			p = cc.conn
			break
		}
	} else if toReplica && c.rslots != nil {
		p = c.rslots[slot]
	} else {
		p = c.pslots[slot]
	}
	c.mu.RUnlock()
	return p
}

func (c *clusterClient) pick(ctx context.Context, slot uint16, toReplica bool) (p conn, err error) {
	if p = c._pick(slot, toReplica); p == nil {
		if err := c.refresh(ctx); err != nil {
			return nil, err
		}
		if p = c._pick(slot, toReplica); p == nil {
			return nil, ErrNoSlot
		}
	}
	return p, nil
}

func (c *clusterClient) redirectOrNew(addr string, prev conn) (p conn) {
	c.mu.RLock()
	cc := c.conns[addr]
	c.mu.RUnlock()
	if cc.conn != nil && prev != cc.conn {
		return cc.conn
	}
	c.mu.Lock()

	if cc = c.conns[addr]; cc.conn == nil {
		p = c.connFn(addr, c.opt)
		c.conns[addr] = connrole{conn: p, replica: false}
	} else if prev == cc.conn {
		// try reconnection if the MOVED redirects to the same host,
		// because the same hostname may actually be resolved into another destination
		// depending on the fail-over implementation. ex: AWS MemoryDB's resize process.
		go func(prev conn) {
			time.Sleep(time.Second * 5)
			prev.Close()
		}(prev)
		p = c.connFn(addr, c.opt)
		c.conns[addr] = connrole{conn: p, replica: cc.replica}
	}
	c.mu.Unlock()
	return p
}

func (c *clusterClient) B() Builder {
	return c.cmd
}

func (c *clusterClient) Do(ctx context.Context, cmd Completed) (resp ValkeyResult) {
	if resp = c.do(ctx, cmd); resp.NonValkeyError() == nil { // not recycle cmds if error, since cmds may be used later in pipe. consider recycle them by pipe
		cmds.PutCompleted(cmd)
	}
	return resp
}

func (c *clusterClient) do(ctx context.Context, cmd Completed) (resp ValkeyResult) {
retry:
	cc, err := c.pick(ctx, cmd.Slot(), c.toReplica(cmd))
	if err != nil {
		return newErrResult(err)
	}
	resp = cc.Do(ctx, cmd)
process:
	switch addr, mode := c.shouldRefreshRetry(resp.Error(), ctx); mode {
	case RedirectMove:
		resp = c.redirectOrNew(addr, cc).Do(ctx, cmd)
		goto process
	case RedirectAsk:
		results := c.redirectOrNew(addr, cc).DoMulti(ctx, cmds.AskingCmd, cmd)
		resp = results.s[1]
		resultsp.Put(results)
		goto process
	case RedirectRetry:
		if c.retry && cmd.IsReadOnly() {
			runtime.Gosched()
			goto retry
		}
	}
	return resp
}

func (c *clusterClient) toReplica(cmd Completed) bool {
	if c.opt.SendToReplicas != nil {
		return c.opt.SendToReplicas(cmd)
	}
	return false
}

func (c *clusterClient) _pickMulti(multi []Completed) (retries *connretry, last uint16, toReplica bool) {
	last = cmds.InitSlot
	init := false

	for _, cmd := range multi {
		if cmd.Slot() == cmds.InitSlot {
			init = true
			break
		}
	}

	c.mu.RLock()
	defer c.mu.RUnlock()

	count := conncountp.Get(len(c.conns), len(c.conns))

	if !init && c.rslots != nil && c.opt.SendToReplicas != nil {
		for _, cmd := range multi {
			var p conn
			if c.opt.SendToReplicas(cmd) {
				p = c.rslots[cmd.Slot()]
			} else {
				p = c.pslots[cmd.Slot()]
			}
			if p == nil {
				return nil, 0, false
			}
			count.m[p]++
		}

		retries = connretryp.Get(len(count.m), len(count.m))
		for cc, n := range count.m {
			retries.m[cc] = retryp.Get(0, n)
		}
		conncountp.Put(count)

		for i, cmd := range multi {
			last = cmd.Slot()

			var cc conn
			if c.opt.SendToReplicas(cmd) {
				toReplica = true
				cc = c.rslots[cmd.Slot()]
			} else {
				cc = c.pslots[cmd.Slot()]
			}

			re := retries.m[cc]
			re.commands = append(re.commands, cmd)
			re.cIndexes = append(re.cIndexes, i)
		}
		return retries, last, toReplica
	}

	for _, cmd := range multi {
		if cmd.Slot() == cmds.InitSlot {
			continue
		}
		p := c.pslots[cmd.Slot()]
		if p == nil {
			return nil, 0, false
		}
		count.m[p]++
	}

	retries = connretryp.Get(len(count.m), len(count.m))
	for cc, n := range count.m {
		retries.m[cc] = retryp.Get(0, n)
	}
	conncountp.Put(count)

	for i, cmd := range multi {
		if cmd.Slot() != cmds.InitSlot {
			if last == cmds.InitSlot {
				last = cmd.Slot()
			} else if init && last != cmd.Slot() {
				panic(panicMixCxSlot)
			}
			cc := c.pslots[cmd.Slot()]
			re := retries.m[cc]
			re.commands = append(re.commands, cmd)
			re.cIndexes = append(re.cIndexes, i)
		}
	}
	return retries, last, false
}

func (c *clusterClient) pickMulti(ctx context.Context, multi []Completed) (*connretry, uint16, bool, error) {
	conns, slot, toReplica := c._pickMulti(multi)
	if conns == nil {
		if err := c.refresh(ctx); err != nil {
			return nil, 0, false, err
		}
		if conns, slot, toReplica = c._pickMulti(multi); conns == nil {
			return nil, 0, false, ErrNoSlot
		}
	}
	return conns, slot, toReplica, nil
}

func (c *clusterClient) doresultfn(ctx context.Context, results *valkeyresults, retries *connretry, mu *sync.Mutex, cc conn, cIndexes []int, commands []Completed, resps []ValkeyResult) {
	for i, resp := range resps {
		ii := cIndexes[i]
		cm := commands[i]
		results.s[ii] = resp
		addr, mode := c.shouldRefreshRetry(resp.Error(), ctx)
		if mode != RedirectNone {
			nc := cc
			if mode == RedirectRetry {
				if !c.retry || !cm.IsReadOnly() {
					continue
				}
			} else {
				nc = c.redirectOrNew(addr, cc)
			}
			mu.Lock()
			nr := retries.m[nc]
			if nr == nil {
				nr = retryp.Get(0, len(commands))
				retries.m[nc] = nr
			}
			if mode == RedirectAsk {
				nr.aIndexes = append(nr.aIndexes, ii)
				nr.cAskings = append(nr.cAskings, cm)
			} else {
				nr.cIndexes = append(nr.cIndexes, ii)
				nr.commands = append(nr.commands, cm)
			}
			mu.Unlock()
		}
	}
}

func (c *clusterClient) doretry(ctx context.Context, cc conn, results *valkeyresults, retries *connretry, re *retry, mu *sync.Mutex, wg *sync.WaitGroup) {
	if len(re.commands) != 0 {
		resps := cc.DoMulti(ctx, re.commands...)
		c.doresultfn(ctx, results, retries, mu, cc, re.cIndexes, re.commands, resps.s)
		resultsp.Put(resps)
	}
	if len(re.cAskings) != 0 {
		resps := askingMulti(cc, ctx, re.cAskings)
		c.doresultfn(ctx, results, retries, mu, cc, re.aIndexes, re.cAskings, resps.s)
		resultsp.Put(resps)
	}
	if ctx.Err() == nil {
		retryp.Put(re)
	}
	wg.Done()
}

func (c *clusterClient) DoMulti(ctx context.Context, multi ...Completed) []ValkeyResult {
	if len(multi) == 0 {
		return nil
	}

	retries, slot, toReplica, err := c.pickMulti(ctx, multi)
	if err != nil {
		return fillErrs(len(multi), err)
	}
	defer connretryp.Put(retries)

	if len(retries.m) <= 1 {
		for _, re := range retries.m {
			retryp.Put(re)
		}
		return c.doMulti(ctx, slot, multi, toReplica)
	}

	var wg sync.WaitGroup
	var mu sync.Mutex

	results := resultsp.Get(len(multi), len(multi))

retry:
	wg.Add(len(retries.m))
	mu.Lock()
	for cc, re := range retries.m {
		delete(retries.m, cc)
		go c.doretry(ctx, cc, results, retries, re, &mu, &wg)
	}
	mu.Unlock()
	wg.Wait()

	if len(retries.m) != 0 {
		goto retry
	}

	for i, cmd := range multi {
		if results.s[i].NonValkeyError() == nil {
			cmds.PutCompleted(cmd)
		}
	}
	return results.s
}

func fillErrs(n int, err error) (results []ValkeyResult) {
	results = resultsp.Get(n, n).s
	for i := range results {
		results[i] = newErrResult(err)
	}
	return results
}

func (c *clusterClient) doMulti(ctx context.Context, slot uint16, multi []Completed, toReplica bool) []ValkeyResult {
retry:
	cc, err := c.pick(ctx, slot, toReplica)
	if err != nil {
		return fillErrs(len(multi), err)
	}
	resps := cc.DoMulti(ctx, multi...)
process:
	for _, resp := range resps.s {
		switch addr, mode := c.shouldRefreshRetry(resp.Error(), ctx); mode {
		case RedirectMove:
			if c.retry && allReadOnly(multi) {
				resultsp.Put(resps)
				resps = c.redirectOrNew(addr, cc).DoMulti(ctx, multi...)
				goto process
			}
		case RedirectAsk:
			if c.retry && allReadOnly(multi) {
				resultsp.Put(resps)
				resps = askingMulti(c.redirectOrNew(addr, cc), ctx, multi)
				goto process
			}
		case RedirectRetry:
			if c.retry && allReadOnly(multi) {
				resultsp.Put(resps)
				runtime.Gosched()
				goto retry
			}
		}
	}
	return resps.s
}

func (c *clusterClient) doCache(ctx context.Context, cmd Cacheable, ttl time.Duration) (resp ValkeyResult) {
retry:
	cc, err := c.pick(ctx, cmd.Slot(), c.toReplica(Completed(cmd)))
	if err != nil {
		return newErrResult(err)
	}
	resp = cc.DoCache(ctx, cmd, ttl)
process:
	switch addr, mode := c.shouldRefreshRetry(resp.Error(), ctx); mode {
	case RedirectMove:
		resp = c.redirectOrNew(addr, cc).DoCache(ctx, cmd, ttl)
		goto process
	case RedirectAsk:
		results := askingMultiCache(c.redirectOrNew(addr, cc), ctx, []CacheableTTL{CT(cmd, ttl)})
		resp = results.s[0]
		resultsp.Put(results)
		goto process
	case RedirectRetry:
		if c.retry {
			runtime.Gosched()
			goto retry
		}
	}
	return resp
}

func (c *clusterClient) DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) (resp ValkeyResult) {
	resp = c.doCache(ctx, cmd, ttl)
	if err := resp.NonValkeyError(); err == nil || err == ErrDoCacheAborted {
		cmds.PutCacheable(cmd)
	}
	return resp
}

func askingMulti(cc conn, ctx context.Context, multi []Completed) *valkeyresults {
	commands := make([]Completed, 0, len(multi)*2)
	for _, cmd := range multi {
		commands = append(commands, cmds.AskingCmd, cmd)
	}
	results := resultsp.Get(0, len(multi))
	resps := cc.DoMulti(ctx, commands...)
	for i := 1; i < len(resps.s); i += 2 {
		results.s = append(results.s, resps.s[i])
	}
	resultsp.Put(resps)
	return results
}

func askingMultiCache(cc conn, ctx context.Context, multi []CacheableTTL) *valkeyresults {
	commands := make([]Completed, 0, len(multi)*6)
	for _, cmd := range multi {
		ck, _ := cmds.CacheKey(cmd.Cmd)
		commands = append(commands, cmds.OptInCmd, cmds.MultiCmd, cmds.AskingCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(cmd.Cmd), cmds.ExecCmd)
	}
	results := resultsp.Get(0, len(multi))
	resps := cc.DoMulti(ctx, commands...)
	for i := 5; i < len(resps.s); i += 6 {
		if arr, err := resps.s[i].ToArray(); err != nil {
			results.s = append(results.s, newErrResult(err))
		} else {
			results.s = append(results.s, newResult(arr[len(arr)-1], nil))
		}
	}
	resultsp.Put(resps)
	return results
}

func (c *clusterClient) _pickMultiCache(multi []CacheableTTL) *connretrycache {
	c.mu.RLock()
	defer c.mu.RUnlock()

	count := conncountp.Get(len(c.conns), len(c.conns))
	if c.opt.SendToReplicas == nil || c.rslots == nil {
		for _, cmd := range multi {
			p := c.pslots[cmd.Cmd.Slot()]
			if p == nil {
				return nil
			}
			count.m[p]++
		}

		retries := connretrycachep.Get(len(count.m), len(count.m))
		for cc, n := range count.m {
			retries.m[cc] = retrycachep.Get(0, n)
		}
		conncountp.Put(count)

		for i, cmd := range multi {
			cc := c.pslots[cmd.Cmd.Slot()]
			re := retries.m[cc]
			re.commands = append(re.commands, cmd)
			re.cIndexes = append(re.cIndexes, i)
		}

		return retries
	} else {
		for _, cmd := range multi {
			var p conn
			if c.opt.SendToReplicas(Completed(cmd.Cmd)) {
				p = c.rslots[cmd.Cmd.Slot()]
			} else {
				p = c.pslots[cmd.Cmd.Slot()]
			}
			if p == nil {
				return nil
			}
			count.m[p]++
		}

		retries := connretrycachep.Get(len(count.m), len(count.m))
		for cc, n := range count.m {
			retries.m[cc] = retrycachep.Get(0, n)
		}
		conncountp.Put(count)

		for i, cmd := range multi {
			var cc conn
			if c.opt.SendToReplicas(Completed(cmd.Cmd)) {
				cc = c.rslots[cmd.Cmd.Slot()]
			} else {
				cc = c.pslots[cmd.Cmd.Slot()]
			}
			re := retries.m[cc]
			re.commands = append(re.commands, cmd)
			re.cIndexes = append(re.cIndexes, i)
		}

		return retries
	}
}

func (c *clusterClient) pickMultiCache(ctx context.Context, multi []CacheableTTL) (*connretrycache, error) {
	conns := c._pickMultiCache(multi)
	if conns == nil {
		if err := c.refresh(ctx); err != nil {
			return nil, err
		}
		if conns = c._pickMultiCache(multi); conns == nil {
			return nil, ErrNoSlot
		}
	}
	return conns, nil
}

func (c *clusterClient) resultcachefn(ctx context.Context, results *valkeyresults, retries *connretrycache, mu *sync.Mutex, cc conn, cIndexes []int, commands []CacheableTTL, resps []ValkeyResult) {
	for i, resp := range resps {
		ii := cIndexes[i]
		cm := commands[i]
		results.s[ii] = resp
		addr, mode := c.shouldRefreshRetry(resp.Error(), ctx)
		if mode != RedirectNone {
			nc := cc
			if mode == RedirectRetry {
				if !c.retry {
					continue
				}
			} else {
				nc = c.redirectOrNew(addr, cc)
			}
			mu.Lock()
			nr := retries.m[nc]
			if nr == nil {
				nr = retrycachep.Get(0, len(commands))
				retries.m[nc] = nr
			}
			if mode == RedirectAsk {
				nr.aIndexes = append(nr.aIndexes, ii)
				nr.cAskings = append(nr.cAskings, cm)
			} else {
				nr.cIndexes = append(nr.cIndexes, ii)
				nr.commands = append(nr.commands, cm)
			}
			mu.Unlock()
		}
	}
}

func (c *clusterClient) doretrycache(ctx context.Context, cc conn, results *valkeyresults, retries *connretrycache, re *retrycache, mu *sync.Mutex, wg *sync.WaitGroup) {
	if len(re.commands) != 0 {
		resps := cc.DoMultiCache(ctx, re.commands...)
		c.resultcachefn(ctx, results, retries, mu, cc, re.cIndexes, re.commands, resps.s)
		resultsp.Put(resps)
	}
	if len(re.cAskings) != 0 {
		resps := askingMultiCache(cc, ctx, re.cAskings)
		c.resultcachefn(ctx, results, retries, mu, cc, re.aIndexes, re.cAskings, resps.s)
		resultsp.Put(resps)
	}
	if ctx.Err() == nil {
		retrycachep.Put(re)
	}
	wg.Done()
}

func (c *clusterClient) DoMultiCache(ctx context.Context, multi ...CacheableTTL) []ValkeyResult {
	if len(multi) == 0 {
		return nil
	}

	retries, err := c.pickMultiCache(ctx, multi)
	if err != nil {
		return fillErrs(len(multi), err)
	}
	defer connretrycachep.Put(retries)

	var wg sync.WaitGroup
	var mu sync.Mutex

	results := resultsp.Get(len(multi), len(multi))

retry:
	wg.Add(len(retries.m))
	mu.Lock()
	for cc, re := range retries.m {
		delete(retries.m, cc)
		go c.doretrycache(ctx, cc, results, retries, re, &mu, &wg)
	}
	mu.Unlock()
	wg.Wait()

	if len(retries.m) != 0 {
		goto retry
	}

	for i, cmd := range multi {
		if err := results.s[i].NonValkeyError(); err == nil || err == ErrDoCacheAborted {
			cmds.PutCacheable(cmd.Cmd)
		}
	}
	return results.s
}

func (c *clusterClient) Receive(ctx context.Context, subscribe Completed, fn func(msg PubSubMessage)) (err error) {
retry:
	cc, err := c.pick(ctx, subscribe.Slot(), c.toReplica(subscribe))
	if err != nil {
		goto ret
	}
	err = cc.Receive(ctx, subscribe, fn)
	if _, mode := c.shouldRefreshRetry(err, ctx); c.retry && mode != RedirectNone {
		runtime.Gosched()
		goto retry
	}
ret:
	if err == nil {
		cmds.PutCompleted(subscribe)
	}
	return err
}

func (c *clusterClient) DoStream(ctx context.Context, cmd Completed) ValkeyResultStream {
	cc, err := c.pick(ctx, cmd.Slot(), c.toReplica(cmd))
	if err != nil {
		return ValkeyResultStream{e: err}
	}
	ret := cc.DoStream(ctx, cmd)
	cmds.PutCompleted(cmd)
	return ret
}

func (c *clusterClient) DoMultiStream(ctx context.Context, multi ...Completed) MultiValkeyResultStream {
	if len(multi) == 0 {
		return ValkeyResultStream{e: io.EOF}
	}
	slot := multi[0].Slot()
	repl := c.toReplica(multi[0])
	for i := 1; i < len(multi); i++ {
		if s := multi[i].Slot(); s != cmds.InitSlot {
			if slot == cmds.InitSlot {
				slot = s
			} else if slot != s {
				panic("DoMultiStream across multiple slots is not supported")
			}
		}
		repl = repl && c.toReplica(multi[i])
	}
	cc, err := c.pick(ctx, slot, repl)
	if err != nil {
		return ValkeyResultStream{e: err}
	}
	ret := cc.DoMultiStream(ctx, multi...)
	for _, cmd := range multi {
		cmds.PutCompleted(cmd)
	}
	return ret
}

func (c *clusterClient) Dedicated(fn func(DedicatedClient) error) (err error) {
	dcc := &dedicatedClusterClient{cmd: c.cmd, client: c, slot: cmds.NoSlot, retry: c.retry}
	err = fn(dcc)
	dcc.release()
	return err
}

func (c *clusterClient) Dedicate() (DedicatedClient, func()) {
	dcc := &dedicatedClusterClient{cmd: c.cmd, client: c, slot: cmds.NoSlot, retry: c.retry}
	return dcc, dcc.release
}

func (c *clusterClient) Nodes() map[string]Client {
	c.mu.RLock()
	nodes := make(map[string]Client, len(c.conns))
	for addr, cc := range c.conns {
		nodes[addr] = newSingleClientWithConn(cc.conn, c.cmd, c.retry)
	}
	c.mu.RUnlock()
	return nodes
}

func (c *clusterClient) Close() {
	atomic.StoreUint32(&c.stop, 1)
	c.mu.RLock()
	for _, cc := range c.conns {
		go cc.conn.Close()
	}
	c.mu.RUnlock()
}

func (c *clusterClient) shouldRefreshRetry(err error, ctx context.Context) (addr string, mode RedirectMode) {
	if err != nil && atomic.LoadUint32(&c.stop) == 0 {
		if err, ok := err.(*ValkeyError); ok {
			if addr, ok = err.IsMoved(); ok {
				mode = RedirectMove
			} else if addr, ok = err.IsAsk(); ok {
				mode = RedirectAsk
			} else if err.IsClusterDown() || err.IsTryAgain() {
				mode = RedirectRetry
			}
		} else if ctx.Err() == nil {
			mode = RedirectRetry
		}
		if mode != RedirectNone {
			c.lazyRefresh()
		}
	}
	return
}

type dedicatedClusterClient struct {
	client *clusterClient
	conn   conn
	wire   wire
	pshks  *pshks

	mu    sync.Mutex
	cmd   Builder
	slot  uint16
	mark  bool
	retry bool
}

func (c *dedicatedClusterClient) acquire(ctx context.Context, slot uint16) (wire wire, err error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.mark {
		panic(dedicatedClientUsedAfterReleased)
	}
	if c.slot == cmds.NoSlot {
		c.slot = slot
	} else if c.slot != slot && slot != cmds.InitSlot {
		panic(panicMsgCxSlot)
	}
	if c.wire != nil {
		return c.wire, nil
	}
	if c.conn, err = c.client.pick(ctx, c.slot, false); err != nil {
		if p := c.pshks; p != nil {
			c.pshks = nil
			p.close <- err
			close(p.close)
		}
		return nil, err
	}
	c.wire = c.conn.Acquire()
	if p := c.pshks; p != nil {
		c.pshks = nil
		ch := c.wire.SetPubSubHooks(p.hooks)
		go func(ch <-chan error) {
			for e := range ch {
				p.close <- e
			}
			close(p.close)
		}(ch)
	}
	return c.wire, nil
}

func (c *dedicatedClusterClient) release() {
	c.mu.Lock()
	if !c.mark {
		if p := c.pshks; p != nil {
			c.pshks = nil
			close(p.close)
		}
		if c.wire != nil {
			c.conn.Store(c.wire)
		}
	}
	c.mark = true
	c.mu.Unlock()
}

func (c *dedicatedClusterClient) B() Builder {
	return c.cmd
}

func (c *dedicatedClusterClient) Do(ctx context.Context, cmd Completed) (resp ValkeyResult) {
retry:
	if w, err := c.acquire(ctx, cmd.Slot()); err != nil {
		resp = newErrResult(err)
	} else {
		resp = w.Do(ctx, cmd)
		switch _, mode := c.client.shouldRefreshRetry(resp.Error(), ctx); mode {
		case RedirectRetry:
			if c.retry && cmd.IsReadOnly() && w.Error() == nil {
				runtime.Gosched()
				goto retry
			}
		}
	}
	if resp.NonValkeyError() == nil {
		cmds.PutCompleted(cmd)
	}
	return resp
}

func (c *dedicatedClusterClient) DoMulti(ctx context.Context, multi ...Completed) (resp []ValkeyResult) {
	if len(multi) == 0 {
		return nil
	}
	slot := chooseSlot(multi)
	if slot == cmds.NoSlot {
		panic(panicMsgCxSlot)
	}
	retryable := c.retry
	if retryable {
		retryable = allReadOnly(multi)
	}
retry:
	if w, err := c.acquire(ctx, slot); err == nil {
		resp = w.DoMulti(ctx, multi...).s
		for _, r := range resp {
			_, mode := c.client.shouldRefreshRetry(r.Error(), ctx)
			if mode == RedirectRetry && retryable && w.Error() == nil {
				runtime.Gosched()
				goto retry
			}
			if mode != RedirectNone {
				break
			}
		}
	} else {
		resp = resultsp.Get(len(multi), len(multi)).s
		for i := range resp {
			resp[i] = newErrResult(err)
		}
	}
	for i, cmd := range multi {
		if resp[i].NonValkeyError() == nil {
			cmds.PutCompleted(cmd)
		}
	}
	return resp
}

func (c *dedicatedClusterClient) Receive(ctx context.Context, subscribe Completed, fn func(msg PubSubMessage)) (err error) {
	var w wire
retry:
	if w, err = c.acquire(ctx, subscribe.Slot()); err == nil {
		err = w.Receive(ctx, subscribe, fn)
		if _, mode := c.client.shouldRefreshRetry(err, ctx); c.retry && mode == RedirectRetry && w.Error() == nil {
			runtime.Gosched()
			goto retry
		}
	}
	if err == nil {
		cmds.PutCompleted(subscribe)
	}
	return err
}

func (c *dedicatedClusterClient) SetPubSubHooks(hooks PubSubHooks) <-chan error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.mark {
		panic(dedicatedClientUsedAfterReleased)
	}
	if p := c.pshks; p != nil {
		c.pshks = nil
		close(p.close)
	}
	if c.wire != nil {
		return c.wire.SetPubSubHooks(hooks)
	}
	if hooks.isZero() {
		return nil
	}
	ch := make(chan error, 1)
	c.pshks = &pshks{hooks: hooks, close: ch}
	return ch
}

func (c *dedicatedClusterClient) Close() {
	c.mu.Lock()
	if p := c.pshks; p != nil {
		c.pshks = nil
		p.close <- ErrClosing
		close(p.close)
	}
	if c.wire != nil {
		c.wire.Close()
	}
	c.mu.Unlock()
	c.release()
}

type RedirectMode int

const (
	RedirectNone RedirectMode = iota
	RedirectMove
	RedirectAsk
	RedirectRetry

	panicMsgCxSlot = "cross slot command in Dedicated is prohibited"
	panicMixCxSlot = "Mixing no-slot and cross slot commands in DoMulti is prohibited"
)

type conncount struct {
	m map[conn]int
	n int
}

func (r *conncount) Capacity() int {
	return r.n
}

func (r *conncount) ResetLen(n int) {
	for k := range r.m {
		delete(r.m, k)
	}
}

var conncountp = util.NewPool(func(capacity int) *conncount {
	return &conncount{m: make(map[conn]int, capacity), n: capacity}
})

type connretry struct {
	m map[conn]*retry
	n int
}

func (r *connretry) Capacity() int {
	return r.n
}

func (r *connretry) ResetLen(n int) {
	for k := range r.m {
		delete(r.m, k)
	}
}

var connretryp = util.NewPool(func(capacity int) *connretry {
	return &connretry{m: make(map[conn]*retry, capacity), n: capacity}
})

type connretrycache struct {
	m map[conn]*retrycache
	n int
}

func (r *connretrycache) Capacity() int {
	return r.n
}

func (r *connretrycache) ResetLen(n int) {
	for k := range r.m {
		delete(r.m, k)
	}
}

var connretrycachep = util.NewPool(func(capacity int) *connretrycache {
	return &connretrycache{m: make(map[conn]*retrycache, capacity), n: capacity}
})