Merge branch 'master' into ndyakov/token-based-auth

Author: ndyakov

.github/workflows/spellcheck.yml

@@ -8,7 +8,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
       - name: Check Spelling
-        uses: rojopolis/spellcheck-github-actions@0.48.0
+        uses: rojopolis/spellcheck-github-actions@0.49.0
         with:
           config_path: .github/spellcheck-settings.yml
           task_name: Markdown

command.go

@@ -2104,7 +2104,9 @@ type XInfoGroup struct {
 	Pending         int64
 	LastDeliveredID string
 	EntriesRead     int64
-	Lag             int64
+	// Lag represents the number of pending messages in the stream not yet
+	// delivered to this consumer group. Returns -1 when the lag cannot be determined.
+	Lag int64
 }
 
 var _ Cmder = (*XInfoGroupsCmd)(nil)
@@ -2187,8 +2189,11 @@ func (cmd *XInfoGroupsCmd) readReply(rd *proto.Reader) error {
 
 				// lag: the number of entries in the stream that are still waiting to be delivered
 				// to the group's consumers, or a NULL(Nil) when that number can't be determined.
+				// In that case, we return -1.
 				if err != nil && err != Nil {
 					return err
+				} else if err == Nil {
+					group.Lag = -1
 				}
 			default:
 				return fmt.Errorf("redis: unexpected key %q in XINFO GROUPS reply", key)

commands_test.go

@@ -6772,6 +6772,36 @@ var _ = Describe("Commands", func() {
 				}))
 			})
 
+			It("should return -1 for nil lag in XINFO GROUPS", func() {
+				_, err := client.XAdd(ctx, &redis.XAddArgs{Stream: "s", ID: "0-1", Values: []string{"foo", "1"}}).Result()
+				Expect(err).NotTo(HaveOccurred())
+
+				client.XAdd(ctx, &redis.XAddArgs{Stream: "s", ID: "0-2", Values: []string{"foo", "2"}})
+				Expect(err).NotTo(HaveOccurred())
+				client.XAdd(ctx, &redis.XAddArgs{Stream: "s", ID: "0-3", Values: []string{"foo", "3"}})
+				Expect(err).NotTo(HaveOccurred())
+
+				err = client.XGroupCreate(ctx, "s", "g", "0").Err()
+				Expect(err).NotTo(HaveOccurred())
+				err = client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: "g", Consumer: "c", Streams: []string{"s", ">"}, Count: 1, Block: -1, NoAck: false}).Err()
+				Expect(err).NotTo(HaveOccurred())
+
+				client.XDel(ctx, "s", "0-2")
+
+				res, err := client.XInfoGroups(ctx, "s").Result()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(res).To(Equal([]redis.XInfoGroup{
+					{
+						Name:            "g",
+						Consumers:       1,
+						Pending:         1,
+						LastDeliveredID: "0-1",
+						EntriesRead:     1,
+						Lag:             -1, // nil lag from Redis is reported as -1
+					},
+				}))
+			})
+
 			It("should XINFO CONSUMERS", func() {
 				res, err := client.XInfoConsumers(ctx, "stream", "group1").Result()
 				Expect(err).NotTo(HaveOccurred())

options.go

@@ -229,6 +229,9 @@ func (opt *Options) init() {
 			opt.Network = "tcp"
 		}
 	}
+	if opt.Protocol < 2 {
+		opt.Protocol = 3
+	}
 	if opt.DialTimeout == 0 {
 		opt.DialTimeout = 5 * time.Second
 	}

options_test.go

@@ -222,3 +222,26 @@ func TestReadTimeoutOptions(t *testing.T) {
 		}
 	}
 }
+
+func TestProtocolOptions(t *testing.T) {
+	testCasesMap := map[int]int{
+		0: 3,
+		1: 3,
+		2: 2,
+		3: 3,
+	}
+
+	o := &Options{}
+	o.init()
+	if o.Protocol != 3 {
+		t.Errorf("got %d instead of %d as protocol option", o.Protocol, 3)
+	}
+
+	for set, want := range testCasesMap {
+		o := &Options{Protocol: set}
+		o.init()
+		if o.Protocol != want {
+			t.Errorf("got %d instead of %d as protocol option", o.Protocol, want)
+		}
+	}
+}

