fix(maintnotifications): keep ModeAuto fail-open (#3853)

* fix maintnotifications ModeAuto fallback

* retire maintnotifications connections on downgrade

* guard maintnotifications tracking during downgrade

internal/pool/conn.go

@@ -114,6 +114,11 @@ type Conn struct {
 	// to inform the goroutine using the connection why the connection was closed.
 	closeReason uberatomic.String
 
+	// closeOnPutReason marks an in-use connection for removal when it is returned
+	// to the pool. The socket is left open for the in-flight command and closed
+	// by ConnPool.Put.
+	closeOnPutReason uberatomic.String
+
 	// maintenanceNotifications upgrade support: relaxed timeouts during migrations/failovers
 
 	// Using atomic operations for lock-free access to avoid mutex contention
@@ -437,6 +442,17 @@ func (cn *Conn) IsPooled() bool {
 	return cn.pooled
 }
 
+// MarkCloseOnPut marks the connection for removal when it is returned to the pool.
+func (cn *Conn) MarkCloseOnPut(reason string) {
+	cn.closeOnPutReason.Store(reason)
+}
+
+// CloseOnPutReason returns a non-empty reason when the connection should be
+// removed instead of pooled on Put.
+func (cn *Conn) CloseOnPutReason() string {
+	return cn.closeOnPutReason.Load()
+}
+
 // IsPubSub returns true if the connection is used for PubSub.
 func (cn *Conn) IsPubSub() bool {
 	return cn.pubsub

internal/pool/pool.go

@@ -33,6 +33,11 @@ const (
 
 	// CloseReasonFailover indicates the connection was closed due to a failover event.
 	CloseReasonFailover = "failover"
+
+	// CloseReasonMaintNotificationsDisabled indicates the connection enabled
+	// maintenance notifications, but the client later downgraded and disabled
+	// maintenance notification handling for the pool.
+	CloseReasonMaintNotificationsDisabled = "maintnotifications_disabled"
 )
 
 // Metric state constants for connection state tracking.
@@ -315,6 +320,10 @@ type Stats struct {
 	PubSubStats PubSubStats
 }
 
+type ConnRetirer interface {
+	RetireConns(ctx context.Context, conns []*Conn, reason string)
+}
+
 type Pooler interface {
 	NewConn(context.Context) (*Conn, error)
 	CloseConn(ctx context.Context, cn *Conn, reason string, fromState string) error
@@ -1217,6 +1226,11 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
 	shouldRemove := false
 	var err error
 
+	if reason := cn.CloseOnPutReason(); reason != "" {
+		p.removeConnInternal(ctx, cn, errors.New(reason), freeTurn)
+		return
+	}
+
 	if cn.HasBufferedData() {
 		// Peek at the reply type to check if it's a push notification
 		if replyType, err := cn.PeekReplyTypeSafe(); err != nil || replyType != proto.RespPush {
@@ -1562,6 +1576,47 @@ func (p *ConnPool) closed() bool {
 	return atomic.LoadUint32(&p._closed) == 1
 }
 
+func (p *ConnPool) RetireConns(ctx context.Context, conns []*Conn, reason string) {
+	if len(conns) == 0 {
+		return
+	}
+
+	idleConnSet := make(map[*Conn]struct{})
+	toClose := make([]*Conn, 0, len(conns))
+
+	p.connsMu.Lock()
+	for _, ic := range p.idleConns {
+		idleConnSet[ic] = struct{}{}
+	}
+	for _, cn := range conns {
+		if cn == nil {
+			continue
+		}
+		if _, ok := p.conns[cn.GetID()]; !ok {
+			continue
+		}
+		if _, isIdle := idleConnSet[cn]; isIdle {
+			if p.removeConn(cn) {
+				toClose = append(toClose, cn)
+			}
+			continue
+		}
+		cn.MarkCloseOnPut(reason)
+	}
+	p.connsMu.Unlock()
+
+	if hookManager := p.hookManager.Load(); hookManager != nil {
+		for _, cn := range toClose {
+			hookManager.ProcessOnRemove(ctx, cn, errors.New(reason))
+		}
+	}
+	for _, cn := range toClose {
+		p.recordConnectionMetrics(ctx, cn, reason, MetricStateIdle)
+		_ = p.closeConn(cn)
+	}
+	p.checkMinIdleConns()
+}
+
 func (p *ConnPool) Filter(fn func(*Conn) bool) error {
 	ctx := context.Background()
 

internal/pool/pool_test.go

@@ -75,6 +75,25 @@ var _ = Describe("ConnPool", func() {
 		}))
 	})
 
+	It("should retire idle conns and mark in-use conns for close on put", func() {
+		idleConn, err := connPool.Get(ctx)
+		Expect(err).NotTo(HaveOccurred())
+		inUseConn, err := connPool.Get(ctx)
+		Expect(err).NotTo(HaveOccurred())
+
+		connPool.Put(ctx, idleConn)
+
+		connPool.RetireConns(ctx, []*pool.Conn{idleConn, inUseConn}, pool.CloseReasonMaintNotificationsDisabled)
+		Expect(idleConn.IsClosed()).To(BeTrue())
+		Expect(inUseConn.IsClosed()).To(BeFalse())
+		Expect(inUseConn.CloseOnPutReason()).To(Equal(pool.CloseReasonMaintNotificationsDisabled))
+
+		connPool.Put(ctx, inUseConn)
+		Expect(inUseConn.IsClosed()).To(BeTrue())
+		Expect(connPool.Len()).To(Equal(0))
+		Expect(connPool.IdleLen()).To(Equal(0))
+	})
+
 	It("should unblock client when conn is removed", func() {
 		// Reserve one connection.
 		cn, err := connPool.Get(ctx)

internal_maint_notif_test.go

@@ -100,7 +100,6 @@ func TestInitConnNilMaintNotificationsConfig(t *testing.T) {
 	_ = c.initConn(context.Background(), cn)
 }
 
-
 // mockRESP2Server is a minimal RESP server used to exercise the HELLO
 // fallback path in initConn. It replies with a Redis protocol error to
 // HELLO (simulating a server that only speaks RESP2) and with +OK to every
@@ -211,21 +210,20 @@ func startMockRESP2Server(t *testing.T) *mockRESP2Server {
 	return s
 }
 
-
 // assertNoMaintNotifications fails the test if a CLIENT MAINT_NOTIFICATIONS
 // command was observed by srv.
 func assertNoMaintNotifications(t *testing.T, srv *mockRESP2Server) {
 	t.Helper()
 	for _, cmd := range srv.Commands() {
-		if cmd == "CLIENT MAINT_NOTIFICATIONS" {
-			t.Fatalf("CLIENT MAINT_NOTIFICATIONS must not be sent when HELLO falls back to RESP2; commands observed: %v", srv.Commands())
+		if cmd == maintNotificationsCommand {
+			t.Fatalf("%s must not be sent when HELLO falls back to RESP2; commands observed: %v", maintNotificationsCommand, srv.Commands())
 		}
 	}
 }
 
 // initConnOnMockServer dials srv, puts the connection into INITIALIZING and
 // runs c.initConn against it. It returns the error from initConn (if any).
-func initConnOnMockServer(t *testing.T, c *baseClient, srv *mockRESP2Server) error {
+func initConnOnMockServer(t *testing.T, c *baseClient, srv interface{ Addr() string }) error {
 	t.Helper()
 	netConn, err := net.DialTimeout("tcp", srv.Addr(), 2*time.Second)
 	if err != nil {
@@ -304,3 +302,440 @@ func TestInitConn_HelloFallback_ModeAuto(t *testing.T) {
 		t.Fatalf("expected mode to be silently set to Disabled after RESP2 fallback, got %q", opt.MaintNotificationsConfig.Mode)
 	}
 }
+
+const maintNotificationsCommand = "CLIENT MAINT_NOTIFICATIONS"
+
+// mockMaintNotificationsChangingServer accepts the first maintnotifications
+// probe and rejects later probes, simulating support changing across reconnects.
+type mockMaintNotificationsChangingServer struct {
+	ln net.Listener
+
+	mu                      sync.Mutex
+	commands                []string
+	maintNotificationsCalls int
+}
+
+func (s *mockMaintNotificationsChangingServer) Addr() string { return s.ln.Addr().String() }
+func (s *mockMaintNotificationsChangingServer) Close()       { _ = s.ln.Close() }
+
+func (s *mockMaintNotificationsChangingServer) Commands() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]string, len(s.commands))
+	copy(out, s.commands)
+	return out
+}
+
+func (s *mockMaintNotificationsChangingServer) record(full string) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.commands = append(s.commands, full)
+	if full != maintNotificationsCommand {
+		return 0
+	}
+	s.maintNotificationsCalls++
+	return s.maintNotificationsCalls
+}
+
+func (s *mockMaintNotificationsChangingServer) handle(c net.Conn) {
+	defer c.Close()
+	r := bufio.NewReader(c)
+	for {
+		args, err := readRESPCommand(r)
+		if err != nil {
+			return
+		}
+		if len(args) == 0 {
+			continue
+		}
+		name := strings.ToUpper(args[0])
+		full := name
+		if len(args) > 1 {
+			full = name + " " + strings.ToUpper(args[1])
+		}
+
+		if call := s.record(full); call > 0 {
+			if call == 1 {
+				_, _ = c.Write([]byte("+OK\r\n"))
+			} else {
+				_, _ = c.Write([]byte("-ERR unknown subcommand 'MAINT_NOTIFICATIONS'\r\n"))
+			}
+			continue
+		}
+
+		if name == "HELLO" {
+			_, _ = c.Write([]byte("%1\r\n+proto\r\n:3\r\n"))
+			continue
+		}
+		_, _ = c.Write([]byte("+OK\r\n"))
+	}
+}
+
+func startMockMaintNotificationsChangingServer(t *testing.T) *mockMaintNotificationsChangingServer {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	s := &mockMaintNotificationsChangingServer{ln: ln}
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go s.handle(conn)
+		}
+	}()
+	return s
+}
+
+func newMaintNotificationsTestClient(addr string, mode maintnotifications.Mode) (*Options, *baseClient) {
+	opt := &Options{
+		Addr:     addr,
+		Protocol: 3,
+		MaintNotificationsConfig: &maintnotifications.Config{
+			Mode: mode,
+		},
+	}
+	opt.init()
+	return opt, newTestBaseClient(opt)
+}
+
+func assertMaintNotificationsCalls(t *testing.T, srv interface{ Commands() []string }, want int) {
+	t.Helper()
+	commands := srv.Commands()
+	got := 0
+	for _, cmd := range commands {
+		if cmd == maintNotificationsCommand {
+			got++
+		}
+	}
+	if got != want {
+		t.Fatalf("expected %d %s calls, got %d; commands: %v", want, maintNotificationsCommand, got, commands)
+	}
+}
+
+// TestInitConn_ModeAuto_RemainsFailOpenAfterSuccessfulHandshake verifies that
+// a successful auto probe does not turn ModeAuto into fail-closed ModeEnabled.
+func TestInitConn_ModeAuto_RemainsFailOpenAfterSuccessfulHandshake(t *testing.T) {
+	srv := startMockMaintNotificationsChangingServer(t)
+	defer srv.Close()
+	opt, c := newMaintNotificationsTestClient(srv.Addr(), maintnotifications.ModeAuto)
+
+	if err := initConnOnMockServer(t, c, srv); err != nil {
+		t.Fatalf("first initConn returned error: %v", err)
+	}
+	if got := opt.MaintNotificationsConfig.Mode; got != maintnotifications.ModeAuto {
+		t.Fatalf("ModeAuto should remain the configured policy after a successful probe, got %q", got)
+	}
+
+	if err := initConnOnMockServer(t, c, srv); err != nil {
+		t.Fatalf("ModeAuto should downgrade and continue when a later maintnotifications probe is unsupported, got: %v", err)
+	}
+	assertMaintNotificationsCalls(t, srv, 2)
+
+	if err := initConnOnMockServer(t, c, srv); err != nil {
+		t.Fatalf("initConn after ModeAuto downgrade returned error: %v", err)
+	}
+	assertMaintNotificationsCalls(t, srv, 2)
+}
+
+// TestInitConn_ModeEnabled_RemainsFailClosedAfterSuccessfulHandshake verifies
+// that explicit ModeEnabled keeps fail-closed semantics.
+func TestInitConn_ModeEnabled_RemainsFailClosedAfterSuccessfulHandshake(t *testing.T) {
+	srv := startMockMaintNotificationsChangingServer(t)
+	defer srv.Close()
+	_, c := newMaintNotificationsTestClient(srv.Addr(), maintnotifications.ModeEnabled)
+
+	if err := initConnOnMockServer(t, c, srv); err != nil {
+		t.Fatalf("first initConn returned error: %v", err)
+	}
+	err := initConnOnMockServer(t, c, srv)
+	if err == nil {
+		t.Fatal("ModeEnabled should fail when a later maintnotifications probe is unsupported")
+	}
+	if !strings.Contains(err.Error(), "MAINT_NOTIFICATIONS") {
+		t.Fatalf("expected maintnotifications error, got: %v", err)
+	}
+	assertMaintNotificationsCalls(t, srv, 2)
+}
+
+type mockMaintNotificationsDowngradeServer struct {
+	ln net.Listener
+
+	mu       sync.Mutex
+	commands []string
+
+	firstPingBlocked chan struct{}
+	releaseFirstPing chan struct{}
+	blockFirstPing   sync.Once
+}
+
+func (s *mockMaintNotificationsDowngradeServer) Addr() string { return s.ln.Addr().String() }
+func (s *mockMaintNotificationsDowngradeServer) Close()       { _ = s.ln.Close() }
+
+func (s *mockMaintNotificationsDowngradeServer) Commands() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]string, len(s.commands))
+	copy(out, s.commands)
+	return out
+}
+
+func (s *mockMaintNotificationsDowngradeServer) record(full string) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.commands = append(s.commands, full)
+	if full != maintNotificationsCommand {
+		return 0
+	}
+	calls := 0
+	for _, cmd := range s.commands {
+		if cmd == maintNotificationsCommand {
+			calls++
+		}
+	}
+	return calls
+}
+
+func (s *mockMaintNotificationsDowngradeServer) handle(c net.Conn) {
+	defer c.Close()
+	r := bufio.NewReader(c)
+	for {
+		args, err := readRESPCommand(r)
+		if err != nil {
+			return
+		}
+		if len(args) == 0 {
+			continue
+		}
+		name := strings.ToUpper(args[0])
+		full := name
+		if len(args) > 1 {
+			full = name + " " + strings.ToUpper(args[1])
+		}
+
+		switch {
+		case name == "HELLO":
+			s.record(full)
+			_, _ = c.Write([]byte("%1\r\n+proto\r\n:3\r\n"))
+		case full == maintNotificationsCommand:
+			if call := s.record(full); call == 1 {
+				_, _ = c.Write([]byte("+OK\r\n"))
+			} else {
+				_, _ = c.Write([]byte("-ERR unknown subcommand 'MAINT_NOTIFICATIONS'\r\n"))
+			}
+		case name == "PING":
+			s.record(full)
+			shouldBlock := false
+			s.blockFirstPing.Do(func() {
+				shouldBlock = true
+				close(s.firstPingBlocked)
+			})
+			if shouldBlock {
+				<-s.releaseFirstPing
+			}
+			_, _ = c.Write([]byte("+PONG\r\n"))
+		default:
+			s.record(full)
+			_, _ = c.Write([]byte("+OK\r\n"))
+		}
+	}
+}
+
+func startMockMaintNotificationsDowngradeServer(t *testing.T) *mockMaintNotificationsDowngradeServer {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	s := &mockMaintNotificationsDowngradeServer{
+		ln:               ln,
+		firstPingBlocked: make(chan struct{}),
+		releaseFirstPing: make(chan struct{}),
+	}
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go s.handle(conn)
+		}
+	}()
+	return s
+}
+
+func TestClient_ModeAutoDowngradeRetiresEnabledInUseConnOnPut(t *testing.T) {
+	srv := startMockMaintNotificationsDowngradeServer(t)
+	defer srv.Close()
+
+	client := NewClient(&Options{
+		Addr:     srv.Addr(),
+		Protocol: 3,
+		PoolSize: 2,
+		MaintNotificationsConfig: &maintnotifications.Config{
+			Mode: maintnotifications.ModeAuto,
+		},
+	})
+	defer client.Close()
+
+	firstPingDone := make(chan error, 1)
+	go func() {
+		firstPingDone <- client.Ping(context.Background()).Err()
+	}()
+
+	select {
+	case <-srv.firstPingBlocked:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for first ping to block")
+	}
+
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		t.Fatalf("second ping should succeed after ModeAuto downgrade, got: %v", err)
+	}
+	assertMaintNotificationsCalls(t, srv, 2)
+	if got := client.connPool.Len(); got != 2 {
+		t.Fatalf("expected both connections to remain in pool before first ping returns, got %d", got)
+	}
+
+	close(srv.releaseFirstPing)
+	select {
+	case err := <-firstPingDone:
+		if err != nil {
+			t.Fatalf("first ping returned error: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for first ping to finish")
+	}
+
+	if got := client.connPool.Len(); got != 1 {
+		t.Fatalf("expected maintnotifications-enabled connection to be retired on put, got pool len %d", got)
+	}
+
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		t.Fatalf("ping after downgrade returned error: %v", err)
+	}
+	assertMaintNotificationsCalls(t, srv, 2)
+}
+
+type mockMaintNotificationsBlockingHandshakeServer struct {
+	ln net.Listener
+
+	handshakeStarted chan struct{}
+	releaseHandshake chan struct{}
+	startOnce        sync.Once
+}
+
+func (s *mockMaintNotificationsBlockingHandshakeServer) Addr() string { return s.ln.Addr().String() }
+func (s *mockMaintNotificationsBlockingHandshakeServer) Close()       { _ = s.ln.Close() }
+
+func (s *mockMaintNotificationsBlockingHandshakeServer) handle(c net.Conn) {
+	defer c.Close()
+	r := bufio.NewReader(c)
+	for {
+		args, err := readRESPCommand(r)
+		if err != nil {
+			return
+		}
+		if len(args) == 0 {
+			continue
+		}
+		name := strings.ToUpper(args[0])
+		full := name
+		if len(args) > 1 {
+			full = name + " " + strings.ToUpper(args[1])
+		}
+
+		switch {
+		case name == "HELLO":
+			_, _ = c.Write([]byte("%1\r\n+proto\r\n:3\r\n"))
+		case full == maintNotificationsCommand:
+			s.startOnce.Do(func() { close(s.handshakeStarted) })
+			<-s.releaseHandshake
+			_, _ = c.Write([]byte("+OK\r\n"))
+		case name == "PING":
+			_, _ = c.Write([]byte("+PONG\r\n"))
+		default:
+			_, _ = c.Write([]byte("+OK\r\n"))
+		}
+	}
+}
+
+func startMockMaintNotificationsBlockingHandshakeServer(t *testing.T) *mockMaintNotificationsBlockingHandshakeServer {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	s := &mockMaintNotificationsBlockingHandshakeServer{
+		ln:               ln,
+		handshakeStarted: make(chan struct{}),
+		releaseHandshake: make(chan struct{}),
+	}
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go s.handle(conn)
+		}
+	}()
+	return s
+}
+
+func TestClient_ModeAutoDowngradeWaitsForInFlightMaintNotificationsHandshake(t *testing.T) {
+	srv := startMockMaintNotificationsBlockingHandshakeServer(t)
+	defer srv.Close()
+
+	client := NewClient(&Options{
+		Addr:     srv.Addr(),
+		Protocol: 3,
+		PoolSize: 1,
+		MaintNotificationsConfig: &maintnotifications.Config{
+			Mode: maintnotifications.ModeAuto,
+		},
+	})
+	defer client.Close()
+
+	pingDone := make(chan error, 1)
+	go func() {
+		pingDone <- client.Ping(context.Background()).Err()
+	}()
+
+	select {
+	case <-srv.handshakeStarted:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for maintnotifications handshake")
+	}
+
+	downgradeDone := make(chan error, 1)
+	go func() {
+		downgradeDone <- client.disableMaintNotificationsUpgrades()
+	}()
+
+	close(srv.releaseHandshake)
+
+	select {
+	case err := <-pingDone:
+		if err != nil {
+			t.Fatalf("ping returned error: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for ping")
+	}
+	select {
+	case err := <-downgradeDone:
+		if err != nil {
+			t.Fatalf("downgrade returned error: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for downgrade")
+	}
+
+	if got := client.connPool.Len(); got != 0 {
+		t.Fatalf("expected in-flight maintnotifications connection to be retired after downgrade, got pool len %d", got)
+	}
+}

maintnotifications/manager.go

@@ -82,6 +82,10 @@ type Manager struct {
 	hooksMu      sync.RWMutex // Protects hooks slice
 	poolHooksRef *PoolHook
 
+	// Connections that successfully enabled maintnotifications. These need to be
+	// retired before the pool-level listeners are removed.
+	maintNotificationsConns sync.Map // connID -> *pool.Conn
+
 	// Cluster state reload callback for SMIGRATED notifications
 	clusterStateReloadCallback ClusterStateReloadCallback
 }
@@ -251,6 +255,49 @@ func (hm *Manager) MarkSMigratedSeqIDProcessed(seqID int64) bool {
 	return !alreadyProcessed // Return true if NOT already processed
 }
 
+// TrackMaintNotificationsConn records a connection that successfully enabled
+// maintnotifications so it can be retired if the feature is later disabled for
+// the pool.
+func (hm *Manager) TrackMaintNotificationsConn(cn *pool.Conn) {
+	if cn == nil {
+		return
+	}
+	hm.maintNotificationsConns.Store(cn.GetID(), cn)
+}
+
+// UntrackMaintNotificationsConn removes a connection from the enabled
+// maintnotifications set.
+func (hm *Manager) UntrackMaintNotificationsConn(connID uint64) {
+	hm.maintNotificationsConns.Delete(connID)
+}
+
+func (hm *Manager) maintNotificationsConnSnapshot() []*pool.Conn {
+	var conns []*pool.Conn
+	hm.maintNotificationsConns.Range(func(_, value interface{}) bool {
+		if cn, ok := value.(*pool.Conn); ok {
+			conns = append(conns, cn)
+		}
+		return true
+	})
+	return conns
+}
+
+func (hm *Manager) retireMaintNotificationsConns(ctx context.Context) {
+	conns := hm.maintNotificationsConnSnapshot()
+	if len(conns) == 0 || hm.pool == nil {
+		return
+	}
+
+	if retirer, ok := hm.pool.(pool.ConnRetirer); ok {
+		retirer.RetireConns(ctx, conns, pool.CloseReasonMaintNotificationsDisabled)
+		return
+	}
+
+	for _, cn := range conns {
+		_ = hm.pool.CloseConn(ctx, cn, pool.CloseReasonMaintNotificationsDisabled, pool.MetricStateIdle)
+	}
+}
+
 // Close closes the manager.
 func (hm *Manager) Close() error {
 	// Use atomic operation for thread-safe close check
@@ -258,6 +305,10 @@ func (hm *Manager) Close() error {
 		return nil // Already closed
 	}
 
+	// Retire connections that enabled maintnotifications before removing the
+	// pool-level listeners that process those push notifications.
+	hm.retireMaintNotificationsConns(context.Background())
+
 	// Shutdown the pool hook if it exists
 	if hm.poolHooksRef != nil {
 		// Use a timeout to prevent hanging indefinitely

maintnotifications/manager_test.go

@@ -2,11 +2,13 @@ package maintnotifications
 
 import (
 	"context"
+	"errors"
 	"net"
 	"testing"
 	"time"
 
 	"github.com/redis/go-redis/v9/internal/interfaces"
+	"github.com/redis/go-redis/v9/internal/pool"
 )
 
 // MockClient implements interfaces.ClientInterface for testing
@@ -78,6 +80,75 @@ func (mo *MockOptions) NewDialer() func(context.Context) (net.Conn, error) {
 	}
 }
 
+type mockRetirePool struct {
+	pool.Pooler
+
+	retiredConns     []*pool.Conn
+	retireReason     string
+	removeHookCalled bool
+	retireBeforeHook bool
+}
+
+func (p *mockRetirePool) AddPoolHook(pool.PoolHook) {}
+func (p *mockRetirePool) RemovePoolHook(pool.PoolHook) {
+	p.removeHookCalled = true
+}
+func (p *mockRetirePool) RetireConns(_ context.Context, conns []*pool.Conn, reason string) {
+	p.retiredConns = append(p.retiredConns, conns...)
+	p.retireReason = reason
+	p.retireBeforeHook = !p.removeHookCalled
+}
+
+func TestManagerMaintNotificationsConnCleanup(t *testing.T) {
+	client := &MockClient{options: &MockOptions{}}
+	mockPool := &mockRetirePool{}
+	manager, err := NewManager(client, mockPool, DefaultConfig())
+	if err != nil {
+		t.Fatalf("failed to create manager: %v", err)
+	}
+	manager.InitPoolHook(func(context.Context, string, string) (net.Conn, error) { return nil, nil })
+
+	conn := createMockPoolConnection()
+	manager.TrackMaintNotificationsConn(conn)
+
+	if err := manager.Close(); err != nil {
+		t.Fatalf("manager close failed: %v", err)
+	}
+	if len(mockPool.retiredConns) != 1 || mockPool.retiredConns[0] != conn {
+		t.Fatalf("expected tracked connection to be retired, got %v", mockPool.retiredConns)
+	}
+	if mockPool.retireReason != pool.CloseReasonMaintNotificationsDisabled {
+		t.Fatalf("expected retire reason %q, got %q", pool.CloseReasonMaintNotificationsDisabled, mockPool.retireReason)
+	}
+	if !mockPool.retireBeforeHook {
+		t.Fatal("expected maintnotifications connections to be retired before removing the pool hook")
+	}
+	if !mockPool.removeHookCalled {
+		t.Fatal("expected pool hook to be removed")
+	}
+}
+
+func TestManagerOnRemoveUntracksMaintNotificationsConn(t *testing.T) {
+	client := &MockClient{options: &MockOptions{}}
+	mockPool := &mockRetirePool{}
+	manager, err := NewManager(client, mockPool, DefaultConfig())
+	if err != nil {
+		t.Fatalf("failed to create manager: %v", err)
+	}
+	manager.InitPoolHook(func(context.Context, string, string) (net.Conn, error) { return nil, nil })
+
+	conn := createMockPoolConnection()
+	manager.TrackMaintNotificationsConn(conn)
+	manager.poolHooksRef.OnRemove(context.Background(), conn, errors.New("removed"))
+
+	if err := manager.Close(); err != nil {
+		t.Fatalf("manager close failed: %v", err)
+	}
+	if len(mockPool.retiredConns) != 0 {
+		t.Fatalf("expected removed connection not to be retired, got %v", mockPool.retiredConns)
+	}
+}
+
 func TestManagerRefactoring(t *testing.T) {
 	t.Run("AtomicStateTracking", func(t *testing.T) {
 		config := DefaultConfig()

maintnotifications/pool_hook.go

@@ -17,6 +17,10 @@ type OperationsManagerInterface interface {
 	UntrackOperationWithConnID(seqID int64, connID uint64)
 }
 
+type maintNotificationsConnTracker interface {
+	UntrackMaintNotificationsConn(connID uint64)
+}
+
 // HandoffRequest represents a request to handoff a connection to a new endpoint
 type HandoffRequest struct {
 	Conn     *pool.Conn
@@ -172,8 +176,10 @@ func (ph *PoolHook) OnPut(ctx context.Context, conn *pool.Conn) (shouldPool bool
 	return true, false, nil
 }
 
-func (ph *PoolHook) OnRemove(_ context.Context, _ *pool.Conn, _ error) {
-	// Not used
+func (ph *PoolHook) OnRemove(_ context.Context, conn *pool.Conn, _ error) {
+	if tracker, ok := ph.operationsManager.(maintNotificationsConnTracker); ok && conn != nil {
+		tracker.UntrackMaintNotificationsConn(conn.GetID())
+	}
 }
 
 // Shutdown gracefully shuts down the processor, waiting for workers to complete

redis.go

@@ -701,11 +701,24 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 
 	var maintNotifHandshakeErr error
 	if maintNotifEnabled && protocol == 3 {
+		// Hold the manager read lock across the handshake and tracking so a
+		// concurrent downgrade cannot remove pool-level listeners before a
+		// successfully enabled connection is tracked for retirement.
+		c.maintNotificationsManagerLock.RLock()
+		manager := c.maintNotificationsManager
 		maintNotifHandshakeErr = conn.ClientMaintNotifications(
 			ctx,
 			true,
 			endpointType.String(),
 		).Err()
+		// A successful handshake enables maintnotifications for this connection,
+		// but must not promote ModeAuto to ModeEnabled. ModeEnabled is the
+		// explicit fail-closed policy; ModeAuto must remain able to downgrade if a
+		// later reconnect/failover reaches an endpoint that rejects the command.
+		if maintNotifHandshakeErr == nil && manager != nil {
+			manager.TrackMaintNotificationsConn(cn)
+		}
+		c.maintNotificationsManagerLock.RUnlock()
 		if maintNotifHandshakeErr != nil {
 			if !isRedisError(maintNotifHandshakeErr) {
 				// if not redis error, fail the connection
@@ -738,13 +751,6 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 					internal.Logger.Printf(ctx, "failed to disable maintnotifications in auto mode: %v", initErr)
 				}
 			}
-		} else {
-			// handshake was executed successfully
-			// to make sure that the handshake will be executed on other connections as well if it was successfully
-			// executed on this connection, we will force the handshake to be executed on all connections
-			c.optLock.Lock()
-			c.opt.MaintNotificationsConfig.Mode = maintnotifications.ModeEnabled
-			c.optLock.Unlock()
 		}
 	}
 
@@ -1069,7 +1075,9 @@ func (c *baseClient) disableMaintNotificationsUpgrades() error {
 	if c.maintNotificationsManager != nil {
 		// Closing the manager will also shutdown the pool hook
 		// and remove it from the pool
-		c.maintNotificationsManager.Close()
+		if err := c.maintNotificationsManager.Close(); err != nil {
+			return err
+		}
 		c.maintNotificationsManager = nil
 	}
 	return nil