Merge pull request #140 from alexgridx/dependabot/go_modules/github.com/redis/go-redis/v9-9.5.4

Bump github.com/redis/go-redis/v9 from 9.5.3 to 9.5.4

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/lib/pq v1.10.9
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.19.1
-	github.com/redis/go-redis/v9 v9.5.3
+	github.com/redis/go-redis/v9 v9.5.4
 )
 
 require (

go.sum

@@ -124,8 +124,8 @@ github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G
 github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
-github.com/redis/go-redis/v9 v9.5.3 h1:fOAp1/uJG+ZtcITgZOfYFmTKPE7n4Vclj1wZFgRciUU=
+github.com/redis/go-redis/v9 v9.5.4 h1:vOFYDKKVgrI5u++QvnMT7DksSMYg7Aw/Np4vLJLKLwY=
-github.com/redis/go-redis/v9 v9.5.3/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
+github.com/redis/go-redis/v9 v9.5.4/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
 github.com/rogpeppe/fastuuid v1.1.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/smartystreets/assertions v1.0.0/go.mod h1:kHHU4qYBaI3q23Pp3VPrmWhuIUrLW/7eUrw0BU5VaoM=

vendor/github.com/redis/go-redis/v9/Makefile

@@ -31,7 +31,7 @@ build:
 
 testdata/redis:
 	mkdir -p $@
-	wget -qO- https://download.redis.io/releases/redis-7.2.1.tar.gz | tar xvz --strip-components=1 -C $@
+	wget -qO- https://download.redis.io/releases/redis-7.4-rc2.tar.gz  | tar xvz --strip-components=1 -C $@
 
 testdata/redis/src/redis-server: testdata/redis
 	cd $< && make all

vendor/github.com/redis/go-redis/v9/command.go

@@ -573,6 +573,10 @@ func (cmd *StatusCmd) Result() (string, error) {
 	return cmd.val, cmd.err
 }
 
+func (cmd *StatusCmd) Bytes() ([]byte, error) {
+	return util.StringToBytes(cmd.val), cmd.err
+}
+
 func (cmd *StatusCmd) String() string {
 	return cmdString(cmd, cmd.val)
 }
@@ -3783,6 +3787,65 @@ func (cmd *MapStringStringSliceCmd) readReply(rd *proto.Reader) error {
 	return nil
 }
 
+// -----------------------------------------------------------------------
+// MapStringInterfaceCmd represents a command that returns a map of strings to interface{}.
+type MapMapStringInterfaceCmd struct {
+	baseCmd
+	val map[string]interface{}
+}
+
+func NewMapMapStringInterfaceCmd(ctx context.Context, args ...interface{}) *MapMapStringInterfaceCmd {
+	return &MapMapStringInterfaceCmd{
+		baseCmd: baseCmd{
+			ctx:  ctx,
+			args: args,
+		},
+	}
+}
+
+func (cmd *MapMapStringInterfaceCmd) String() string {
+	return cmdString(cmd, cmd.val)
+}
+
+func (cmd *MapMapStringInterfaceCmd) SetVal(val map[string]interface{}) {
+	cmd.val = val
+}
+
+func (cmd *MapMapStringInterfaceCmd) Result() (map[string]interface{}, error) {
+	return cmd.val, cmd.err
+}
+
+func (cmd *MapMapStringInterfaceCmd) Val() map[string]interface{} {
+	return cmd.val
+}
+
+func (cmd *MapMapStringInterfaceCmd) readReply(rd *proto.Reader) (err error) {
+	n, err := rd.ReadArrayLen()
+	if err != nil {
+		return err
+	}
+
+	data := make(map[string]interface{}, n/2)
+	for i := 0; i < n; i += 2 {
+		_, err := rd.ReadArrayLen()
+		if err != nil {
+			cmd.err = err
+		}
+		key, err := rd.ReadString()
+		if err != nil {
+			cmd.err = err
+		}
+		value, err := rd.ReadString()
+		if err != nil {
+			cmd.err = err
+		}
+		data[key] = value
+	}
+
+	cmd.val = data
+	return nil
+}
+
 //-----------------------------------------------------------------------
 
 type MapStringInterfaceSliceCmd struct {
@@ -4997,6 +5060,7 @@ type ClientInfo struct {
 	PSub               int           // number of pattern matching subscriptions
 	SSub               int           // redis version 7.0.3, number of shard channel subscriptions
 	Multi              int           // number of commands in a MULTI/EXEC context
+	Watch              int           // redis version 7.4 RC1, number of keys this client is currently watching.
 	QueryBuf           int           // qbuf, query buffer length (0 means no query pending)
 	QueryBufFree       int           // qbuf-free, free space of the query buffer (0 means the buffer is full)
 	ArgvMem            int           // incomplete arguments for the next command (already extracted from query buffer)
@@ -5149,6 +5213,8 @@ func parseClientInfo(txt string) (info *ClientInfo, err error) {
 			info.SSub, err = strconv.Atoi(val)
 		case "multi":
 			info.Multi, err = strconv.Atoi(val)
+		case "watch":
+			info.Watch, err = strconv.Atoi(val)
 		case "qbuf":
 			info.QueryBuf, err = strconv.Atoi(val)
 		case "qbuf-free":

vendor/github.com/redis/go-redis/v9/commands.go

@@ -220,6 +220,7 @@ type Cmdable interface {
 	ProbabilisticCmdable
 	PubSubCmdable
 	ScriptingFunctionsCmdable
+	SearchCmdable
 	SetCmdable
 	SortedSetCmdable
 	StringCmdable

vendor/github.com/redis/go-redis/v9/hash_commands.go

@@ -1,6 +1,9 @@
 package redis
 
-import "context"
+import (
+	"context"
+	"time"
+)
 
 type HashCmdable interface {
 	HDel(ctx context.Context, key string, fields ...string) *IntCmd
@@ -16,9 +19,23 @@ type HashCmdable interface {
 	HMSet(ctx context.Context, key string, values ...interface{}) *BoolCmd
 	HSetNX(ctx context.Context, key, field string, value interface{}) *BoolCmd
 	HScan(ctx context.Context, key string, cursor uint64, match string, count int64) *ScanCmd
+	HScanNoValues(ctx context.Context, key string, cursor uint64, match string, count int64) *ScanCmd
 	HVals(ctx context.Context, key string) *StringSliceCmd
 	HRandField(ctx context.Context, key string, count int) *StringSliceCmd
 	HRandFieldWithValues(ctx context.Context, key string, count int) *KeyValueSliceCmd
+	HExpire(ctx context.Context, key string, expiration time.Duration, fields ...string) *IntSliceCmd
+	HExpireWithArgs(ctx context.Context, key string, expiration time.Duration, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd
+	HPExpire(ctx context.Context, key string, expiration time.Duration, fields ...string) *IntSliceCmd
+	HPExpireWithArgs(ctx context.Context, key string, expiration time.Duration, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd
+	HExpireAt(ctx context.Context, key string, tm time.Time, fields ...string) *IntSliceCmd
+	HExpireAtWithArgs(ctx context.Context, key string, tm time.Time, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd
+	HPExpireAt(ctx context.Context, key string, tm time.Time, fields ...string) *IntSliceCmd
+	HPExpireAtWithArgs(ctx context.Context, key string, tm time.Time, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd
+	HPersist(ctx context.Context, key string, fields ...string) *IntSliceCmd
+	HExpireTime(ctx context.Context, key string, fields ...string) *IntSliceCmd
+	HPExpireTime(ctx context.Context, key string, fields ...string) *IntSliceCmd
+	HTTL(ctx context.Context, key string, fields ...string) *IntSliceCmd
+	HPTTL(ctx context.Context, key string, fields ...string) *IntSliceCmd
 }
 
 func (c cmdable) HDel(ctx context.Context, key string, fields ...string) *IntCmd {
@@ -172,3 +189,262 @@ func (c cmdable) HScan(ctx context.Context, key string, cursor uint64, match str
 	_ = c(ctx, cmd)
 	return cmd
 }
+
+func (c cmdable) HScanNoValues(ctx context.Context, key string, cursor uint64, match string, count int64) *ScanCmd {
+	args := []interface{}{"hscan", key, cursor}
+	if match != "" {
+		args = append(args, "match", match)
+	}
+	if count > 0 {
+		args = append(args, "count", count)
+	}
+	args = append(args, "novalues")
+	cmd := NewScanCmd(ctx, c, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+type HExpireArgs struct {
+	NX bool
+	XX bool
+	GT bool
+	LT bool
+}
+
+// HExpire - Sets the expiration time for specified fields in a hash in seconds.
+// The command constructs an argument list starting with "HEXPIRE", followed by the key, duration, any conditional flags, and the specified fields.
+// For more information - https://redis.io/commands/hexpire/
+func (c cmdable) HExpire(ctx context.Context, key string, expiration time.Duration, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HEXPIRE", key, formatSec(ctx, expiration), "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HExpire - Sets the expiration time for specified fields in a hash in seconds.
+// It requires a key, an expiration duration, a struct with boolean flags for conditional expiration settings (NX, XX, GT, LT), and a list of fields.
+// The command constructs an argument list starting with "HEXPIRE", followed by the key, duration, any conditional flags, and the specified fields.
+// For more information - https://redis.io/commands/hexpire/
+func (c cmdable) HExpireWithArgs(ctx context.Context, key string, expiration time.Duration, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HEXPIRE", key, formatSec(ctx, expiration)}
+
+	// only if one argument is true, we can add it to the args
+	// if more than one argument is true, it will cause an error
+	if expirationArgs.NX {
+		args = append(args, "NX")
+	} else if expirationArgs.XX {
+		args = append(args, "XX")
+	} else if expirationArgs.GT {
+		args = append(args, "GT")
+	} else if expirationArgs.LT {
+		args = append(args, "LT")
+	}
+
+	args = append(args, "FIELDS", len(fields))
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HPExpire - Sets the expiration time for specified fields in a hash in milliseconds.
+// Similar to HExpire, it accepts a key, an expiration duration in milliseconds, a struct with expiration condition flags, and a list of fields.
+// The command modifies the standard time.Duration to milliseconds for the Redis command.
+// For more information - https://redis.io/commands/hpexpire/
+func (c cmdable) HPExpire(ctx context.Context, key string, expiration time.Duration, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPEXPIRE", key, formatMs(ctx, expiration), "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+func (c cmdable) HPExpireWithArgs(ctx context.Context, key string, expiration time.Duration, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPEXPIRE", key, formatMs(ctx, expiration)}
+
+	// only if one argument is true, we can add it to the args
+	// if more than one argument is true, it will cause an error
+	if expirationArgs.NX {
+		args = append(args, "NX")
+	} else if expirationArgs.XX {
+		args = append(args, "XX")
+	} else if expirationArgs.GT {
+		args = append(args, "GT")
+	} else if expirationArgs.LT {
+		args = append(args, "LT")
+	}
+
+	args = append(args, "FIELDS", len(fields))
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HExpireAt - Sets the expiration time for specified fields in a hash to a UNIX timestamp in seconds.
+// Takes a key, a UNIX timestamp, a struct of conditional flags, and a list of fields.
+// The command sets absolute expiration times based on the UNIX timestamp provided.
+// For more information - https://redis.io/commands/hexpireat/
+func (c cmdable) HExpireAt(ctx context.Context, key string, tm time.Time, fields ...string) *IntSliceCmd {
+
+	args := []interface{}{"HEXPIREAT", key, tm.Unix(), "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+func (c cmdable) HExpireAtWithArgs(ctx context.Context, key string, tm time.Time, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HEXPIREAT", key, tm.Unix()}
+
+	// only if one argument is true, we can add it to the args
+	// if more than one argument is true, it will cause an error
+	if expirationArgs.NX {
+		args = append(args, "NX")
+	} else if expirationArgs.XX {
+		args = append(args, "XX")
+	} else if expirationArgs.GT {
+		args = append(args, "GT")
+	} else if expirationArgs.LT {
+		args = append(args, "LT")
+	}
+
+	args = append(args, "FIELDS", len(fields))
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HPExpireAt - Sets the expiration time for specified fields in a hash to a UNIX timestamp in milliseconds.
+// Similar to HExpireAt but for timestamps in milliseconds. It accepts the same parameters and adjusts the UNIX time to milliseconds.
+// For more information - https://redis.io/commands/hpexpireat/
+func (c cmdable) HPExpireAt(ctx context.Context, key string, tm time.Time, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPEXPIREAT", key, tm.UnixNano() / int64(time.Millisecond), "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+func (c cmdable) HPExpireAtWithArgs(ctx context.Context, key string, tm time.Time, expirationArgs HExpireArgs, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPEXPIREAT", key, tm.UnixNano() / int64(time.Millisecond)}
+
+	// only if one argument is true, we can add it to the args
+	// if more than one argument is true, it will cause an error
+	if expirationArgs.NX {
+		args = append(args, "NX")
+	} else if expirationArgs.XX {
+		args = append(args, "XX")
+	} else if expirationArgs.GT {
+		args = append(args, "GT")
+	} else if expirationArgs.LT {
+		args = append(args, "LT")
+	}
+
+	args = append(args, "FIELDS", len(fields))
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HPersist - Removes the expiration time from specified fields in a hash.
+// Accepts a key and the fields themselves.
+// This command ensures that each field specified will have its expiration removed if present.
+// For more information - https://redis.io/commands/hpersist/
+func (c cmdable) HPersist(ctx context.Context, key string, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPERSIST", key, "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HExpireTime - Retrieves the expiration time for specified fields in a hash as a UNIX timestamp in seconds.
+// Requires a key and the fields themselves to fetch their expiration timestamps.
+// This command returns the expiration times for each field or error/status codes for each field as specified.
+// For more information - https://redis.io/commands/hexpiretime/
+func (c cmdable) HExpireTime(ctx context.Context, key string, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HEXPIRETIME", key, "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HPExpireTime - Retrieves the expiration time for specified fields in a hash as a UNIX timestamp in milliseconds.
+// Similar to HExpireTime, adjusted for timestamps in milliseconds. It requires the same parameters.
+// Provides the expiration timestamp for each field in milliseconds.
+// For more information - https://redis.io/commands/hexpiretime/
+func (c cmdable) HPExpireTime(ctx context.Context, key string, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPEXPIRETIME", key, "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HTTL - Retrieves the remaining time to live for specified fields in a hash in seconds.
+// Requires a key and the fields themselves. It returns the TTL for each specified field.
+// This command fetches the TTL in seconds for each field or returns error/status codes as appropriate.
+// For more information - https://redis.io/commands/httl/
+func (c cmdable) HTTL(ctx context.Context, key string, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HTTL", key, "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// HPTTL - Retrieves the remaining time to live for specified fields in a hash in milliseconds.
+// Similar to HTTL, but returns the TTL in milliseconds. It requires a key and the specified fields.
+// This command provides the TTL in milliseconds for each field or returns error/status codes as needed.
+// For more information - https://redis.io/commands/hpttl/
+func (c cmdable) HPTTL(ctx context.Context, key string, fields ...string) *IntSliceCmd {
+	args := []interface{}{"HPTTL", key, "FIELDS", len(fields)}
+
+	for _, field := range fields {
+		args = append(args, field)
+	}
+	cmd := NewIntSliceCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}

vendor/github.com/redis/go-redis/v9/internal/pool/conn_check.go

@@ -3,6 +3,7 @@
 package pool
 
 import (
+	"crypto/tls"
 	"errors"
 	"io"
 	"net"
@@ -16,6 +17,10 @@ func connCheck(conn net.Conn) error {
 	// Reset previous timeout.
 	_ = conn.SetDeadline(time.Time{})
 
+	// Check if tls.Conn.
+	if c, ok := conn.(*tls.Conn); ok {
+		conn = c.NetConn()
+	}
 	sysConn, ok := conn.(syscall.Conn)
 	if !ok {
 		return nil

vendor/github.com/redis/go-redis/v9/internal/util.go

@@ -3,6 +3,7 @@ package internal
 import (
 	"context"
 	"net"
+	"strconv"
 	"strings"
 	"time"
 
@@ -81,3 +82,47 @@ func GetAddr(addr string) string {
 	}
 	return net.JoinHostPort(addr[:ind], addr[ind+1:])
 }
+
+func ToInteger(val interface{}) int {
+	switch v := val.(type) {
+	case int:
+		return v
+	case int64:
+		return int(v)
+	case string:
+		i, _ := strconv.Atoi(v)
+		return i
+	default:
+		return 0
+	}
+}
+
+func ToFloat(val interface{}) float64 {
+	switch v := val.(type) {
+	case float64:
+		return v
+	case string:
+		f, _ := strconv.ParseFloat(v, 64)
+		return f
+	default:
+		return 0.0
+	}
+}
+
+func ToString(val interface{}) string {
+	if str, ok := val.(string); ok {
+		return str
+	}
+	return ""
+}
+
+func ToStringSlice(val interface{}) []string {
+	if arr, ok := val.([]interface{}); ok {
+		result := make([]string, len(arr))
+		for i, v := range arr {
+			result[i] = ToString(v)
+		}
+		return result
+	}
+	return nil
+}

vendor/github.com/redis/go-redis/v9/osscluster.go

@@ -341,6 +341,8 @@ func (n *clusterNode) Close() error {
 	return n.Client.Close()
 }
 
+const maximumNodeLatency = 1 * time.Minute
+
 func (n *clusterNode) updateLatency() {
 	const numProbe = 10
 	var dur uint64
@@ -361,7 +363,7 @@ func (n *clusterNode) updateLatency() {
 	if successes == 0 {
 		// If none of the pings worked, set latency to some arbitrarily high value so this node gets
 		// least priority.
-		latency = float64((1 * time.Minute) / time.Microsecond)
+		latency = float64((maximumNodeLatency) / time.Microsecond)
 	} else {
 		latency = float64(dur) / float64(successes)
 	}
@@ -735,20 +737,40 @@ func (c *clusterState) slotClosestNode(slot int) (*clusterNode, error) {
 		return c.nodes.Random()
 	}
 
-	var node *clusterNode
+	var allNodesFailing = true
+	var (
+		closestNonFailingNode *clusterNode
+		closestNode           *clusterNode
+		minLatency            time.Duration
+	)
+
+	// setting the max possible duration as zerovalue for minlatency
+	minLatency = time.Duration(math.MaxInt64)
+
 	for _, n := range nodes {
-		if n.Failing() {
+		if closestNode == nil || n.Latency() < minLatency {
-			continue
+			closestNode = n
+			minLatency = n.Latency()
+			if !n.Failing() {
+				closestNonFailingNode = n
+				allNodesFailing = false
+			}
 		}
-		if node == nil || n.Latency() < node.Latency() {
-			node = n
-		}
-	}
-	if node != nil {
-		return node, nil
 	}
 
-	// If all nodes are failing - return random node
+	// pick the healthly node with the lowest latency
+	if !allNodesFailing && closestNonFailingNode != nil {
+		return closestNonFailingNode, nil
+	}
+
+	// if all nodes are failing, we will pick the temporarily failing node with lowest latency
+	if minLatency < maximumNodeLatency && closestNode != nil {
+		internal.Logger.Printf(context.TODO(), "redis: all nodes are marked as failed, picking the temporarily failing node with lowest latency")
+		return closestNode, nil
+	}
+
+	// If all nodes are having the maximum latency(all pings are failing) - return a random node across the cluster
+	internal.Logger.Printf(context.TODO(), "redis: pings to all nodes are failing, picking a random node across the cluster")
 	return c.nodes.Random()
 }
 

vendor/github.com/redis/go-redis/v9/pubsub.go

@@ -84,7 +84,7 @@ func (c *PubSub) conn(ctx context.Context, newChannels []string) (*pool.Conn, er
 }
 
 func (c *PubSub) writeCmd(ctx context.Context, cn *pool.Conn, cmd Cmder) error {
-	return cn.WithWriter(context.Background(), c.opt.WriteTimeout, func(wr *proto.Writer) error {
+	return cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmd(wr, cmd)
 	})
 }

vendor/github.com/redis/go-redis/v9/stream_commands.go

@@ -137,10 +137,11 @@ type XReadArgs struct {
 	Streams []string // list of streams and ids, e.g. stream1 stream2 id1 id2
 	Count   int64
 	Block   time.Duration
+	ID      string
 }
 
 func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
-	args := make([]interface{}, 0, 6+len(a.Streams))
+	args := make([]interface{}, 0, 2*len(a.Streams)+6)
 	args = append(args, "xread")
 
 	keyPos := int8(1)
@@ -159,6 +160,11 @@ func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
 	for _, s := range a.Streams {
 		args = append(args, s)
 	}
+	if a.ID != "" {
+		for range a.Streams {
+			args = append(args, a.ID)
+		}
+	}
 
 	cmd := NewXStreamSliceCmd(ctx, args...)
 	if a.Block >= 0 {

vendor/github.com/redis/go-redis/v9/timeseries_commands.go

@@ -40,25 +40,32 @@ type TimeseriesCmdable interface {
 }
 
 type TSOptions struct {
-	Retention       int
+	Retention         int
-	ChunkSize       int
+	ChunkSize         int
-	Encoding        string
+	Encoding          string
-	DuplicatePolicy string
+	DuplicatePolicy   string
-	Labels          map[string]string
+	Labels            map[string]string
+	IgnoreMaxTimeDiff int64
+	IgnoreMaxValDiff  float64
 }
 type TSIncrDecrOptions struct {
-	Timestamp    int64
+	Timestamp         int64
-	Retention    int
+	Retention         int
-	ChunkSize    int
+	ChunkSize         int
-	Uncompressed bool
+	Uncompressed      bool
-	Labels       map[string]string
+	DuplicatePolicy   string
+	Labels            map[string]string
+	IgnoreMaxTimeDiff int64
+	IgnoreMaxValDiff  float64
 }
 
 type TSAlterOptions struct {
-	Retention       int
+	Retention         int
-	ChunkSize       int
+	ChunkSize         int
-	DuplicatePolicy string
+	DuplicatePolicy   string
-	Labels          map[string]string
+	Labels            map[string]string
+	IgnoreMaxTimeDiff int64
+	IgnoreMaxValDiff  float64
 }
 
 type TSCreateRuleOptions struct {
@@ -223,6 +230,9 @@ func (c cmdable) TSAddWithArgs(ctx context.Context, key string, timestamp interf
 				args = append(args, label, value)
 			}
 		}
+		if options.IgnoreMaxTimeDiff != 0 || options.IgnoreMaxValDiff != 0 {
+			args = append(args, "IGNORE", options.IgnoreMaxTimeDiff, options.IgnoreMaxValDiff)
+		}
 	}
 	cmd := NewIntCmd(ctx, args...)
 	_ = c(ctx, cmd)
@@ -264,6 +274,9 @@ func (c cmdable) TSCreateWithArgs(ctx context.Context, key string, options *TSOp
 				args = append(args, label, value)
 			}
 		}
+		if options.IgnoreMaxTimeDiff != 0 || options.IgnoreMaxValDiff != 0 {
+			args = append(args, "IGNORE", options.IgnoreMaxTimeDiff, options.IgnoreMaxValDiff)
+		}
 	}
 	cmd := NewStatusCmd(ctx, args...)
 	_ = c(ctx, cmd)
@@ -292,6 +305,9 @@ func (c cmdable) TSAlter(ctx context.Context, key string, options *TSAlterOption
 				args = append(args, label, value)
 			}
 		}
+		if options.IgnoreMaxTimeDiff != 0 || options.IgnoreMaxValDiff != 0 {
+			args = append(args, "IGNORE", options.IgnoreMaxTimeDiff, options.IgnoreMaxValDiff)
+		}
 	}
 	cmd := NewStatusCmd(ctx, args...)
 	_ = c(ctx, cmd)
@@ -351,12 +367,18 @@ func (c cmdable) TSIncrByWithArgs(ctx context.Context, key string, timestamp flo
 		if options.Uncompressed {
 			args = append(args, "UNCOMPRESSED")
 		}
+		if options.DuplicatePolicy != "" {
+			args = append(args, "DUPLICATE_POLICY", options.DuplicatePolicy)
+		}
 		if options.Labels != nil {
 			args = append(args, "LABELS")
 			for label, value := range options.Labels {
 				args = append(args, label, value)
 			}
 		}
+		if options.IgnoreMaxTimeDiff != 0 || options.IgnoreMaxValDiff != 0 {
+			args = append(args, "IGNORE", options.IgnoreMaxTimeDiff, options.IgnoreMaxValDiff)
+		}
 	}
 	cmd := NewIntCmd(ctx, args...)
 	_ = c(ctx, cmd)
@@ -391,12 +413,18 @@ func (c cmdable) TSDecrByWithArgs(ctx context.Context, key string, timestamp flo
 		if options.Uncompressed {
 			args = append(args, "UNCOMPRESSED")
 		}
+		if options.DuplicatePolicy != "" {
+			args = append(args, "DUPLICATE_POLICY", options.DuplicatePolicy)
+		}
 		if options.Labels != nil {
 			args = append(args, "LABELS")
 			for label, value := range options.Labels {
 				args = append(args, label, value)
 			}
 		}
+		if options.IgnoreMaxTimeDiff != 0 || options.IgnoreMaxValDiff != 0 {
+			args = append(args, "IGNORE", options.IgnoreMaxTimeDiff, options.IgnoreMaxValDiff)
+		}
 	}
 	cmd := NewIntCmd(ctx, args...)
 	_ = c(ctx, cmd)

vendor/modules.txt

@@ -193,7 +193,7 @@ github.com/prometheus/common/model
 github.com/prometheus/procfs
 github.com/prometheus/procfs/internal/fs
 github.com/prometheus/procfs/internal/util
-# github.com/redis/go-redis/v9 v9.5.3
+# github.com/redis/go-redis/v9 v9.5.4
 ## explicit; go 1.18
 github.com/redis/go-redis/v9
 github.com/redis/go-redis/v9/internal