osscluster.go

@@ -448,6 +448,14 @@ func (n *clusterNode) SetLastLatencyMeasurement(t time.Time) {
 	}
 }
 
+func (n *clusterNode) Loading() bool {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	err := n.Client.Ping(ctx).Err()
+	return err != nil && isLoadingError(err)
+}
+
 //------------------------------------------------------------------------------
 
 type clusterNodes struct {
@@ -757,7 +765,8 @@ func (c *clusterState) slotSlaveNode(slot int) (*clusterNode, error) {
 	case 1:
 		return nodes[0], nil
 	case 2:
-		if slave := nodes[1]; !slave.Failing() {
+		slave := nodes[1]
+		if !slave.Failing() && !slave.Loading() {
 			return slave, nil
 		}
 		return nodes[0], nil
@@ -766,7 +775,7 @@ func (c *clusterState) slotSlaveNode(slot int) (*clusterNode, error) {
 		for i := 0; i < 10; i++ {
 			n := rand.Intn(len(nodes)-1) + 1
 			slave = nodes[n]
-			if !slave.Failing() {
+			if !slave.Failing() && !slave.Loading() {
 				return slave, nil
 			}
 		}
@@ -1501,7 +1510,7 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err
 		return err
 	}
 
-	cmdsMap := c.mapCmdsBySlot(ctx, cmds)
+	cmdsMap := c.mapCmdsBySlot(cmds)
 	for slot, cmds := range cmdsMap {
 		node, err := state.slotMasterNode(slot)
 		if err != nil {
@@ -1540,7 +1549,7 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err
 	return cmdsFirstErr(cmds)
 }
 
-func (c *ClusterClient) mapCmdsBySlot(ctx context.Context, cmds []Cmder) map[int][]Cmder {
+func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
 	cmdsMap := make(map[int][]Cmder)
 	for _, cmd := range cmds {
 		slot := c.cmdSlot(cmd)

redis.go

@@ -360,12 +360,6 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 	connPool := pool.NewSingleConnPool(c.connPool, cn)
 	conn := newConn(c.opt, connPool, &c.hooksMixin)
 
-	protocol := c.opt.Protocol
-	// By default, use RESP3 in current version.
-	if protocol < 2 {
-		protocol = 3
-	}
-
 	username, password := "", ""
 	if c.opt.StreamingCredentialsProvider != nil {
 		credentials, unsubscribeFromCredentialsProvider, err := c.opt.StreamingCredentialsProvider.
@@ -389,7 +383,7 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 
 	// for redis-server versions that do not support the HELLO command,
 	// RESP2 will continue to be used.
-	if err = conn.Hello(ctx, protocol, username, password, c.opt.ClientName).Err(); err == nil {
+  if err = conn.Hello(ctx, c.opt.Protocol, username, password, c.opt.ClientName).Err(); err == nil {
 		// Authentication successful with HELLO command
 	} else if !isRedisError(err) {
 		// When the server responds with the RESP protocol and the result is not a normal

ring.go

@@ -702,7 +702,7 @@ func (c *Ring) cmdsInfo(ctx context.Context) (map[string]*CommandInfo, error) {
 	return nil, firstErr
 }
 
-func (c *Ring) cmdShard(ctx context.Context, cmd Cmder) (*ringShard, error) {
+func (c *Ring) cmdShard(cmd Cmder) (*ringShard, error) {
 	pos := cmdFirstKeyPos(cmd)
 	if pos == 0 {
 		return c.sharding.Random()
@@ -720,7 +720,7 @@ func (c *Ring) process(ctx context.Context, cmd Cmder) error {
 			}
 		}
 
-		shard, err := c.cmdShard(ctx, cmd)
+		shard, err := c.cmdShard(cmd)
 		if err != nil {
 			return err
 		}