diff --git a/command.go b/command.go index 1688f474..30517b9d 100644 --- a/command.go +++ b/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 { diff --git a/command_nil_element_desync_test.go b/command_nil_element_desync_test.go index 0092788f..26ece94c 100644 --- a/command_nil_element_desync_test.go +++ b/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) + } +} diff --git a/probabilistic.go b/probabilistic.go index 0eeb8d97..635ef111 100644 --- a/probabilistic.go +++ b/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)) diff --git a/timeseries_commands.go b/timeseries_commands.go index ba1372c6..bf125ccb 100644 --- a/timeseries_commands.go +++ b/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"}