From c6d385a63e606e55d30b18205e9cbded1b286706 Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov Date: Mon, 16 Jun 2025 18:27:57 +0300 Subject: [PATCH 1/5] fix(txpipeline): should return error on multi/exec on multiple slots --- osscluster.go | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/osscluster.go b/osscluster.go index a68f7eab1..06a588567 100644 --- a/osscluster.go +++ b/osscluster.go @@ -1504,6 +1504,16 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err } cmdsMap := c.mapCmdsBySlot(cmds) + // TxPipeline does not support cross slot transaction. + if len(cmdsMap) > 1 { + err := fmt.Errorf("redis: CROSSSLOT Keys in request don't hash to the same slot") + setCmdsErr(cmds, err) + return err + } + if len(cmdsMap) == 0 { + return nil + } + for slot, cmds := range cmdsMap { node, err := state.slotMasterNode(slot) if err != nil { From 9b71da1801eee13829308380d407dd8373cf298d Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov Date: Mon, 16 Jun 2025 18:46:01 +0300 Subject: [PATCH 2/5] fix(txpipeline): test normal tx pipeline behaviour --- osscluster_test.go | 34 ++++++++++++++++++---------------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/osscluster_test.go b/osscluster_test.go index 993411837..7b0a8b292 100644 --- a/osscluster_test.go +++ b/osscluster_test.go @@ -462,8 +462,7 @@ var _ = Describe("ClusterClient", func() { Describe("pipelining", func() { var pipe *redis.Pipeline - assertPipeline := func() { - keys := []string{"A", "B", "C", "D", "E", "F", "G"} + assertPipeline := func(keys []string) { It("follows redirects", func() { if !failover { @@ -482,13 +481,12 @@ var _ = Describe("ClusterClient", func() { Expect(err).NotTo(HaveOccurred()) Expect(cmds).To(HaveLen(14)) - _ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error { - defer GinkgoRecover() - Eventually(func() int64 { - return node.DBSize(ctx).Val() - }, 30*time.Second).ShouldNot(BeZero()) - return nil - }) + // 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 { @@ -517,14 +515,14 @@ var _ = Describe("ClusterClient", func() { }) It("works with missing keys", func() { - pipe.Set(ctx, "A", "A_value", 0) - pipe.Set(ctx, "C", "C_value", 0) + 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") - b := pipe.Get(ctx, "B") - c := pipe.Get(ctx, "C") + 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)) @@ -547,7 +545,8 @@ var _ = Describe("ClusterClient", func() { AfterEach(func() {}) - assertPipeline() + keys := []string{"A", "B", "C", "D", "E", "F", "G"} + assertPipeline(keys) It("doesn't fail node with context.Canceled error", func() { ctx, cancel := context.WithCancel(context.Background()) @@ -590,7 +589,10 @@ var _ = Describe("ClusterClient", func() { AfterEach(func() {}) - assertPipeline() + // 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) }) }) From 68c90f78de79804009a679820e885460c94d4cda Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov Date: Tue, 17 Jun 2025 11:43:28 +0300 Subject: [PATCH 3/5] chore(err): Extract crossslot err and add test --- error.go | 6 ++++++ osscluster.go | 5 ++--- osscluster_test.go | 14 ++++++++++++++ 3 files changed, 22 insertions(+), 3 deletions(-) diff --git a/error.go b/error.go index 8c811966f..8013de44a 100644 --- a/error.go +++ b/error.go @@ -22,6 +22,12 @@ var ErrPoolExhausted = pool.ErrPoolExhausted // ErrPoolTimeout timed out waiting to get a connection from the connection pool. var ErrPoolTimeout = pool.ErrPoolTimeout +// ErrCrossSlot is returned when keys are used in the same Redis command and +// the keys are not in the same hash slot. This error is returned by Redis +// Cluster and will be returned by the client when TxPipeline or TxPipelined +// is used on a ClusterClient with keys in different slots. +var ErrCrossSlot = proto.RedisError("CROSSSLOT Keys in request don't hash to the same slot") + // HasErrorPrefix checks if the err is a Redis error and the message contains a prefix. func HasErrorPrefix(err error, prefix string) bool { var rErr Error diff --git a/osscluster.go b/osscluster.go index 06a588567..71c9a892e 100644 --- a/osscluster.go +++ b/osscluster.go @@ -1506,9 +1506,8 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err cmdsMap := c.mapCmdsBySlot(cmds) // TxPipeline does not support cross slot transaction. if len(cmdsMap) > 1 { - err := fmt.Errorf("redis: CROSSSLOT Keys in request don't hash to the same slot") - setCmdsErr(cmds, err) - return err + setCmdsErr(cmds, ErrCrossSlot) + return ErrCrossSlot } if len(cmdsMap) == 0 { return nil diff --git a/osscluster_test.go b/osscluster_test.go index 7b0a8b292..80dfad7b2 100644 --- a/osscluster_test.go +++ b/osscluster_test.go @@ -593,6 +593,20 @@ var _ = Describe("ClusterClient", func() { // 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) + _, err := pipe.Exec(ctx) + Expect(err).To(MatchError(redis.ErrCrossSlot)) + }) + + // 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()) + }) }) }) From c22db9a297031f7f67669b77c952fa7ae786eabf Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov Date: Tue, 17 Jun 2025 14:26:10 +0300 Subject: [PATCH 4/5] fix(txpipeline): short curcuit the tx if there are no commands --- osscluster.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/osscluster.go b/osscluster.go index 71c9a892e..0dce50a4a 100644 --- a/osscluster.go +++ b/osscluster.go @@ -1497,6 +1497,10 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err // Trim multi .. exec. cmds = cmds[1 : len(cmds)-1] + if len(cmds) == 0 { + return nil + } + state, err := c.state.Get(ctx) if err != nil { setCmdsErr(cmds, err) @@ -1509,9 +1513,6 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err setCmdsErr(cmds, ErrCrossSlot) return ErrCrossSlot } - if len(cmdsMap) == 0 { - return nil - } for slot, cmds := range cmdsMap { node, err := state.slotMasterNode(slot) From 33aecb8350d831e11a6f74b2754087668d8c14e2 Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov Date: Wed, 18 Jun 2025 11:07:48 +0300 Subject: [PATCH 5/5] chore(tests): validate keys are in different slots --- osscluster_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/osscluster_test.go b/osscluster_test.go index 80dfad7b2..10023218d 100644 --- a/osscluster_test.go +++ b/osscluster_test.go @@ -598,6 +598,7 @@ var _ = Describe("ClusterClient", func() { 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)) })