redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2025-07-28 06:42:00 +03:00
Code Activity

fix(txpipeline): should return error on multi/exec on multiple slots [CAE-1028] (#3408)

Browse Source 
* fix(txpipeline): should return error on multi/exec on multiple slots

* fix(txpipeline): test normal tx pipeline behaviour

* chore(err): Extract crossslot err and add test

* fix(txpipeline): short curcuit the tx if there are no commands

* chore(tests): validate keys are in different slots
This commit is contained in:
Nedyalko Dyakov
2025-06-18 14:18:00 +03:00
committed by GitHub
parent 68717412c9
commit 4c635cc563
3 changed files with 49 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
osscluster_test.go
View File

@ -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())
})
})
})