fix: don't send CLIENT CACHING YES in BCAST and OPTOUT mode (#801)

Signed-off-by: Rueian <[email protected]>

Author: rueian

client_test.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/redis/rueidis/internal/cmds"
 )
 
 type mockConn struct {
@@ -191,6 +193,10 @@ func (m *mockConn) Addr() string {
 	return ""
 }
 
+func (m *mockConn) OptInCmd() cmds.Completed {
+	return cmds.OptInCmd
+}
+
 func TestNewSingleClientNoNode(t *testing.T) {
 	defer ShouldNotLeaked(SetupLeakDetection())
 	if _, err := newSingleClient(

cluster.go

@@ -899,7 +899,7 @@ func askingMultiCache(cc conn, ctx context.Context, multi []CacheableTTL) *redis
 	commands := make([]Completed, 0, len(multi)*6)
 	for _, cmd := range multi {
 		ck, _ := cmds.CacheKey(cmd.Cmd)
-		commands = append(commands, cmds.OptInCmd, cmds.AskingCmd, cmds.MultiCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(cmd.Cmd), cmds.ExecCmd)
+		commands = append(commands, cc.OptInCmd(), cmds.AskingCmd, cmds.MultiCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(cmd.Cmd), cmds.ExecCmd)
 	}
 	results := resultsp.Get(0, len(multi))
 	resps := cc.DoMulti(ctx, commands...)

internal/cmds/cmds.go

@@ -7,10 +7,10 @@ const (
 	blockTag = uint16(1 << 14)
 	readonly = uint16(1 << 13)
 	noRetTag = uint16(1<<12) | readonly | pipeTag // make noRetTag can also be retried and auto pipelining
-	mtGetTag = uint16(1<<11) | readonly // make mtGetTag can also be retried
-	scrRoTag = uint16(1<<10) | readonly // make scrRoTag can also be retried
+	mtGetTag = uint16(1<<11) | readonly           // make mtGetTag can also be retried
+	scrRoTag = uint16(1<<10) | readonly           // make scrRoTag can also be retried
 	unsubTag = uint16(1<<9) | noRetTag
-	pipeTag  = uint16(1<<8) // make blocking mode request can use auto pipelining
+	pipeTag  = uint16(1 << 8) // make blocking mode request can use auto pipelining
 	// InitSlot indicates that the command be sent to any redis node in cluster
 	InitSlot = uint16(1 << 14)
 	// NoSlot indicates that the command has no key slot specified
@@ -23,6 +23,11 @@ var (
 		cs: newCommandSlice([]string{"CLIENT", "CACHING", "YES"}),
 		cf: optInTag,
 	}
+	// OptInNopCmd is a predefined alternative for CLIENT CACHING YES in BCAST/OPTOUT mode.
+	OptInNopCmd = Completed{
+		cs: newCommandSlice([]string{"ECHO", ""}),
+		cf: optInTag,
+	}
 	// MultiCmd is predefined MULTI
 	MultiCmd = Completed{
 		cs: newCommandSlice([]string{"MULTI"}),

internal/cmds/cmds_test.go

@@ -90,6 +90,9 @@ func TestCompleted_IsOptIn(t *testing.T) {
 	if cmd := OptInCmd; !cmd.IsOptIn() {
 		t.Fatalf("should be opt in command")
 	}
+	if cmd := OptInNopCmd; !cmd.IsOptIn() {
+		t.Fatalf("should be opt in command")
+	}
 }
 
 func TestCompleted_NoReply(t *testing.T) {

mux.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net"
 	"runtime"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -41,6 +42,7 @@ type conn interface {
 	Store(w wire)
 	Addr() string
 	SetOnCloseHook(func(error))
+	OptInCmd() cmds.Completed
 }
 
 var _ conn = (*mux)(nil)
@@ -60,6 +62,7 @@ type mux struct {
 	maxm   int
 
 	usePool bool
+	optin   bool
 }
 
 func makeMux(dst string, option *ClientOption, dialFn dialFn) *mux {
@@ -95,6 +98,7 @@ func newMux(dst string, option *ClientOption, init, dead wire, wireFn wireFn, wi
 		maxm: option.BlockingPipeline,
 
 		usePool: option.DisableAutoPipelining,
+		optin:   isOptIn(option.ClientTrackingOptions),
 	}
 	m.clhks.Store(emptyclhks)
 	for i := 0; i < len(m.wire); i++ {
@@ -106,6 +110,22 @@ func newMux(dst string, option *ClientOption, init, dead wire, wireFn wireFn, wi
 	return m
 }
 
+func isOptIn(opts []string) bool {
+	for _, opt := range opts {
+		if opt := strings.ToUpper(opt); opt == "BCAST" || opt == "OPTOUT" {
+			return false
+		}
+	}
+	return true
+}
+
+func (m *mux) OptInCmd() cmds.Completed {
+	if m.optin {
+		return cmds.OptInCmd
+	}
+	return cmds.OptInNopCmd
+}
+
 func (m *mux) SetOnCloseHook(fn func(error)) {
 	m.clhks.Store(fn)
 }

mux_test.go

@@ -126,6 +126,26 @@ func TestMuxAddr(t *testing.T) {
 	}
 }
 
+func TestMuxOptInCmd(t *testing.T) {
+	defer ShouldNotLeaked(SetupLeakDetection())
+
+	if m := makeMux("dst1", &ClientOption{
+		ClientTrackingOptions: []string{"OPTOUT"},
+	}, nil); m.OptInCmd() != cmds.OptInNopCmd {
+		t.Fatalf("unexpected OptInCmd")
+	}
+	if m := makeMux("dst1", &ClientOption{
+		ClientTrackingOptions: []string{"PREFIX", "a", "BCAST"},
+	}, nil); m.OptInCmd() != cmds.OptInNopCmd {
+		t.Fatalf("unexpected OptInCmd")
+	}
+	if m := makeMux("dst1", &ClientOption{
+		ClientTrackingOptions: nil,
+	}, nil); m.OptInCmd() != cmds.OptInCmd {
+		t.Fatalf("unexpected OptInCmd")
+	}
+}
+
 func TestMuxDialSuppress(t *testing.T) {
 	defer ShouldNotLeaked(SetupLeakDetection())
 	var wires, waits, done int64

pipe.go

@@ -89,6 +89,7 @@ type pipe struct {
 	recvs           int32
 	r2ps            bool // identify this pipe is used for resp2 pubsub or not
 	noNoDelay       bool
+	optin           bool
 }
 
 type pipeFn func(connFn func() (net.Conn, error), option *ClientOption) (p *pipe, err error)
@@ -116,7 +117,8 @@ func _newPipe(connFn func() (net.Conn, error), option *ClientOption, r2ps, nobg
 		maxFlushDelay: option.MaxFlushDelay,
 		noNoDelay:     option.DisableTCPNoDelay,
 
-		r2ps: r2ps,
+		r2ps:  r2ps,
+		optin: isOptIn(option.ClientTrackingOptions),
 	}
 	if !nobg {
 		p.queue = newRing(option.RingScaleEachConn)
@@ -1300,6 +1302,13 @@ next:
 	return m, nil
 }
 
+func (p *pipe) optInCmd() cmds.Completed {
+	if p.optin {
+		return cmds.OptInCmd
+	}
+	return cmds.OptInNopCmd
+}
+
 func (p *pipe) DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) RedisResult {
 	if p.cache == nil {
 		return p.Do(ctx, Completed(cmd))
@@ -1319,7 +1328,7 @@ func (p *pipe) DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) Re
 	}
 	resp := p.DoMulti(
 		ctx,
-		cmds.OptInCmd,
+		p.optInCmd(),
 		cmds.MultiCmd,
 		cmds.NewCompleted([]string{"PTTL", ck}),
 		Completed(cmd),
@@ -1387,7 +1396,7 @@ func (p *pipe) doCacheMGet(ctx context.Context, cmd Cacheable, ttl time.Duration
 		}
 
 		multi := make([]Completed, 0, keys+4)
-		multi = append(multi, cmds.OptInCmd, cmds.MultiCmd)
+		multi = append(multi, p.optInCmd(), cmds.MultiCmd)
 		for _, key := range rewritten.Commands()[1 : keys+1] {
 			multi = append(multi, builder.Pttl().Key(key).Build())
 		}
@@ -1473,7 +1482,7 @@ func (p *pipe) DoMultiCache(ctx context.Context, multi ...CacheableTTL) *redisre
 		for _, i := range missed {
 			ct := multi[i]
 			ck, _ := cmds.CacheKey(ct.Cmd)
-			missing = append(missing, cmds.OptInCmd, cmds.MultiCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(ct.Cmd), cmds.ExecCmd)
+			missing = append(missing, p.optInCmd(), cmds.MultiCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(ct.Cmd), cmds.ExecCmd)
 		}
 	} else {
 		for i, ct := range multi {
@@ -1487,7 +1496,7 @@ func (p *pipe) DoMultiCache(ctx context.Context, multi ...CacheableTTL) *redisre
 				entries.e[i] = entry // store entries for later entry.Wait() to avoid MGET deadlock each others.
 				continue
 			}
-			missing = append(missing, cmds.OptInCmd, cmds.MultiCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(ct.Cmd), cmds.ExecCmd)
+			missing = append(missing, p.optInCmd(), cmds.MultiCmd, cmds.NewCompleted([]string{"PTTL", ck}), Completed(ct.Cmd), cmds.ExecCmd)
 		}
 	}