chore(refactor): move per-connection CLIENT commands to statefulCmdable

.github/workflows/govulncheck.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v7
         with:
-          go-version: "1.26.x"
+          go-version: "stable"
           cache: true
 
       - name: Install govulncheck

autopipeline.go

@@ -1064,6 +1064,57 @@ func (ap *AutoPipeliner) HImportDiscardAll(ctx context.Context) *IntCmd {
 	return ap.pipeliner.HImportDiscardAll(ctx)
 }
 
+// ClientTracking and friends are per-connection commands (statefulCmdable);
+// through the autopipeliner they fail with guidance exactly as on the
+// underlying pooled client — a batch executes on an arbitrary pool
+// connection. Use a dedicated connection (Client.Conn) or the built-in
+// client-side cache. On a closed autopipeliner they return ErrClosed, like
+// every other dispatch path.
+func (ap *AutoPipeliner) ClientTracking(ctx context.Context, on bool, opt *ClientTrackingOptions) *StatusCmd {
+	if !on {
+		return ap.ClientTrackingOff(ctx)
+	}
+	return ap.ClientTrackingOn(ctx, opt)
+}
+
+// ClientTrackingOn through the autopipeliner fails with guidance; see ClientTracking.
+func (ap *AutoPipeliner) ClientTrackingOn(ctx context.Context, opt *ClientTrackingOptions) *StatusCmd {
+	args := []interface{}{"client", "tracking", "on"}
+	if opt != nil {
+		args = appendClientTrackingOptions(args, opt)
+	}
+	if ap.isClosed() {
+		return pooledConnStateCmd(ctx, ErrClosed, args...)
+	}
+	return pooledConnStateCmd(ctx, errClientTrackingOnPooledClient, args...)
+}
+
+// ClientTrackingOff through the autopipeliner fails with guidance; see ClientTracking.
+func (ap *AutoPipeliner) ClientTrackingOff(ctx context.Context) *StatusCmd {
+	if ap.isClosed() {
+		return pooledConnStateCmd(ctx, ErrClosed, "client", "tracking", "off")
+	}
+	return pooledConnStateCmd(ctx, errClientTrackingOnPooledClient, "client", "tracking", "off")
+}
+
+// ClientMaintNotifications through the autopipeliner fails with guidance;
+// set the MaintNotificationsConfig option instead.
+func (ap *AutoPipeliner) ClientMaintNotifications(ctx context.Context, enabled bool, endpointType string) *StatusCmd {
+	args := []interface{}{"client", "maint_notifications"}
+	if enabled {
+		if endpointType == "" {
+			endpointType = "none"
+		}
+		args = append(args, "on", "moving-endpoint-type", endpointType)
+	} else {
+		args = append(args, "off")
+	}
+	if ap.isClosed() {
+		return pooledConnStateCmd(ctx, ErrClosed, args...)
+	}
+	return pooledConnStateCmd(ctx, errClientMaintNotificationsOnPooledClient, args...)
+}
+
 // Watch runs a transactional function on the underlying client (not batched).
 func (ap *AutoPipeliner) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
 	return ap.pipeliner.Watch(ctx, fn, keys...)

client_tracking_pooled_test.go

