remove thread debugging code && reject commands with policy that cannot be used in pipeline

Author: ofekshenawa

internal/routing/policy.go

@@ -129,3 +129,7 @@ type CommandPolicy struct {
 	// e.g nondeterministic_output, nondeterministic_output_order.
 	Tips map[string]string
 }
+
+func (p *CommandPolicy) CanBeUsedInPipeline() bool {
+	return p.Request != ReqAllNodes && p.Request != ReqAllShards && p.Request != ReqMultiShard
+}

osscluster.go

@@ -933,9 +933,6 @@ type ClusterClient struct {
 // NewClusterClient returns a Redis Cluster client as described in
 // http://redis.io/topics/cluster-spec.
 func NewClusterClient(opt *ClusterOptions) *ClusterClient {
-	if opt == nil {
-		panic("redis: NewClusterClient nil options")
-	}
 	opt.init()
 
 	c := &ClusterClient{
@@ -1293,23 +1290,11 @@ func (c *ClusterClient) Pipelined(ctx context.Context, fn func(Pipeliner) error)
 }
 
 func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error {
-	// Separate commands into those that can be batched vs those that need individual routing
-	batchableCmds := make([]Cmder, 0)
-	individualCmds := make([]Cmder, 0)
+	cmdsMap := newCmdsMap()
 
-	for _, cmd := range cmds {
-		policy := c.getCommandPolicy(ctx, cmd)
-
-		// Commands that need special routing should be handled individually
-		if policy != nil && (policy.Request == routing.ReqAllNodes ||
-			policy.Request == routing.ReqAllShards ||
-			policy.Request == routing.ReqMultiShard ||
-			policy.Request == routing.ReqSpecial) {
-			individualCmds = append(individualCmds, cmd)
-		} else {
-			// Single-node commands can be batched
-			batchableCmds = append(batchableCmds, cmd)
-		}
+	if err := c.mapCmdsByNode(ctx, cmdsMap, cmds); err != nil {
+		setCmdsErr(cmds, err)
+		return err
 	}
 
 	for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
@@ -1320,68 +1305,74 @@ func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error
 			}
 		}
 
-		var allSucceeded = true
-		var failedBatchableCmds []Cmder
-		var failedIndividualCmds []Cmder
+		failedCmds := newCmdsMap()
+		var wg sync.WaitGroup
 
-		// Handle individual commands using existing router
-		for _, cmd := range individualCmds {
-			if err := c.routeAndRun(ctx, cmd, nil); err != nil {
-				allSucceeded = false
-				failedIndividualCmds = append(failedIndividualCmds, cmd)
-			}
+		for node, cmds := range cmdsMap.m {
+			wg.Add(1)
+			go func(node *clusterNode, cmds []Cmder) {
+				defer wg.Done()
+				c.processPipelineNode(ctx, node, cmds, failedCmds)
+			}(node, cmds)
 		}
 
-		// Handle batchable commands using original pipeline logic
-		if len(batchableCmds) > 0 {
-			cmdsMap := newCmdsMap()
+		wg.Wait()
+		if len(failedCmds.m) == 0 {
+			break
+		}
+		cmdsMap = failedCmds
+	}
 
-			if err := c.mapCmdsByNode(ctx, cmdsMap, batchableCmds); err != nil {
-				setCmdsErr(batchableCmds, err)
-				allSucceeded = false
-				failedBatchableCmds = append(failedBatchableCmds, batchableCmds...)
-			} else {
-				batchFailedCmds := newCmdsMap()
-				var wg sync.WaitGroup
-
-				for node, nodeCmds := range cmdsMap.m {
-					wg.Add(1)
-					go func(node *clusterNode, nodeCmds []Cmder) {
-						defer wg.Done()
-						c.processPipelineNode(ctx, node, nodeCmds, batchFailedCmds)
-					}(node, nodeCmds)
-				}
+	return cmdsFirstErr(cmds)
+}
 
-				wg.Wait()
+func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmds []Cmder) error {
+	state, err := c.state.Get(ctx)
+	if err != nil {
+		return err
+	}
 
-				if len(batchFailedCmds.m) > 0 {
-					allSucceeded = false
-					for _, nodeCmds := range batchFailedCmds.m {
-						failedBatchableCmds = append(failedBatchableCmds, nodeCmds...)
-					}
-				}
+	if c.opt.ReadOnly && c.cmdsAreReadOnly(ctx, cmds) {
+		for _, cmd := range cmds {
+			policy := c.getCommandPolicy(ctx, cmd)
+			if policy != nil && !policy.CanBeUsedInPipeline() {
+				return fmt.Errorf("redis: cannot pipeline command %q with request policy ReqAllNodes/ReqAllShards/ReqMultiShard", cmd.Name())
+			}
+			slot := c.cmdSlot(ctx, cmd)
+			node, err := c.slotReadOnlyNode(state, slot)
+			if err != nil {
+				return err
 			}
+			cmdsMap.Add(node, cmd)
 		}
+		return nil
+	}
 
-		// If all commands succeeded, we're done
-		if allSucceeded {
-			break
+	for _, cmd := range cmds {
+		policy := c.getCommandPolicy(ctx, cmd)
+		if policy != nil && !policy.CanBeUsedInPipeline() {
+			return fmt.Errorf("redis: cannot pipeline command %q with request policy ReqAllNodes/ReqAllShards/ReqMultiShard", cmd.Name())
 		}
-
-		// If this was the last attempt, return the error
-		if attempt == c.opt.MaxRedirects {
-			break
+		slot := c.cmdSlot(ctx, cmd)
+		node, err := state.slotMasterNode(slot)
+		if err != nil {
+			return err
 		}
-
-		// Update command lists for retry - no reclassification needed
-		batchableCmds = failedBatchableCmds
-		individualCmds = failedIndividualCmds
+		cmdsMap.Add(node, cmd)
 	}
+	return nil
+}
 
-	return cmdsFirstErr(cmds)
+func (c *ClusterClient) cmdsAreReadOnly(ctx context.Context, cmds []Cmder) bool {
+	for _, cmd := range cmds {
+		cmdInfo := c.cmdInfo(ctx, cmd.Name())
+		if cmdInfo == nil || !cmdInfo.ReadOnly {
+			return false
+		}
+	}
+	return true
 }
 
-// processPipelineNode handles batched pipeline commands for a single node
 func (c *ClusterClient) processPipelineNode(
 	ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
 ) {
@@ -1391,8 +1382,7 @@ func (c *ClusterClient) processPipelineNode(
 			if !isContextError(err) {
 				node.MarkAsFailing()
 			}
-			// Commands are already mapped to this node, just add them as failed
-			failedCmds.Add(node, cmds...)
+			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 			setCmdsErr(cmds, err)
 			return err
 		}
@@ -1417,8 +1407,7 @@ func (c *ClusterClient) processPipelineNodeConn(
 			node.MarkAsFailing()
 		}
 		if shouldRetry(err, true) {
-			// Commands are already mapped to this node, just add them as failed
-			failedCmds.Add(node, cmds...)
+			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 		}
 		setCmdsErr(cmds, err)
 		return err
@@ -1454,63 +1443,21 @@ func (c *ClusterClient) pipelineReadCmds(
 
 		if !isRedisError(err) {
 			if shouldRetry(err, true) {
-				// Commands are already mapped to this node, just add them as failed
-				failedCmds.Add(node, cmds[i:]...)
+				_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 			}
 			setCmdsErr(cmds[i+1:], err)
 			return err
 		}
 	}
 
 	if err := cmds[0].Err(); err != nil && shouldRetry(err, true) {
-		// Commands are already mapped to this node, just add them as failed
-		failedCmds.Add(node, cmds...)
+		_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 		return err
 	}
 
 	return nil
 }
 
-// Legacy functions needed for transaction pipeline processing
-func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmds []Cmder) error {
-	state, err := c.state.Get(ctx)
-	if err != nil {
-		return err
-	}
-
-	if c.opt.ReadOnly && c.cmdsAreReadOnly(ctx, cmds) {
-		for _, cmd := range cmds {
-			slot := c.cmdSlot(ctx, cmd)
-			node, err := c.slotReadOnlyNode(state, slot)
-			if err != nil {
-				return err
-			}
-			cmdsMap.Add(node, cmd)
-		}
-		return nil
-	}
-
-	for _, cmd := range cmds {
-		slot := c.cmdSlot(ctx, cmd)
-		node, err := state.slotMasterNode(slot)
-		if err != nil {
-			return err
-		}
-		cmdsMap.Add(node, cmd)
-	}
-	return nil
-}
-
-func (c *ClusterClient) cmdsAreReadOnly(ctx context.Context, cmds []Cmder) bool {
-	for _, cmd := range cmds {
-		cmdInfo := c.cmdInfo(ctx, cmd.Name())
-		if cmdInfo == nil || !cmdInfo.ReadOnly {
-			return false
-		}
-	}
-	return true
-}
-
 func (c *ClusterClient) checkMovedErr(
 	ctx context.Context, cmd Cmder, err error, failedCmds *cmdsMap,
 ) bool {
@@ -1538,35 +1485,6 @@ func (c *ClusterClient) checkMovedErr(
 	panic("not reached")
 }
 
-func (c *ClusterClient) cmdsMoved(
-	ctx context.Context, cmds []Cmder,
-	moved, ask bool,
-	addr string,
-	failedCmds *cmdsMap,
-) error {
-	node, err := c.nodes.GetOrCreate(addr)
-	if err != nil {
-		return err
-	}
-
-	if moved {
-		c.state.LazyReload()
-		for _, cmd := range cmds {
-			failedCmds.Add(node, cmd)
-		}
-		return nil
-	}
-
-	if ask {
-		for _, cmd := range cmds {
-			failedCmds.Add(node, NewCmd(ctx, "asking"), cmd)
-		}
-		return nil
-	}
-
-	return nil
-}
-
 // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
 func (c *ClusterClient) TxPipeline() Pipeliner {
 	pipe := Pipeline{
@@ -1734,6 +1652,35 @@ func (c *ClusterClient) txPipelineReadQueued(
 	return nil
 }
 
+func (c *ClusterClient) cmdsMoved(
+	ctx context.Context, cmds []Cmder,
+	moved, ask bool,
+	addr string,
+	failedCmds *cmdsMap,
+) error {
+	node, err := c.nodes.GetOrCreate(addr)
+	if err != nil {
+		return err
+	}
+
+	if moved {
+		c.state.LazyReload()
+		for _, cmd := range cmds {
+			failedCmds.Add(node, cmd)
+		}
+		return nil
+	}
+
+	if ask {
+		for _, cmd := range cmds {
+			failedCmds.Add(node, NewCmd(ctx, "asking"), cmd)
+		}
+		return nil
+	}
+
+	return nil
+}
+
 func (c *ClusterClient) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
 	if len(keys) == 0 {
 		return fmt.Errorf("redis: Watch requires at least one key")

osscluster_test.go

@@ -6,9 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"net"
-	"os"
-	"runtime"
-	"runtime/pprof"
 	"slices"
 	"strconv"
 	"strings"
@@ -17,19 +14,10 @@ import (
 
 	. "github.com/bsm/ginkgo/v2"
 	. "github.com/bsm/gomega"
-	"github.com/fortytw2/leaktest"
 	"github.com/redis/go-redis/v9"
 	"github.com/redis/go-redis/v9/internal/hashtag"
 )
 
-// leakCleanup holds the per-spec leak check function
-var leakCleanup func()
-
-// sanitizeFilename converts spaces and slashes into underscores
-func sanitizeFilename(s string) string {
-	return strings.NewReplacer(" ", "_", "/", "_").Replace(s)
-}
-
 type clusterScenario struct {
 	ports   []string
 	nodeIDs []string
@@ -266,7 +254,7 @@ func slotEqual(s1, s2 redis.ClusterSlot) bool {
 
 // ------------------------------------------------------------------------------
 
-var _ = Describe("ClusterClient", func() {
+var _ = FDescribe("ClusterClient", func() {
 	var failover bool
 	var opt *redis.ClusterOptions
 	var client *redis.ClusterClient
@@ -1391,8 +1379,6 @@ var _ = Describe("ClusterClient", func() {
 
 	Describe("ClusterClient with ClusterSlots with multiple nodes per slot", func() {
 		BeforeEach(func() {
-			leakCleanup = leaktest.Check(GinkgoT())
-			GinkgoWriter.Printf("[DEBUG] goroutines at start: %d\n", runtime.NumGoroutine())
 			failover = true
 
 			opt = redisClusterOptions()
@@ -1442,21 +1428,6 @@ var _ = Describe("ClusterClient", func() {
 		})
 
 		AfterEach(func() {
-			leakCleanup()
-
-			// on failure, write out a full goroutine dump
-			if CurrentSpecReport().Failed() {
-				fname := fmt.Sprintf("goroutines-%s.txt", sanitizeFilename(CurrentSpecReport().LeafNodeText))
-				if f, err := os.Create(fname); err == nil {
-					pprof.Lookup("goroutine").WriteTo(f, 2)
-					f.Close()
-					GinkgoWriter.Printf("[DEBUG] wrote goroutine dump to %s\n", fname)
-				} else {
-					GinkgoWriter.Printf("[DEBUG] failed to write goroutine dump: %v\n", err)
-				}
-			}
-
-			GinkgoWriter.Printf("[DEBUG] goroutines at end:   %d\n", runtime.NumGoroutine())
 			failover = false
 
 			err := client.Close()