Repository: redis/go-redis

Issue:
Why does go-redis allow cluster mode cross-shard transactions and pipelines?
Cross-shard/cross-slot batch operations in cluster mode-enabled clusters, like transactions and pipelines, are not supported in a Redis cluster, by specification.

* There is no server-side primitive that enables cross-shard transactional atomicity.
* A single pipeline cannot physically be applied to multiple targets, since it occurs over a single physical connection.

Other client libraries properly reject these operations with a cross-slot validation error. The go-redis library seems to instead (1) partition the batch commands by slot, (2) concurrently dispatch individual batch commands to each shard, and (3) aggregate the results client-side.

https://github.com/redis/go-redis/blob/v9.5.2/osscluster.go#L1237

As a simple reproducible example:

```
127.0.0.1:7000> set a aaaaa
-> Redirected to slot [15495] located at 127.0.0.1:7002
OK
127.0.0.1:7002> set b bbbbb
-> Redirected to slot [3300] located at 127.0.0.1:7000
OK
```

```go
ctx := context.Background()
pipe := client.TxPipeline()
pipe.Get(ctx, "a")
pipe.Get(ctx, "b")
fmt.Println(pipe.Exec(context.Background()))  // [aaaaa bbbbb]
```

This is both extremely surprising behavior, and the source of potentially severe correctness/consistency problems. I don't see any mechanism to opt out or otherwise disable the "smart virtual batching" exercised by the library.

Why does go-redis do this? Is it possible to implement a client option to disable this behavior?

## Expected Behavior

Cross-slot transactions and pipelines should return an error.

## Current Behavior

go-redis internally separates out commands by shard and executes them independently without error.

## Possible Solution

Provide a switch to allow the client to reject batch operations when they per-shard partitioning hashes the set of commands to multiple slots.

## Steps to Reproduce

1. Create a cluster with at least 2 nodes.
2. Write sample keys spanning both nodes.
3. Exercise the code snippet above.
4. Observe that no error is returned.

## Context (Environment)

Clients are expecting transactional atomicity which can only be guaranteed when the transaction affects keys that hash to the same slot.

## Detailed Description

See above.

## Possible Implementation

N/A

Hints from issue comments:
@ndyakov: Hello @LINKIWI, thank you for the detailed issue. I am currently reviewing this and discussing with the rest of the libraries how this is handled with them. What you are suggesting makes sense and if this is the case for the rest of the clients I would prefer to have it consistent. If we consider this a bug, we can fix it before `v10`, otherwise it will be a breaking change and should be included in `v10`@axilis-marko: I was watching this issue as I noticed the same thing. What I came up on today is that while I see this as a useful change for Transactions, the current behaviour could remain for Pipelines. Pipelines per themselves have only the guarantee that there is a single round trip. So a client side optimisation. If I need things to happen/fetch from different Nodes, it makes sense that I don't wait for the first one to happen, only to emit the other commands. And as I didn't request a transaction, it doesn't break any contracts.@ndyakov: @LINKIWI , I do agree with @axilis-marko and actually prepared a PR #3408 for resolving this for transactions, but kept the logic for pipelines. Let me know what you think. @turfaa: I got affected by this change 😅

I was making use of the old `TxPipelined` behavior, then things started to go on fire.

I agree that this should be the correct behavior though.

Solution patch:
diff --git a/error.go b/error.go
index 8c81196..8013de4 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 a68f7ea..0dce50a 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)
@@ -1504,6 +1508,12 @@ 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 {
+		setCmdsErr(cmds, ErrCrossSlot)
+		return ErrCrossSlot
+	}
+
 	for slot, cmds := range cmdsMap {
 		node, err := state.slotMasterNode(slot)
 		if err != nil {



Test patch:
diff --git a/osscluster_test.go b/osscluster_test.go
index 9934118..1002321 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,25 @@ 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)
+
+				// 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))
+				})
+
+				// 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())
+				})
 			})
 		})