mirror of
https://github.com/redis/go-redis.git
synced 2026-10-28 08:24:54 +08:00
feat(cmd): bless (#4041)
* feat: add bless commands * fix: use int64 for count * fix: keylessnes, docs, tests
This commit is contained in:
1 parent
6f2b2639e5
commit
e4a198a8c7
8 files changed
+630
-2
No files matched your search
@@ -51,6 +51,7 @@ runs:
|
||||
elif [[ -n "$REDIS_VERSION" ]]; then
|
||||
# Mapping of redis version to redis testing containers
|
||||
declare -A redis_version_mapping=(
|
||||
["8.12.x"]="8.12-m02-int"
|
||||
["8.10.x"]="8.10.0"
|
||||
["8.6.x"]="8.6.1"
|
||||
["8.4.x"]="8.4.0"
|
||||
|
||||
@@ -46,6 +46,7 @@ jobs:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
redis-version:
|
||||
- "8.12.x" # Redis CE 8.12 (pre-release)
|
||||
- "8.10.x" # Redis CE 8.10
|
||||
- "8.6.x" # Redis CE 8.6
|
||||
- "8.4.x" # Redis CE 8.4
|
||||
@@ -72,6 +73,7 @@ jobs:
|
||||
|
||||
# Mapping of redis version to redis testing containers
|
||||
declare -A redis_version_mapping=(
|
||||
["8.12.x"]="8.12-m02-int"
|
||||
["8.10.x"]="8.10.0"
|
||||
["8.6.x"]="8.6.1"
|
||||
["8.4.x"]="8.4.0"
|
||||
@@ -105,6 +107,7 @@ jobs:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
redis-version:
|
||||
- "8.12.x" # Redis CE 8.12 (pre-release)
|
||||
- "8.10.x" # Redis CE 8.10
|
||||
- "8.6.x" # Redis CE 8.6
|
||||
- "8.4.x" # Redis CE 8.4
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package redis
|
||||
|
||||
import "context"
|
||||
|
||||
type BlessFlag string
|
||||
|
||||
const (
|
||||
BlessNoEvict BlessFlag = "NO-EVICT"
|
||||
)
|
||||
|
||||
type BlessCmdable interface {
|
||||
// BlessSet sets flag on key. It returns 1 if the flag was newly set and 0
|
||||
// if key already carried it.
|
||||
BlessSet(ctx context.Context, key string, flag BlessFlag) *IntCmd
|
||||
// BlessGet returns the flags currently set on key.
|
||||
BlessGet(ctx context.Context, key string) *StringSliceCmd
|
||||
// BlessClear removes flag from key. It returns 1 if the flag was present
|
||||
// and 0 otherwise.
|
||||
BlessClear(ctx context.Context, key string, flag BlessFlag) *IntCmd
|
||||
// BlessScan iterates the keys on a single server that carry flag, using
|
||||
// the usual SCAN cursor protocol. Pass count <= 0 to omit the COUNT hint.
|
||||
//
|
||||
// The cursor is only meaningful on the node that issued it, and each node
|
||||
// only knows about its own blessed keys. BLESS SCAN is keyless, so a
|
||||
// ClusterClient or Ring routes every call to an arbitrary node, and a
|
||||
// cursor returned by one node is useless on another: the iteration will
|
||||
// miss keys or never terminate. Do not call BlessScan on a ClusterClient
|
||||
// or Ring directly. Scan each master instead, e.g. with
|
||||
// ClusterClient.ForEachMaster or Ring.ForEachShard, and merge the pages.
|
||||
BlessScan(ctx context.Context, cursor uint64, flag BlessFlag, count int64) *ScanCmd
|
||||
}
|
||||
|
||||
func (c cmdable) BlessSet(ctx context.Context, key string, flag BlessFlag) *IntCmd {
|
||||
args := []any{"bless", "set", key, string(flag)}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
cmd.SetFirstKeyPos(2)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
func (c cmdable) BlessGet(ctx context.Context, key string) *StringSliceCmd {
|
||||
args := []any{"bless", "get", key}
|
||||
cmd := NewStringSliceCmd(ctx, args...)
|
||||
cmd.SetFirstKeyPos(2)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
func (c cmdable) BlessClear(ctx context.Context, key string, flag BlessFlag) *IntCmd {
|
||||
args := []any{"bless", "clear", key, string(flag)}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
cmd.SetFirstKeyPos(2)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// BlessScan iterates the blessed keys of a single server. The cursor is
|
||||
// node-local, so on a ClusterClient or Ring scan each master separately; see
|
||||
// BlessCmdable.BlessScan.
|
||||
func (c cmdable) BlessScan(ctx context.Context, cursor uint64, flag BlessFlag, count int64) *ScanCmd {
|
||||
args := []any{"bless", "scan", cursor, string(flag)}
|
||||
if count > 0 {
|
||||
args = append(args, "count", count)
|
||||
}
|
||||
cmd := NewScanCmd(ctx, c, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
@@ -0,0 +1,404 @@
|
||||
package redis_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"sync"
|
||||
|
||||
. "github.com/bsm/ginkgo/v2"
|
||||
. "github.com/bsm/gomega"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// blessedKeysRe matches the top-level "blessed_keys:N" field in INFO output.
|
||||
var blessedKeysRe = regexp.MustCompile(`(?m)^blessed_keys:(\d+)\r?$`)
|
||||
|
||||
// blessScanAll drains BLESS SCAN on a single node and returns every key it
|
||||
// reports. The BLESS SCAN cursor is node-local, so this is the building block
|
||||
// for scanning a ClusterClient or Ring master by master.
|
||||
func blessScanAll(ctx context.Context, node *redis.Client, flag redis.BlessFlag) ([]string, error) {
|
||||
var keys []string
|
||||
var cursor uint64
|
||||
for {
|
||||
page, next, err := node.BlessScan(ctx, cursor, flag, 5).Result()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keys = append(keys, page...)
|
||||
if next == 0 {
|
||||
return keys, nil
|
||||
}
|
||||
cursor = next
|
||||
}
|
||||
}
|
||||
|
||||
// blessScanEachNode runs blessScanAll on every node visited by forEach
|
||||
// (ClusterClient.ForEachMaster or Ring.ForEachShard) and merges the results.
|
||||
// forEach runs its callback concurrently, hence the mutex.
|
||||
func blessScanEachNode(
|
||||
ctx context.Context,
|
||||
flag redis.BlessFlag,
|
||||
forEach func(context.Context, func(context.Context, *redis.Client) error) error,
|
||||
) ([]string, error) {
|
||||
var mu sync.Mutex
|
||||
var all []string
|
||||
err := forEach(ctx, func(ctx context.Context, node *redis.Client) error {
|
||||
keys, err := blessScanAll(ctx, node, flag)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mu.Lock()
|
||||
all = append(all, keys...)
|
||||
mu.Unlock()
|
||||
return nil
|
||||
})
|
||||
return all, err
|
||||
}
|
||||
|
||||
var _ = Describe("BLESS commands", func() {
|
||||
ctx := context.TODO()
|
||||
var client *redis.Client
|
||||
|
||||
const unknownFlag = redis.BlessFlag("NO-SUCH-FLAG")
|
||||
|
||||
// blessedKeys returns the blessed_keys counter from INFO, failing the spec
|
||||
// if the server does not report one.
|
||||
blessedKeys := func() int {
|
||||
GinkgoHelper()
|
||||
info, err := client.Info(ctx, "everything").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
m := blessedKeysRe.FindStringSubmatch(info)
|
||||
Expect(m).NotTo(BeNil(), "INFO does not report blessed_keys")
|
||||
n, err := strconv.Atoi(m[1])
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
return n
|
||||
}
|
||||
|
||||
BeforeEach(func() {
|
||||
SkipBeforeRedisVersion("8.12", "BLESS requires Redis 8.12")
|
||||
client = redis.NewClient(redisOptions())
|
||||
Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
AfterEach(func() {
|
||||
Expect(client.Close()).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
Describe("BLESS SET", func() {
|
||||
It("returns 1 when the flag is newly set and 0 when it was already set", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
Expect(client.BlessSet(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(client.BlessSet(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(0)))
|
||||
|
||||
flags, err := client.BlessGet(ctx, "key1").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(flags).To(Equal([]string{string(redis.BlessNoEvict)}))
|
||||
})
|
||||
|
||||
It("rejects an unknown flag", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
err := client.BlessSet(ctx, "key1", unknownFlag).Err()
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err.Error()).To(HavePrefix("ERR"))
|
||||
|
||||
// A rejected flag must not leave any state behind.
|
||||
flags, err := client.BlessGet(ctx, "key1").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(flags).To(BeEmpty())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("BLESS GET", func() {
|
||||
It("reports no flags on a key that never had one set", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
flags, err := client.BlessGet(ctx, "key1").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(flags).To(BeEmpty())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("BLESS CLEAR", func() {
|
||||
It("returns 1 when the flag was present and 0 when it was not", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.BlessSet(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
|
||||
Expect(client.BlessClear(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(client.BlessClear(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(0)))
|
||||
|
||||
flags, err := client.BlessGet(ctx, "key1").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(flags).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns 0 for a key that was never blessed", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
Expect(client.BlessClear(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(0)))
|
||||
})
|
||||
|
||||
It("rejects an unknown flag", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.BlessSet(ctx, "key1", redis.BlessNoEvict).Err()).NotTo(HaveOccurred())
|
||||
|
||||
err := client.BlessClear(ctx, "key1", unknownFlag).Err()
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err.Error()).To(HavePrefix("ERR"))
|
||||
|
||||
// The existing flag must survive a rejected CLEAR.
|
||||
flags, err := client.BlessGet(ctx, "key1").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(flags).To(Equal([]string{string(redis.BlessNoEvict)}))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("BLESS SCAN", func() {
|
||||
It("scans keys carrying a flag across multiple pages", func() {
|
||||
want := make([]string, 0, 20)
|
||||
for i := range 20 {
|
||||
key := fmt.Sprintf("scan:%02d", i)
|
||||
want = append(want, key)
|
||||
Expect(client.Set(ctx, key, "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.BlessSet(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
}
|
||||
// An unblessed key must not show up in the scan.
|
||||
Expect(client.Set(ctx, "plain", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
var keys []string
|
||||
cursor := uint64(0)
|
||||
for {
|
||||
page, next, err := client.BlessScan(ctx, cursor, redis.BlessNoEvict, 5).Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
keys = append(keys, page...)
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
Expect(keys).To(ConsistOf(want))
|
||||
})
|
||||
|
||||
It("returns an empty result when no key carries the flag", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
keys, cursor, err := client.BlessScan(ctx, 0, redis.BlessNoEvict, 0).Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(keys).To(BeEmpty())
|
||||
Expect(cursor).To(Equal(uint64(0)))
|
||||
})
|
||||
|
||||
It("rejects an unknown flag", func() {
|
||||
err := client.BlessScan(ctx, 0, unknownFlag, 0).Err()
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err.Error()).To(HavePrefix("ERR"))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("INFO blessed_keys", func() {
|
||||
It("tracks the number of blessed keys through SET, CLEAR and DEL", func() {
|
||||
Expect(blessedKeys()).To(Equal(0))
|
||||
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.Set(ctx, "key2", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
|
||||
Expect(client.BlessSet(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(blessedKeys()).To(Equal(1))
|
||||
|
||||
Expect(client.BlessSet(ctx, "key2", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(blessedKeys()).To(Equal(2))
|
||||
|
||||
// Re-setting an existing flag must not double count.
|
||||
Expect(client.BlessSet(ctx, "key2", redis.BlessNoEvict).Result()).To(Equal(int64(0)))
|
||||
Expect(blessedKeys()).To(Equal(2))
|
||||
|
||||
Expect(client.BlessClear(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(blessedKeys()).To(Equal(1))
|
||||
|
||||
// Deleting a blessed key releases its slot in the counter.
|
||||
Expect(client.Del(ctx, "key2").Result()).To(Equal(int64(1)))
|
||||
Expect(blessedKeys()).To(Equal(0))
|
||||
})
|
||||
|
||||
It("resets to zero on FLUSHDB", func() {
|
||||
Expect(client.Set(ctx, "key1", "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.BlessSet(ctx, "key1", redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(blessedKeys()).To(Equal(1))
|
||||
|
||||
Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
|
||||
Expect(blessedKeys()).To(Equal(0))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
// BLESS SET/GET/CLEAR carry the key at position 2, so a ClusterClient routes
|
||||
// them to the owning master like any other keyed command. BLESS SCAN is
|
||||
// keyless and its cursor is node-local, so it has to be issued to each master
|
||||
// individually; these specs pin both halves of that contract.
|
||||
var _ = Describe("BLESS commands on ClusterClient", func() {
|
||||
ctx := context.TODO()
|
||||
var client *redis.ClusterClient
|
||||
|
||||
// Enough keys to land on every master of the three-shard test cluster
|
||||
// with overwhelming probability.
|
||||
const numKeys = 60
|
||||
|
||||
BeforeEach(func() {
|
||||
if RECluster {
|
||||
Skip("OSS cluster scenario is not configured under RE_CLUSTER")
|
||||
}
|
||||
SkipBeforeRedisVersion("8.12", "BLESS requires Redis 8.12")
|
||||
|
||||
client = cluster.newClusterClient(ctx, redisClusterOptions())
|
||||
Expect(client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
|
||||
return master.FlushDB(ctx).Err()
|
||||
})).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
AfterEach(func() {
|
||||
_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
|
||||
return master.FlushDB(ctx).Err()
|
||||
})
|
||||
Expect(client.Close()).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
It("routes SET, GET and CLEAR to the master owning the key", func() {
|
||||
for i := range numKeys {
|
||||
key := fmt.Sprintf("cluster:%02d", i)
|
||||
Expect(client.Set(ctx, key, "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.BlessSet(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(client.BlessGet(ctx, key).Result()).To(Equal([]string{string(redis.BlessNoEvict)}))
|
||||
}
|
||||
for i := range numKeys {
|
||||
key := fmt.Sprintf("cluster:%02d", i)
|
||||
Expect(client.BlessClear(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(client.BlessGet(ctx, key).Result()).To(BeEmpty())
|
||||
}
|
||||
})
|
||||
|
||||
It("finds every blessed key only when each master is scanned separately", func() {
|
||||
want := make([]string, 0, numKeys)
|
||||
for i := range numKeys {
|
||||
key := fmt.Sprintf("cluster:%02d", i)
|
||||
want = append(want, key)
|
||||
Expect(client.Set(ctx, key, "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.BlessSet(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
}
|
||||
|
||||
// Each master only reports its own slice of the keyspace.
|
||||
var mu sync.Mutex
|
||||
perMaster := map[string][]string{}
|
||||
Expect(client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
|
||||
keys, err := blessScanAll(ctx, master, redis.BlessNoEvict)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mu.Lock()
|
||||
perMaster[master.Options().Addr] = keys
|
||||
mu.Unlock()
|
||||
return nil
|
||||
})).NotTo(HaveOccurred())
|
||||
Expect(perMaster).To(HaveLen(len(cluster.masters())))
|
||||
for addr, keys := range perMaster {
|
||||
Expect(keys).NotTo(BeEmpty(), "master %s reported no blessed keys", addr)
|
||||
Expect(len(keys)).To(BeNumerically("<", numKeys), "master %s reported every key", addr)
|
||||
}
|
||||
|
||||
// Merging the per-master scans yields the complete, duplicate-free set.
|
||||
all, err := blessScanEachNode(ctx, redis.BlessNoEvict, client.ForEachMaster)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(all).To(ConsistOf(want))
|
||||
|
||||
// A single BLESS SCAN through the ClusterClient lands on one arbitrary
|
||||
// master and therefore cannot see the whole keyspace, which is why the
|
||||
// per-master loop above is the supported pattern.
|
||||
page, _, err := client.BlessScan(ctx, 0, redis.BlessNoEvict, numKeys*2).Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(len(page)).To(BeNumerically("<", numKeys))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("BLESS commands on Ring", func() {
|
||||
ctx := context.TODO()
|
||||
var ring *redis.Ring
|
||||
|
||||
// Enough keys to land on both ring shards with overwhelming probability.
|
||||
const numKeys = 40
|
||||
|
||||
BeforeEach(func() {
|
||||
if RECluster {
|
||||
Skip("ring shards are not configured under RE_CLUSTER")
|
||||
}
|
||||
SkipBeforeRedisVersion("8.12", "BLESS requires Redis 8.12")
|
||||
|
||||
ring = redis.NewRing(redisRingOptions())
|
||||
Expect(ring.ForEachShard(ctx, func(ctx context.Context, shard *redis.Client) error {
|
||||
return shard.FlushDB(ctx).Err()
|
||||
})).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
AfterEach(func() {
|
||||
_ = ring.ForEachShard(ctx, func(ctx context.Context, shard *redis.Client) error {
|
||||
return shard.FlushDB(ctx).Err()
|
||||
})
|
||||
Expect(ring.Close()).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
It("routes SET, GET and CLEAR to the shard owning the key", func() {
|
||||
for i := range numKeys {
|
||||
key := fmt.Sprintf("ring:%02d", i)
|
||||
Expect(ring.Set(ctx, key, "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(ring.BlessSet(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(ring.BlessGet(ctx, key).Result()).To(Equal([]string{string(redis.BlessNoEvict)}))
|
||||
}
|
||||
for i := range numKeys {
|
||||
key := fmt.Sprintf("ring:%02d", i)
|
||||
Expect(ring.BlessClear(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
Expect(ring.BlessGet(ctx, key).Result()).To(BeEmpty())
|
||||
}
|
||||
})
|
||||
|
||||
It("finds every blessed key only when each shard is scanned separately", func() {
|
||||
want := make([]string, 0, numKeys)
|
||||
for i := range numKeys {
|
||||
key := fmt.Sprintf("ring:%02d", i)
|
||||
want = append(want, key)
|
||||
Expect(ring.Set(ctx, key, "v", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(ring.BlessSet(ctx, key, redis.BlessNoEvict).Result()).To(Equal(int64(1)))
|
||||
}
|
||||
|
||||
// Each shard only reports the keys hashed to it.
|
||||
var mu sync.Mutex
|
||||
perShard := map[string][]string{}
|
||||
Expect(ring.ForEachShard(ctx, func(ctx context.Context, shard *redis.Client) error {
|
||||
keys, err := blessScanAll(ctx, shard, redis.BlessNoEvict)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mu.Lock()
|
||||
perShard[shard.Options().Addr] = keys
|
||||
mu.Unlock()
|
||||
return nil
|
||||
})).NotTo(HaveOccurred())
|
||||
Expect(perShard).To(HaveLen(ring.Len()))
|
||||
for addr, keys := range perShard {
|
||||
Expect(keys).NotTo(BeEmpty(), "shard %s reported no blessed keys", addr)
|
||||
Expect(len(keys)).To(BeNumerically("<", numKeys), "shard %s reported every key", addr)
|
||||
}
|
||||
|
||||
// Merging the per-shard scans yields the complete, duplicate-free set.
|
||||
all, err := blessScanEachNode(ctx, redis.BlessNoEvict, ring.ForEachShard)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(all).To(ConsistOf(want))
|
||||
|
||||
// A single BLESS SCAN through the Ring goes to a random shard and
|
||||
// cannot see the whole keyspace.
|
||||
page, _, err := ring.BlessScan(ctx, 0, redis.BlessNoEvict, numKeys*2).Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(len(page)).To(BeNumerically("<", numKeys))
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,125 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// These tests assert the argument lists and key-position hints built by the
|
||||
// BLESS commands without dispatching them to a server, since a test Redis
|
||||
// instance is not guaranteed to support the BLESS command family.
|
||||
|
||||
func TestBlessSet_Args(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
var captured Cmder
|
||||
c := captureCmdable(&captured)
|
||||
c.BlessSet(ctx, "key1", BlessNoEvict)
|
||||
|
||||
want := []any{"bless", "set", "key1", "NO-EVICT"}
|
||||
if !reflect.DeepEqual(captured.Args(), want) {
|
||||
t.Errorf("args = %#v, want %#v", captured.Args(), want)
|
||||
}
|
||||
if pos := captured.firstKeyPos(); pos != 2 {
|
||||
t.Errorf("firstKeyPos = %d, want 2", pos)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlessGet_Args(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
var captured Cmder
|
||||
c := captureCmdable(&captured)
|
||||
c.BlessGet(ctx, "key1")
|
||||
|
||||
want := []any{"bless", "get", "key1"}
|
||||
if !reflect.DeepEqual(captured.Args(), want) {
|
||||
t.Errorf("args = %#v, want %#v", captured.Args(), want)
|
||||
}
|
||||
if pos := captured.firstKeyPos(); pos != 2 {
|
||||
t.Errorf("firstKeyPos = %d, want 2", pos)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlessClear_Args(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
var captured Cmder
|
||||
c := captureCmdable(&captured)
|
||||
c.BlessClear(ctx, "key1", BlessNoEvict)
|
||||
|
||||
want := []any{"bless", "clear", "key1", "NO-EVICT"}
|
||||
if !reflect.DeepEqual(captured.Args(), want) {
|
||||
t.Errorf("args = %#v, want %#v", captured.Args(), want)
|
||||
}
|
||||
if pos := captured.firstKeyPos(); pos != 2 {
|
||||
t.Errorf("firstKeyPos = %d, want 2", pos)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlessScan_Args(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
cursor uint64
|
||||
count int64
|
||||
want []any
|
||||
}{
|
||||
{
|
||||
name: "count_omitted",
|
||||
cursor: 0,
|
||||
count: 0,
|
||||
want: []any{"bless", "scan", uint64(0), "NO-EVICT"},
|
||||
},
|
||||
{
|
||||
name: "with_count",
|
||||
cursor: 10,
|
||||
count: 100,
|
||||
want: []any{"bless", "scan", uint64(10), "NO-EVICT", "count", int64(100)},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
var captured Cmder
|
||||
c := captureCmdable(&captured)
|
||||
c.BlessScan(ctx, tt.cursor, BlessNoEvict, tt.count)
|
||||
|
||||
if !reflect.DeepEqual(captured.Args(), tt.want) {
|
||||
t.Errorf("args = %#v, want %#v", captured.Args(), tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Only BLESS SCAN is keyless; BLESS SET/GET/CLEAR carry the key at position 2.
|
||||
// Raw Do("bless", ...) calls have no SetFirstKeyPos hint, so the fallback in
|
||||
// cmdFirstKeyPosWithInfo must distinguish the subcommand. Otherwise a Ring
|
||||
// would route key-bearing BLESS calls to a random shard.
|
||||
func TestBlessFirstKeyPos_RawDo(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
tests := []struct {
|
||||
name string
|
||||
args []any
|
||||
want int
|
||||
}{
|
||||
{name: "scan_keyless", args: []any{"bless", "scan", 0, "NO-EVICT"}, want: 0},
|
||||
{name: "scan_uppercase", args: []any{"BLESS", "SCAN", 0, "NO-EVICT"}, want: 0},
|
||||
{name: "set", args: []any{"bless", "set", "key1", "NO-EVICT"}, want: 2},
|
||||
{name: "get", args: []any{"bless", "get", "key1"}, want: 2},
|
||||
{name: "clear", args: []any{"bless", "clear", "key1", "NO-EVICT"}, want: 2},
|
||||
{name: "set_uppercase", args: []any{"BLESS", "SET", "key1", "NO-EVICT"}, want: 2},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
cmd := NewCmd(ctx, tt.args...)
|
||||
if got := cmdFirstKeyPosWithInfo(cmd, nil); got != tt.want {
|
||||
t.Errorf("cmdFirstKeyPosWithInfo(%v) = %d, want %d", tt.args, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
if _, ok := keylessCommands["bless"]; ok {
|
||||
t.Error(`keylessCommands["bless"] present: raw BLESS SET/GET/CLEAR would route to a random shard`)
|
||||
}
|
||||
}
|
||||
@@ -411,6 +411,15 @@ func cmdFirstKeyPosWithInfo(cmd Cmder, info *CommandInfo) int {
|
||||
return 3
|
||||
}
|
||||
return migrateKeysPos(cmd)
|
||||
case "bless":
|
||||
// BLESS SCAN is keyless (cursor-based, server-wide); BLESS SET/GET/CLEAR
|
||||
// take the key at position 2. The typed methods set this via
|
||||
// SetFirstKeyPos; this fallback covers raw Do("bless", ...) calls so a
|
||||
// Ring/Cluster doesn't send them to a random shard.
|
||||
if internal.ToLower(cmd.stringArg(1)) == "scan" {
|
||||
return 0
|
||||
}
|
||||
return 2
|
||||
}
|
||||
|
||||
// Use CommandInfo cache when warm (in-memory only, no extra round-trips).
|
||||
|
||||
@@ -251,6 +251,7 @@ type Cmdable interface {
|
||||
TimeseriesCmdable
|
||||
JSONCmdable
|
||||
VectorSetCmdable
|
||||
BlessCmdable
|
||||
}
|
||||
|
||||
type StatefulCmdable interface {
|
||||
|
||||
+19
-2
@@ -2729,6 +2729,24 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
|
||||
// query never times out and the spec hangs until the suite timeout.
|
||||
It("should stop processing and return an error when a timeout occurs", Label("search", "ftaggregate", "NonRedisEnterprise"), func() {
|
||||
SkipBeforeRedisVersion("7.9", "requires Redis 8.x")
|
||||
|
||||
// The 1 ms timeout is only enforced deterministically when the query
|
||||
// runs on the main thread. With search-workers > 0 the aggregate is
|
||||
// offloaded to the worker pool, where the timeout check can be skipped
|
||||
// and the query completes successfully. Pin workers to 0 for this spec
|
||||
// and restore the previous values afterwards.
|
||||
prevConfig, err := rawClient.ConfigGet(ctx, "search-*").Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
for _, param := range []string{"search-workers", "search-on-timeout"} {
|
||||
Expect(prevConfig).To(HaveKey(param), "server does not expose %s", param)
|
||||
}
|
||||
defer func() {
|
||||
for _, param := range []string{"search-workers", "search-on-timeout"} {
|
||||
Expect(rawClient.ConfigSet(ctx, param, prevConfig[param]).Err()).NotTo(HaveOccurred())
|
||||
}
|
||||
}()
|
||||
Expect(rawClient.ConfigSet(ctx, "search-workers", "0").Err()).NotTo(HaveOccurred())
|
||||
|
||||
val, err := client.FTCreate(ctx, "aggTimeoutHeavy", &redis.FTCreateOptions{},
|
||||
&redis.FieldSchema{FieldName: "n", FieldType: redis.SearchFieldTypeNumeric, Sortable: true},
|
||||
).Result()
|
||||
@@ -2743,8 +2761,7 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
}
|
||||
// default behaviour was changed in 8.0.1, set to fail to validate the timeout was triggered
|
||||
err = rawClient.ConfigSet(ctx, "search-on-timeout", "fail").Err()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(rawClient.ConfigSet(ctx, "search-on-timeout", "fail").Err()).NotTo(HaveOccurred())
|
||||
|
||||
options := &redis.FTAggregateOptions{
|
||||
SortBy: []redis.FTAggregateSortBy{{FieldName: "@n", Desc: true}},
|
||||
|
||||
Reference in new issue
Block a user