fix(command): read past error elements in IntSliceCmd (#4071)

* fix(command): read past error elements in IntSliceCmd

TS.MADD and CMS.INCRBY reply with an error element for a rejected item
and keep answering for the items after it. Returning on that element
left the tail of the array unread, and in a pipeline the next command
read it as its own reply.

* docs(command): note per-item errors on TSMAdd/CMSIncrBy

Also move the error-element tests next to the nil-element cases.

---------

Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com>

command.go

@@ -1581,17 +1581,27 @@ func (cmd *IntSliceCmd) readReply(rd *proto.Reader) error {
 		return err
 	}
 	cmd.val = make([]int64, n)
+	var elemErr error
 	for i := 0; i < len(cmd.val); i++ {
 		switch num, err := rd.ReadInt(); {
 		case err == Nil:
 			cmd.val[i] = 0
 		case err != nil:
-			return err
+			if !isRedisError(err) {
+				return err
+			}
+			// TS.MADD and CMS.INCRBY report a rejected item as an error
+			// element and still answer for the items after it. The element
+			// is a complete frame, so read on and return the first one once
+			// the whole array is consumed.
+			if elemErr == nil {
+				elemErr = err
+			}
 		default:
 			cmd.val[i] = num
 		}
 	}
-	return nil
+	return elemErr
 }
 
 func (cmd *IntSliceCmd) Clone() Cmder {

command_nil_element_desync_test.go

@@ -2,6 +2,7 @@ package redis
 
 import (
 	"context"
+	"slices"
 	"strings"
 	"testing"
 
@@ -65,3 +66,76 @@ func assertNextReplyInt(t *testing.T, rd *proto.Reader, want int64) {
 		t.Fatalf("connection desynced: next reply = %d, want %d", got, want)
 	}
 }
+
+// TS.MADD and CMS.INCRBY answer item by item: a rejected item becomes an error
+// element and the items after it are still reported. IntSliceCmd used to
+// return on the error element with the rest of the array unread. Callers take
+// a Redis error from readReply as a fully read reply, so in a pipeline the
+// next command was handed the leftover element and every later reply shifted
+// by one. The first two replies are the ones Redis 8.10 sends.
+func TestIntSliceCmdErrorElementNoDesync(t *testing.T) {
+	tests := []struct {
+		name    string
+		reply   string
+		wantVal []int64
+		wantErr string
+	}{
+		{
+			name:    "TS.MADD sample older than retention",
+			reply:   "*3\r\n:10000\r\n-ERR TSDB: Timestamp is older than retention\r\n:10001\r\n",
+			wantVal: []int64{10000, 0, 10001},
+			wantErr: "ERR TSDB: Timestamp is older than retention",
+		},
+		{
+			name:    "CMS.INCRBY counter overflow",
+			reply:   "*4\r\n:4294967294\r\n:1\r\n-CMS: INCRBY overflow\r\n:2\r\n",
+			wantVal: []int64{4294967294, 1, 0, 2},
+			wantErr: "CMS: INCRBY overflow",
+		},
+		{
+			name:    "first of several errors is reported",
+			reply:   "*4\r\n-ERR first\r\n:7\r\n-ERR second\r\n:8\r\n",
+			wantVal: []int64{0, 7, 0, 8},
+			wantErr: "ERR first",
+		},
+		{
+			name:    "RESP3 blob error element",
+			reply:   "*3\r\n:1\r\n!9\r\nERR boom!\r\n:3\r\n",
+			wantVal: []int64{1, 0, 3},
+			wantErr: "ERR boom!",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// The reply is followed by the next command's reply, :999.
+			rd := proto.NewReader(strings.NewReader(tt.reply + ":999\r\n"))
+			cmd := NewIntSliceCmd(context.Background())
+			err := cmd.readReply(rd)
+			if err == nil || err.Error() != tt.wantErr {
+				t.Fatalf("readReply err = %v, want %q", err, tt.wantErr)
+			}
+			if !isRedisError(err) {
+				t.Fatalf("readReply err = %T, want a Redis error", err)
+			}
+			if !slices.Equal(cmd.val, tt.wantVal) {
+				t.Fatalf("unexpected val: %+v, want %+v", cmd.val, tt.wantVal)
+			}
+			assertNextReplyInt(t, rd, 999)
+		})
+	}
+}
+
+// A read error in the middle of the array is not an error element: the reply
+// is incomplete and readReply must fail with it so the connection is dropped.
+func TestIntSliceCmdErrorElementTruncatedReply(t *testing.T) {
+	rd := proto.NewReader(strings.NewReader("*3\r\n:10000\r\n-ERR TSDB: Timestamp is older than retention\r\n"))
+	cmd := NewIntSliceCmd(context.Background())
+	err := cmd.readReply(rd)
+	if err == nil {
+		t.Fatal("readReply returned nil for a truncated reply")
+	}
+	if isRedisError(err) {
+		t.Fatalf("readReply err = %v, want the read error, not the element error", err)
+	}
+}

probabilistic.go

@@ -806,6 +806,8 @@ func (c cmdable) CFMExists(ctx context.Context, key string, elements ...interfac
 
 // CMSIncrBy increments the count of one or more items in a Count-Min Sketch filter.
 // Returns an array of integers representing the updated count of each item.
+// On a per-item error, Err() returns the first error and the failed items are 0
+// in Val(); use Do() to get per-item errors.
 // For more information - https://redis.io/commands/cms.incrby/
 func (c cmdable) CMSIncrBy(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd {
 	args := make([]interface{}, 2, 2+len(elements))

timeseries_commands.go

@@ -706,6 +706,8 @@ func (c cmdable) TSInfoWithArgs(ctx context.Context, key string, options *TSInfo
 // TSMAdd - Adds multiple samples to multiple time-series keys.
 // It accepts a slice of 'ktv' slices, each containing exactly three elements: key, timestamp, and value.
 // This struct must be provided for this command to work.
+// On a per-item error, Err() returns the first error and the failed items are 0
+// in Val(); use Do() to get per-item errors.
 // For more information - https://redis.io/commands/ts.madd/
 func (c cmdable) TSMAdd(ctx context.Context, ktvSlices [][]interface{}) *IntSliceCmd {
 	args := []interface{}{"TS.MADD"}