@@ -0,0 +1,168 @@
+package redis
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+)
+
+// TestPooledClientTrackingPreservesArgs verifies the pooled-client CLIENT
+// TRACKING / MAINT_NOTIFICATIONS wrappers build the full argument list (mirroring
+// the stateful command) before failing with guidance, instead of dropping the
+// caller's options (#3961). The command is pre-failed without dispatch, so no
+// server is needed.
+func TestPooledClientTrackingPreservesArgs(t *testing.T) {
+	ctx := context.Background()
+	c := NewClient(&Options{Addr: ":6379"})
+	defer c.Close()
+
+	opt := &ClientTrackingOptions{Redirect: 42, Bcast: true, Prefixes: []string{"foo"}, NoLoop: true}
+
+	cases := []struct {
+		name string
+		cmd  *StatusCmd
+		err  error
+		want []interface{}
+	}{
+		{
+			name: "ClientTrackingOn",
+			cmd:  c.ClientTrackingOn(ctx, opt),
+			err:  errClientTrackingOnPooledClient,
+			want: []interface{}{"client", "tracking", "on", "redirect", int64(42), "bcast", "prefix", "foo", "noloop"},
+		},
+		{
+			name: "ClientTracking(on)",
+			cmd:  c.ClientTracking(ctx, true, opt),
+			err:  errClientTrackingOnPooledClient,
+			want: []interface{}{"client", "tracking", "on", "redirect", int64(42), "bcast", "prefix", "foo", "noloop"},
+		},
+		{
+			name: "ClientMaintNotifications(on)",
+			cmd:  c.ClientMaintNotifications(ctx, true, "external"),
+			err:  errClientMaintNotificationsOnPooledClient,
+			want: []interface{}{"client", "maint_notifications", "on", "moving-endpoint-type", "external"},
+		},
+		{
+			name: "ClientMaintNotifications(off)",
+			cmd:  c.ClientMaintNotifications(ctx, false, ""),
+			err:  errClientMaintNotificationsOnPooledClient,
+			want: []interface{}{"client", "maint_notifications", "off"},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if !errors.Is(tc.cmd.Err(), tc.err) {
+				t.Fatalf("Err() = %v, want %v", tc.cmd.Err(), tc.err)
+			}
+			if got := tc.cmd.Args(); !reflect.DeepEqual(got, tc.want) {
+				t.Fatalf("Args() = %v, want %v (options dropped)", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestPooledClientMaintNotificationsDefaultEndpoint verifies the empty endpoint
+// type defaults to "none", matching the stateful command.
+func TestPooledClientMaintNotificationsDefaultEndpoint(t *testing.T) {
+	c := NewClient(&Options{Addr: ":6379"})
+	defer c.Close()
+	cmd := c.ClientMaintNotifications(context.Background(), true, "")
+	want := []interface{}{"client", "maint_notifications", "on", "moving-endpoint-type", "none"}
+	if got := cmd.Args(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("Args() = %v, want %v", got, want)
+	}
+}
+
+// TestPipelineRejectsStateCommandsWhenPooled verifies per-connection state
+// commands (CLIENT TRACKING / MAINT_NOTIFICATIONS) are rejected in a POOLED
+// pipeline — whose borrowed connection returns to the pool after Exec — but
+// allowed (queued) in a pipeline from a dedicated *Conn (#3961). No server needed:
+// rejections are pre-failed, and the Conn pipeline only queues.
+func TestPipelineRejectsStateCommandsWhenPooled(t *testing.T) {
+	ctx := context.Background()
+	// localhost:1 is never dialed: the pooled-pipeline guard in
+	// generalProcessPipeline fires before any connection is acquired.
+	c := NewClient(&Options{Addr: "localhost:1"})
+	defer c.Close()
+
+	// Pooled pipeline: the returned command carries the guidance error AND the
+	// command is queued, so Exec surfaces the rejection even when the caller
+	// ignores the returned command. The command must never be sent to a borrowed
+	// pooled connection.
+	pp := c.Pipeline()
+	if cmd := pp.ClientTrackingOn(ctx, &ClientTrackingOptions{Bcast: true}); !errors.Is(cmd.Err(), errClientTrackingOnPooledClient) {
+		t.Fatalf("pooled Pipeline ClientTrackingOn err = %v, want errClientTrackingOnPooledClient", cmd.Err())
+	}
+	if cmd := pp.ClientTrackingOff(ctx); !errors.Is(cmd.Err(), errClientTrackingOnPooledClient) {
+		t.Fatalf("pooled Pipeline ClientTrackingOff err = %v, want reject", cmd.Err())
+	}
+	if cmd := pp.ClientMaintNotifications(ctx, true, "none"); !errors.Is(cmd.Err(), errClientMaintNotificationsOnPooledClient) {
+		t.Fatalf("pooled Pipeline ClientMaintNotifications err = %v, want reject", cmd.Err())
+	}
+	if pp.Len() != 3 {
+		t.Fatalf("pooled Pipeline queued %d state commands, want 3 (queued so Exec surfaces the rejection)", pp.Len())
+	}
+	// Exec surfaces the rejection through the guard, before any dial (this would
+	// otherwise fail dialing localhost:1 with a different error).
+	if _, err := pp.Exec(ctx); !errors.Is(err, errClientTrackingOnPooledClient) {
+		t.Fatalf("pooled Pipeline Exec err = %v, want errClientTrackingOnPooledClient (guarded before dial)", err)
+	}
+
+	// The common Pipelined pattern — callback ignores the returned command — must
+	// still fail rather than silently reporting success (the dropped-error bug).
+	if _, err := c.Pipelined(ctx, func(pipe Pipeliner) error {
+		pipe.ClientTrackingOff(ctx)
+		return nil
+	}); !errors.Is(err, errClientTrackingOnPooledClient) {
+		t.Fatalf("Pipelined ClientTrackingOff err = %v, want errClientTrackingOnPooledClient", err)
+	}
+
+	// Discard drops queued rejections like any other queued command.
+	pp2 := c.Pipeline()
+	pp2.ClientTrackingOff(ctx)
+	pp2.Discard()
+	if _, err := pp2.Exec(ctx); err != nil {
+		t.Fatalf("after Discard, Exec err = %v, want nil (empty pipeline)", err)
+	}
+
+	// Dedicated-Conn pipeline: sticky → allowed → queued (state stays on the conn).
+	conn := c.Conn()
+	defer conn.Close()
+	cp := conn.Pipeline()
+	if cmd := cp.ClientTrackingOn(ctx, &ClientTrackingOptions{Bcast: true}); errors.Is(cmd.Err(), errClientTrackingOnPooledClient) {
+		t.Fatalf("Conn Pipeline wrongly rejected ClientTrackingOn: %v", cmd.Err())
+	}
+	if cp.Len() != 1 {
+		t.Fatalf("Conn Pipeline queued %d, want 1 (state command should be queued on a dedicated conn)", cp.Len())
+	}
+}
+
+// TestTxPipelineAllowsStateCommands pins that a Tx pipeline is sticky: WATCH
+// pins one connection for the whole Tx, so CLIENT TRACKING queues there instead
+// of being rejected as pooled (#3961 regression flagged by review). Needs a
+// server because Watch dials.
+func TestTxPipelineAllowsStateCommands(t *testing.T) {
+	ctx := context.Background()
+	c := NewClient(&Options{Addr: ":6379"})
+	defer c.Close()
+	if err := c.Ping(ctx).Err(); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+
+	err := c.Watch(ctx, func(tx *Tx) error {
+		for _, p := range []Pipeliner{tx.Pipeline(), tx.TxPipeline()} {
+			cmd := p.ClientTrackingOn(ctx, nil)
+			if errors.Is(cmd.Err(), errClientTrackingOnPooledClient) {
+				t.Fatalf("Tx pipeline wrongly rejected CLIENT TRACKING (should be sticky): %v", cmd.Err())
+			}
+			if p.Len() != 1 {
+				t.Fatalf("Tx pipeline queued %d, want 1 (state command allowed on the pinned conn)", p.Len())
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+}

command.go

@@ -235,6 +235,11 @@ type Cmder interface {
 	SetErr(error)
 	Err() error
 
+	// markStateRejected/isStateRejected flag a per-connection state command a
+	// pooled pipeline queued only to surface a rejection through Exec.
+	markStateRejected()
+	isStateRejected() bool
+
 	// setReady marks a command as asynchronously pending (autopipeline async
 	// faces); await blocks the public accessors until it has executed; rawErr
 	// reads the error without awaiting (internal execution path).
@@ -374,6 +379,11 @@ type baseCmd struct {
 	rawVal       interface{}
 	_readTimeout *time.Duration
 	cmdType      CmdType
+	// stateRejected marks a per-connection state command that a pooled
+	// *Pipeline queued only to surface a rejection through Exec (see
+	// markStateRejected). generalProcessPipeline returns its pre-set error
+	// without sending it.
+	stateRejected bool
 	// slotCache memoizes the cluster slot once computed, so the cluster
 	// autopipeline shard router and the pipeline flush router don't each
 	// recompute it. 0 = not computed; it stores slot+1 so a real slot of 0 is
@@ -550,6 +560,20 @@ func (cmd *baseCmd) Err() error {
 	return cmd.err
 }
 
+// markStateRejected / isStateRejected carry the "queued only to be rejected"
+// signal from a pooled *Pipeline to generalProcessPipeline. A pooled pipeline
+// queues a per-connection state command (CLIENT TRACKING / MAINT_NOTIFICATIONS)
+// carrying a pre-set guidance error and this flag, so Exec surfaces the error
+// instead of the command being sent to an arbitrary pooled connection. A sticky
+// (dedicated-connection) pipeline never sets it, so those commands run normally.
+func (cmd *baseCmd) markStateRejected() {
+	cmd.stateRejected = true
+}
+
+func (cmd *baseCmd) isStateRejected() bool {
+	return cmd.stateRejected
+}
+
 func (cmd *baseCmd) readTimeout() *time.Duration {
 	return cmd._readTimeout
 }

commands.go

@@ -554,7 +554,15 @@ func (c cmdable) ClientInfo(ctx context.Context) *ClientInfoCmd {
 
 // ClientMaintNotifications enables or disables maintenance notifications for maintenance upgrades.
 // When enabled, the client will receive push notifications about Redis maintenance events.
-func (c cmdable) ClientMaintNotifications(ctx context.Context, enabled bool, endpointType string) *StatusCmd {
+//
+// CLIENT MAINT_NOTIFICATIONS is per-connection state, exactly like CLIENT
+// TRACKING: it lives on statefulCmdable and is directly usable on a dedicated
+// connection (Client.Conn) and inside Pipeline/Tx. The library enables it on
+// every connection automatically during connection init when
+// Options.MaintNotificationsConfig is set — that is the supported way to turn
+// it on for a whole client. The pooled clients keep a same-named method for
+// interface compatibility that returns an explanatory error.
+func (c statefulCmdable) ClientMaintNotifications(ctx context.Context, enabled bool, endpointType string) *StatusCmd {
 	args := []interface{}{"client", "maint_notifications"}
 	if enabled {
 		if endpointType == "" {
@@ -591,7 +599,16 @@ type ClientTrackingOptions struct {
 // configured with Options.ClientSideCache or ClientSideCacheConfig this
 // command is rejected, because changing a pool connection's tracking state
 // would silently break the cache's invalidation.
-func (c cmdable) ClientTracking(ctx context.Context, on bool, opt *ClientTrackingOptions) *StatusCmd {
+//
+// CLIENT TRACKING is per-connection state, so this lives on statefulCmdable
+// next to ClientSetName/ClientSetInfo: it is directly usable on a dedicated
+// connection (Client.Conn) and inside Pipeline/Tx. The pooled clients keep
+// same-named methods for interface compatibility, but those return an error
+// explaining that the command must target a specific connection — on a pooled
+// client it would land on an arbitrary connection that is immediately
+// returned to the pool, so subsequent reads run on other connections and the
+// invalidation pushes arrive on a connection nothing is reading.
+func (c statefulCmdable) ClientTracking(ctx context.Context, on bool, opt *ClientTrackingOptions) *StatusCmd {
 	if !on {
 		return c.ClientTrackingOff(ctx)
 	}
@@ -600,7 +617,7 @@ func (c cmdable) ClientTracking(ctx context.Context, on bool, opt *ClientTrackin
 
 // ClientTrackingOn enables tracking on the serving connection. See
 // ClientTracking for the pooled-client and built-in-CSC caveats.
-func (c cmdable) ClientTrackingOn(ctx context.Context, opt *ClientTrackingOptions) *StatusCmd {
+func (c statefulCmdable) ClientTrackingOn(ctx context.Context, opt *ClientTrackingOptions) *StatusCmd {
 	args := []interface{}{"client", "tracking", "on"}
 	if opt != nil {
 		if err := validateClientTrackingOptions(opt); err != nil {
@@ -617,7 +634,7 @@ func (c cmdable) ClientTrackingOn(ctx context.Context, opt *ClientTrackingOption
 
 // ClientTrackingOff disables tracking on the serving connection. See
 // ClientTracking for the pooled-client and built-in-CSC caveats.
-func (c cmdable) ClientTrackingOff(ctx context.Context) *StatusCmd {
+func (c statefulCmdable) ClientTrackingOff(ctx context.Context) *StatusCmd {
 	cmd := NewStatusCmd(ctx, "client", "tracking", "off")
 	_ = c(ctx, cmd)
 	return cmd

commands_test.go

@@ -453,24 +453,57 @@ var _ = Describe("Commands", func() {
 		})
 
 		It("should reject invalid ClientTracking option combinations", func() {
-			err := client.ClientTrackingOn(ctx, &redis.ClientTrackingOptions{
+			// Option validation lives on the per-connection (statefulCmdable)
+			// variant: the pooled-client methods reject the command outright
+			// (per-connection state), so validation is only reachable on a
+			// dedicated connection or in a pipeline.
+			conn := rawClient.Conn()
+			defer conn.Close()
+
+			err := conn.ClientTrackingOn(ctx, &redis.ClientTrackingOptions{
 				OptIn:  true,
 				OptOut: true,
 			}).Err()
 			Expect(err).To(MatchError(ContainSubstring("OPTIN and OPTOUT")))
 
-			err = client.ClientTrackingOn(ctx, &redis.ClientTrackingOptions{
+			err = conn.ClientTrackingOn(ctx, &redis.ClientTrackingOptions{
 				Bcast: true,
 				OptIn: true,
 			}).Err()
 			Expect(err).To(MatchError(ContainSubstring("BCAST cannot be combined")))
 
-			err = client.ClientTrackingOn(ctx, &redis.ClientTrackingOptions{
+			err = conn.ClientTrackingOn(ctx, &redis.ClientTrackingOptions{
 				Prefixes: []string{"k:"},
 			}).Err()
 			Expect(err).To(MatchError(ContainSubstring("PREFIX requires BCAST")))
 		})
 
+		It("should reject per-connection commands on pooled clients", func() {
+			// CLIENT TRACKING and CLIENT MAINT_NOTIFICATIONS are per-connection
+			// state: through a pool they would land on an arbitrary connection,
+			// so the pooled-client methods fail with guidance instead. The
+			// subject may be the raw client or an autopipeliner face — both
+			// must reject identically.
+			for _, cmd := range []*redis.StatusCmd{
+				client.ClientTracking(ctx, true, nil),
+				client.ClientTrackingOn(ctx, nil),
+				client.ClientTrackingOff(ctx),
+			} {
+				Expect(cmd.Err()).To(MatchError(ContainSubstring("per-connection state")))
+			}
+			Expect(client.ClientMaintNotifications(ctx, true, "none").Err()).
+				To(MatchError(ContainSubstring("per-connection state")))
+
+			// The dedicated-connection variant keeps working — gated, since
+			// CLIENT TRACKING may be unavailable on RE/ACL-restricted targets
+			// (the pooled-rejection assertions above are client-side, no gate).
+			skipIfClientTrackingUnavailable(ctx, rawClient)
+			conn := rawClient.Conn()
+			defer conn.Close()
+			Expect(conn.ClientTrackingOn(ctx, nil).Err()).NotTo(HaveOccurred())
+			Expect(conn.ClientTrackingOff(ctx).Err()).NotTo(HaveOccurred())
+		})
+
 		It("should ConfigGet", func() {
 			val, err := rawClient.ConfigGet(ctx, "*").Result()
 			Expect(err).NotTo(HaveOccurred())