cleanup, added logic to refresh the cache

Author: htemelski-oss

command.go

@@ -4343,8 +4343,9 @@ func (cmd *CommandsInfoCmd) Clone() Cmder {
 type cmdsInfoCache struct {
 	fn func(ctx context.Context) (map[string]*CommandInfo, error)
 
-	once internal.Once
-	cmds map[string]*CommandInfo
+	once        internal.Once
+	refreshLock sync.Mutex
+	cmds        map[string]*CommandInfo
 }
 
 func newCmdsInfoCache(fn func(ctx context.Context) (map[string]*CommandInfo, error)) *cmdsInfoCache {
@@ -4354,6 +4355,9 @@ func newCmdsInfoCache(fn func(ctx context.Context) (map[string]*CommandInfo, err
 }
 
 func (c *cmdsInfoCache) Get(ctx context.Context) (map[string]*CommandInfo, error) {
+	c.refreshLock.Lock()
+	defer c.refreshLock.Unlock()
+
 	err := c.once.Do(func() error {
 		cmds, err := c.fn(ctx)
 		if err != nil {
@@ -4374,6 +4378,14 @@ func (c *cmdsInfoCache) Get(ctx context.Context) (map[string]*CommandInfo, error
 	return c.cmds, err
 }
 
+// TODO: Call it on client reconnect
+func (c *cmdsInfoCache) Refresh() {
+	c.refreshLock.Lock()
+	defer c.refreshLock.Unlock()
+
+	c.once = internal.Once{}
+}
+
 // ------------------------------------------------------------------------------
 const requestPolicy = "request_policy"
 const responsePolicy = "response_policy"

command_policy_manager.go

@@ -922,11 +922,10 @@ func (c *clusterStateHolder) ReloadOrGet(ctx context.Context) (*clusterState, er
 // or more underlying connections. It's safe for concurrent use by
 // multiple goroutines.
 type ClusterClient struct {
-	opt              *ClusterOptions
-	nodes            *clusterNodes
-	state            *clusterStateHolder
-	cmdsInfoCache    *cmdsInfoCache
-	cmdPolicyManager *commandPolicyManager
+	opt           *ClusterOptions
+	nodes         *clusterNodes
+	state         *clusterStateHolder
+	cmdsInfoCache *cmdsInfoCache
 	cmdable
 	hooksMixin
 }
@@ -942,9 +941,9 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
 	}
 
 	c.state = newClusterStateHolder(c.loadState)
+	// TODO: execute on handshake, should be called again on reconnect
 	c.cmdsInfoCache = newCmdsInfoCache(c.cmdsInfo)
 	c.cmdable = c.Process
-	c.cmdPolicyManager = newCommandPolicyManager(nil)
 	c.initHooks(hooks{
 		dial:       nil,
 		process:    c.process,
@@ -1336,7 +1335,7 @@ func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmd
 
 	if c.opt.ReadOnly && c.cmdsAreReadOnly(ctx, cmds) {
 		for _, cmd := range cmds {
-			policy := c.cmdPolicyManager.getCmdPolicy(cmd)
+			policy := c.getCommandPolicy(ctx, cmd)
 			if policy != nil && !policy.CanBeUsedInPipeline() {
 				return fmt.Errorf(
 					"redis: cannot pipeline command %q with request policy ReqAllNodes/ReqAllShards/ReqMultiShard; Note: This behavior is subject to change in the future", cmd.Name(),
@@ -1353,7 +1352,7 @@ func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmd
 	}
 
 	for _, cmd := range cmds {
-		policy := c.cmdPolicyManager.getCmdPolicy(cmd)
+		policy := c.getCommandPolicy(ctx, cmd)
 		if policy != nil && !policy.CanBeUsedInPipeline() {
 			return fmt.Errorf(
 				"redis: cannot pipeline command %q with request policy ReqAllNodes/ReqAllShards/ReqMultiShard; Note: This behavior is subject to change in the future", cmd.Name(),
@@ -1863,6 +1862,7 @@ func (c *ClusterClient) cmdsInfo(ctx context.Context) (map[string]*CommandInfo,
 	return nil, firstErr
 }
 
+// cmdInfo will fetch and cache the command policies after the first execution
 func (c *ClusterClient) cmdInfo(ctx context.Context, name string) *CommandInfo {
 	// Use a separate context that won't be canceled to ensure command info lookup
 	// doesn't fail due to original context cancellation

osscluster.go

@@ -21,7 +21,7 @@ type slotResult struct {
 
 // routeAndRun routes a command to the appropriate cluster nodes and executes it
 func (c *ClusterClient) routeAndRun(ctx context.Context, cmd Cmder, node *clusterNode) error {
-	policy := c.cmdPolicyManager.getCmdPolicy(cmd)
+	policy := c.getCommandPolicy(ctx, cmd)
 
 	switch {
 	case policy != nil && policy.Request == routing.ReqAllNodes:
@@ -37,6 +37,14 @@ func (c *ClusterClient) routeAndRun(ctx context.Context, cmd Cmder, node *cluste
 	}
 }
 
+// getCommandPolicy retrieves the routing policy for a command
+func (c *ClusterClient) getCommandPolicy(ctx context.Context, cmd Cmder) *routing.CommandPolicy {
+	if cmdInfo := c.cmdInfo(ctx, cmd.Name()); cmdInfo != nil && cmdInfo.Tips != nil {
+		return cmdInfo.Tips
+	}
+	return nil
+}
+
 // executeDefault handles standard command routing based on keys
 func (c *ClusterClient) executeDefault(ctx context.Context, cmd Cmder, node *clusterNode) error {
 	if c.hasKeys(cmd) {

osscluster_router.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"reflect"
 	"slices"
 	"strconv"
 	"strings"
@@ -1567,6 +1568,7 @@ var _ = Describe("ClusterClient timeout", func() {
 				return nil
 			})
 			Expect(err).To(HaveOccurred())
+			fmt.Println("qko greshki male", reflect.TypeOf(err).String(), reflect.TypeOf(err).Kind().String())
 			Expect(err.(net.Error).Timeout()).To(BeTrue())
 		})