fix(multidb): honor NoRetry commands in pipeline retries

Batches containing a non-retryable command (RawWriteToCmd and friends)
execute at most once, and the Watch documentation now states that a
transaction stays bound to the member that was active at call time.

Author: ndyakov

multidb_autopipeline_test.go

@@ -3,6 +3,7 @@ package redis_test
 import (
 	"context"
 	"errors"
+	"io"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -129,6 +130,30 @@ func TestMultiDBPipelinePerCommandRecording(t *testing.T) {
 	}
 }
 
+func TestMultiDBPipelineNoRetryCommandDisablesRetry(t *testing.T) {
+	db1 := newTestDB("db1", "127.0.0.1:1", 2.0, true)
+	db2 := newTestDB("db2", "127.0.0.1:2", 1.0, true)
+
+	opts := baseOptions()
+	opts.CommandRetries = 3 // must be ignored when the batch has a NoRetry command
+	opts.CircuitBreakerConfig = fastBreaker()
+	mdb := newTestMultiDB(t, opts, db1, db2)
+	ctx := context.Background()
+
+	db1.hook.fail.Store(true)
+
+	pipe := mdb.Pipeline()
+	pipe.Get(ctx, "k")
+	_ = pipe.Process(ctx, redis.NewRawWriteToCmd(ctx, io.Discard, "get", "k"))
+	_, err := pipe.Exec(ctx)
+	if err == nil {
+		t.Fatal("batch with a NoRetry command should surface the failure, not retry")
+	}
+	if db2.hook.batches.Load() != 0 {
+		t.Error("batch containing a NoRetry command was retried on db2")
+	}
+}
+
 func TestMultiDBTxPipelineNotRetried(t *testing.T) {
 	db1 := newTestDB("db1", "127.0.0.1:1", 2.0, true)
 	db2 := newTestDB("db2", "127.0.0.1:2", 1.0, true)

multidb_pipeline.go

@@ -51,6 +51,12 @@ func (c *multidbCore) processPipeline(ctx context.Context, cmds []Cmder) error {
 		return nil
 	}
 	attempts := c.opts.CommandRetries + 1
+	// A batch containing a non-retryable command (e.g. a streaming
+	// RawWriteToCmd) must be executed at most once: retrying it after a
+	// transport failure could duplicate execution or corrupt a partial write.
+	if cmdsContainNoRetry(cmds) {
+		attempts = 1
+	}
 
 	for attempt := 0; attempt < attempts; attempt++ {
 		if err := ctx.Err(); err != nil {
@@ -180,9 +186,11 @@ func (c *MultiDBClient) TxPipelined(ctx context.Context, fn func(Pipeliner) erro
 }
 
 // Watch runs a WATCH/MULTI/EXEC transaction on the database that is active
-// when Watch is called. WATCH state is connection-bound: a failover while the
-// transaction is open aborts it with an error, and it is never automatically
-// retried on another database.
+// when Watch is called. The transaction is bound to that member for its whole
+// lifetime: it does NOT follow a MultiDB failover, and it is never
+// automatically retried on another database. If the bound member fails while
+// the transaction is open, the transaction errors like it would on a plain
+// client; MultiDB moves only subsequent operations to the new active member.
 func (c *MultiDBClient) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
 	db, _ := c.core.activeSnapshot()
 	if db == nil {