Size computation for allocation may overflow (#99)

* Size computation for allocation may overflow

Performing calculations involving the size of potentially large strings or slices can result in an overflow (for signed integer types) or a wraparound (for unsigned types). An overflow causes the result of the calculation to become negative, while a wraparound results in a small (positive) number.

vendor/github.com/redis/go-redis/v9/redis.go

@@ -11,9 +11,12 @@ import (
 
 	"github.com/redis/go-redis/v9/auth"
 	"github.com/redis/go-redis/v9/internal"
+	"github.com/redis/go-redis/v9/internal/auth/streaming"
 	"github.com/redis/go-redis/v9/internal/hscan"
 	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/internal/proto"
+	"github.com/redis/go-redis/v9/maintnotifications"
+	"github.com/redis/go-redis/v9/push"
 )
 
 // Scanner internal/hscan.Scanner exposed interface.
@@ -23,10 +26,16 @@ type Scanner = hscan.Scanner
 const Nil = proto.Nil
 
 // SetLogger set custom log
+// Use with VoidLogger to disable logging.
 func SetLogger(logger internal.Logging) {
 	internal.Logger = logger
 }
 
+// SetLogLevel sets the log level for the library.
+func SetLogLevel(logLevel internal.LogLevelT) {
+	internal.LogLevel = logLevel
+}
+
 //------------------------------------------------------------------------------
 
 type Hook interface {
@@ -202,16 +211,39 @@ func (hs *hooksMixin) processTxPipelineHook(ctx context.Context, cmds []Cmder) e
 //------------------------------------------------------------------------------
 
 type baseClient struct {
-	opt      *Options
-	connPool pool.Pooler
+	opt        *Options
+	optLock    sync.RWMutex
+	connPool   pool.Pooler
+	pubSubPool *pool.PubSubPool
 	hooksMixin
 
 	onClose func() error // hook called when client is closed
+
+	// Push notification processing
+	pushProcessor push.NotificationProcessor
+
+	// Maintenance notifications manager
+	maintNotificationsManager     *maintnotifications.Manager
+	maintNotificationsManagerLock sync.RWMutex
+
+	// streamingCredentialsManager is used to manage streaming credentials
+	streamingCredentialsManager *streaming.Manager
 }
 
 func (c *baseClient) clone() *baseClient {
-	clone := *c
-	return &clone
+	c.maintNotificationsManagerLock.RLock()
+	maintNotificationsManager := c.maintNotificationsManager
+	c.maintNotificationsManagerLock.RUnlock()
+
+	clone := &baseClient{
+		opt:                         c.opt,
+		connPool:                    c.connPool,
+		onClose:                     c.onClose,
+		pushProcessor:               c.pushProcessor,
+		maintNotificationsManager:   maintNotificationsManager,
+		streamingCredentialsManager: c.streamingCredentialsManager,
+	}
+	return clone
 }
 
 func (c *baseClient) withTimeout(timeout time.Duration) *baseClient {
@@ -229,21 +261,6 @@ func (c *baseClient) String() string {
 	return fmt.Sprintf("Redis<%s db:%d>", c.getAddr(), c.opt.DB)
 }
 
-func (c *baseClient) newConn(ctx context.Context) (*pool.Conn, error) {
-	cn, err := c.connPool.NewConn(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	err = c.initConn(ctx, cn)
-	if err != nil {
-		_ = c.connPool.CloseConn(cn)
-		return nil, err
-	}
-
-	return cn, nil
-}
-
 func (c *baseClient) getConn(ctx context.Context) (*pool.Conn, error) {
 	if c.opt.Limiter != nil {
 		err := c.opt.Limiter.Allow()
@@ -269,7 +286,7 @@ func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
 		return nil, err
 	}
 
-	if cn.Inited {
+	if cn.IsInited() {
 		return cn, nil
 	}
 
@@ -281,35 +298,39 @@ func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
 		return nil, err
 	}
 
+	// initConn will transition to IDLE state, so we need to acquire it
+	// before returning it to the user.
+	if !cn.TryAcquire() {
+		return nil, fmt.Errorf("redis: connection is not usable")
+	}
+
 	return cn, nil
 }
 
-func (c *baseClient) newReAuthCredentialsListener(poolCn *pool.Conn) auth.CredentialsListener {
-	return auth.NewReAuthCredentialsListener(
-		c.reAuthConnection(poolCn),
-		c.onAuthenticationErr(poolCn),
-	)
-}
-
-func (c *baseClient) reAuthConnection(poolCn *pool.Conn) func(credentials auth.Credentials) error {
-	return func(credentials auth.Credentials) error {
+func (c *baseClient) reAuthConnection() func(poolCn *pool.Conn, credentials auth.Credentials) error {
+	return func(poolCn *pool.Conn, credentials auth.Credentials) error {
 		var err error
 		username, password := credentials.BasicAuth()
+
+		// Use background context - timeout is handled by ReadTimeout in WithReader/WithWriter
 		ctx := context.Background()
+
 		connPool := pool.NewSingleConnPool(c.connPool, poolCn)
-		// hooksMixin are intentionally empty here
-		cn := newConn(c.opt, connPool, nil)
+
+		// Pass hooks so that reauth commands are recorded/traced
+		cn := newConn(c.opt, connPool, &c.hooksMixin)
 
 		if username != "" {
 			err = cn.AuthACL(ctx, username, password).Err()
 		} else {
 			err = cn.Auth(ctx, password).Err()
 		}
+
 		return err
 	}
 }
-func (c *baseClient) onAuthenticationErr(poolCn *pool.Conn) func(err error) {
-	return func(err error) {
+func (c *baseClient) onAuthenticationErr() func(poolCn *pool.Conn, err error) {
+	return func(poolCn *pool.Conn, err error) {
 		if err != nil {
 			if isBadConn(err, false, c.opt.Addr) {
 				// Close the connection to force a reconnection.
@@ -351,29 +372,113 @@ func (c *baseClient) wrappedOnClose(newOnClose func() error) func() error {
 }
 
 func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
-	if cn.Inited {
+	// This function is called in two scenarios:
+	// 1. First-time init: Connection is in CREATED state (from pool.Get())
+	//    - We need to transition CREATED → INITIALIZING and do the initialization
+	//    - If another goroutine is already initializing, we WAIT for it to finish
+	// 2. Re-initialization: Connection is in INITIALIZING state (from SetNetConnAndInitConn())
+	//    - We're already in INITIALIZING, so just proceed with initialization
+
+	currentState := cn.GetStateMachine().GetState()
+
+	// Fast path: Check if already initialized (IDLE or IN_USE)
+	if currentState == pool.StateIdle || currentState == pool.StateInUse {
 		return nil
 	}
 
-	var err error
-	cn.Inited = true
+	// If in CREATED state, try to transition to INITIALIZING
+	if currentState == pool.StateCreated {
+		finalState, err := cn.GetStateMachine().TryTransition([]pool.ConnState{pool.StateCreated}, pool.StateInitializing)
+		if err != nil {
+			// Another goroutine is initializing or connection is in unexpected state
+			// Check what state we're in now
+			if finalState == pool.StateIdle || finalState == pool.StateInUse {
+				// Already initialized by another goroutine
+				return nil
+			}
+
+			if finalState == pool.StateInitializing {
+				// Another goroutine is initializing - WAIT for it to complete
+				// Use a context with timeout = min(remaining command timeout, DialTimeout)
+				// This prevents waiting too long while respecting the caller's deadline
+				var waitCtx context.Context
+				var cancel context.CancelFunc
+				dialTimeout := c.opt.DialTimeout
+
+				if cmdDeadline, hasCmdDeadline := ctx.Deadline(); hasCmdDeadline {
+					// Calculate remaining time until command deadline
+					remainingTime := time.Until(cmdDeadline)
+					// Use the minimum of remaining time and DialTimeout
+					if remainingTime < dialTimeout {
+						// Command deadline is sooner, use it
+						waitCtx = ctx
+					} else {
+						// DialTimeout is shorter, cap the wait at DialTimeout
+						waitCtx, cancel = context.WithTimeout(ctx, dialTimeout)
+					}
+				} else {
+					// No command deadline, use DialTimeout to prevent waiting indefinitely
+					waitCtx, cancel = context.WithTimeout(ctx, dialTimeout)
+				}
+				if cancel != nil {
+					defer cancel()
+				}
+
+				finalState, err := cn.GetStateMachine().AwaitAndTransition(
+					waitCtx,
+					[]pool.ConnState{pool.StateIdle, pool.StateInUse},
+					pool.StateIdle, // Target is IDLE (but we're already there, so this is a no-op)
+				)
+				if err != nil {
+					return err
+				}
+				// Verify we're now initialized
+				if finalState == pool.StateIdle || finalState == pool.StateInUse {
+					return nil
+				}
+				// Unexpected state after waiting
+				return fmt.Errorf("connection in unexpected state after initialization: %s", finalState)
+			}
+
+			// Unexpected state (CLOSED, UNUSABLE, etc.)
+			return err
+		}
+	}
+
+	// At this point, we're in INITIALIZING state and we own the initialization
+	// If we fail, we must transition to CLOSED
+	var initErr error
 	connPool := pool.NewSingleConnPool(c.connPool, cn)
 	conn := newConn(c.opt, connPool, &c.hooksMixin)
 
 	username, password := "", ""
 	if c.opt.StreamingCredentialsProvider != nil {
-		credentials, unsubscribeFromCredentialsProvider, err := c.opt.StreamingCredentialsProvider.
-			Subscribe(c.newReAuthCredentialsListener(cn))
-		if err != nil {
-			return fmt.Errorf("failed to subscribe to streaming credentials: %w", err)
+		credListener, initErr := c.streamingCredentialsManager.Listener(
+			cn,
+			c.reAuthConnection(),
+			c.onAuthenticationErr(),
+		)
+		if initErr != nil {
+			cn.GetStateMachine().Transition(pool.StateClosed)
+			return fmt.Errorf("failed to create credentials listener: %w", initErr)
 		}
+
+		credentials, unsubscribeFromCredentialsProvider, initErr := c.opt.StreamingCredentialsProvider.
+			Subscribe(credListener)
+		if initErr != nil {
+			cn.GetStateMachine().Transition(pool.StateClosed)
+			return fmt.Errorf("failed to subscribe to streaming credentials: %w", initErr)
+		}
+
 		c.onClose = c.wrappedOnClose(unsubscribeFromCredentialsProvider)
 		cn.SetOnClose(unsubscribeFromCredentialsProvider)
+
 		username, password = credentials.BasicAuth()
 	} else if c.opt.CredentialsProviderContext != nil {
-		username, password, err = c.opt.CredentialsProviderContext(ctx)
-		if err != nil {
-			return fmt.Errorf("failed to get credentials from context provider: %w", err)
+		username, password, initErr = c.opt.CredentialsProviderContext(ctx)
+		if initErr != nil {
+			cn.GetStateMachine().Transition(pool.StateClosed)
+			return fmt.Errorf("failed to get credentials from context provider: %w", initErr)
 		}
 	} else if c.opt.CredentialsProvider != nil {
 		username, password = c.opt.CredentialsProvider()
@@ -383,9 +488,9 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 
 	// for redis-server versions that do not support the HELLO command,
 	// RESP2 will continue to be used.
-	if err = conn.Hello(ctx, c.opt.Protocol, username, password, c.opt.ClientName).Err(); err == nil {
+	if initErr = conn.Hello(ctx, c.opt.Protocol, username, password, c.opt.ClientName).Err(); initErr == nil {
 		// Authentication successful with HELLO command
-	} else if !isRedisError(err) {
+	} else if !isRedisError(initErr) {
 		// When the server responds with the RESP protocol and the result is not a normal
 		// execution result of the HELLO command, we consider it to be an indication that
 		// the server does not support the HELLO command.
@@ -393,20 +498,22 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 		// or it could be DragonflyDB or a third-party redis-proxy. They all respond
 		// with different error string results for unsupported commands, making it
 		// difficult to rely on error strings to determine all results.
-		return err
+		cn.GetStateMachine().Transition(pool.StateClosed)
+		return initErr
 	} else if password != "" {
 		// Try legacy AUTH command if HELLO failed
 		if username != "" {
-			err = conn.AuthACL(ctx, username, password).Err()
+			initErr = conn.AuthACL(ctx, username, password).Err()
 		} else {
-			err = conn.Auth(ctx, password).Err()
+			initErr = conn.Auth(ctx, password).Err()
 		}
-		if err != nil {
-			return fmt.Errorf("failed to authenticate: %w", err)
+		if initErr != nil {
+			cn.GetStateMachine().Transition(pool.StateClosed)
+			return fmt.Errorf("failed to authenticate: %w", initErr)
 		}
 	}
 
-	_, err = conn.Pipelined(ctx, func(pipe Pipeliner) error {
+	_, initErr = conn.Pipelined(ctx, func(pipe Pipeliner) error {
 		if c.opt.DB > 0 {
 			pipe.Select(ctx, c.opt.DB)
 		}
@@ -421,8 +528,58 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 
 		return nil
 	})
-	if err != nil {
-		return fmt.Errorf("failed to initialize connection options: %w", err)
+	if initErr != nil {
+		cn.GetStateMachine().Transition(pool.StateClosed)
+		return fmt.Errorf("failed to initialize connection options: %w", initErr)
+	}
+
+	// Enable maintnotifications if maintnotifications are configured
+	c.optLock.RLock()
+	maintNotifEnabled := c.opt.MaintNotificationsConfig != nil && c.opt.MaintNotificationsConfig.Mode != maintnotifications.ModeDisabled
+	protocol := c.opt.Protocol
+	endpointType := c.opt.MaintNotificationsConfig.EndpointType
+	c.optLock.RUnlock()
+	var maintNotifHandshakeErr error
+	if maintNotifEnabled && protocol == 3 {
+		maintNotifHandshakeErr = conn.ClientMaintNotifications(
+			ctx,
+			true,
+			endpointType.String(),
+		).Err()
+		if maintNotifHandshakeErr != nil {
+			if !isRedisError(maintNotifHandshakeErr) {
+				// if not redis error, fail the connection
+				cn.GetStateMachine().Transition(pool.StateClosed)
+				return maintNotifHandshakeErr
+			}
+			c.optLock.Lock()
+			// handshake failed - check and modify config atomically
+			switch c.opt.MaintNotificationsConfig.Mode {
+			case maintnotifications.ModeEnabled:
+				// enabled mode, fail the connection
+				c.optLock.Unlock()
+				cn.GetStateMachine().Transition(pool.StateClosed)
+				return fmt.Errorf("failed to enable maintnotifications: %w", maintNotifHandshakeErr)
+			default: // will handle auto and any other
+				// Disabling logging here as it's too noisy.
+				// TODO: Enable when we have a better logging solution for log levels
+				// internal.Logger.Printf(ctx, "auto mode fallback: maintnotifications disabled due to handshake error: %v", maintNotifHandshakeErr)
+				c.opt.MaintNotificationsConfig.Mode = maintnotifications.ModeDisabled
+				c.optLock.Unlock()
+				// auto mode, disable maintnotifications and continue
+				if initErr := c.disableMaintNotificationsUpgrades(); initErr != nil {
+					// Log error but continue - auto mode should be resilient
+					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()
+		}
 	}
 
 	if !c.opt.DisableIdentity && !c.opt.DisableIndentity {
@@ -436,13 +593,31 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 		p.ClientSetInfo(ctx, WithLibraryVersion(libVer))
 		// Handle network errors (e.g. timeouts) in CLIENT SETINFO to avoid
 		// out of order responses later on.
-		if _, err = p.Exec(ctx); err != nil && !isRedisError(err) {
-			return err
+		if _, initErr = p.Exec(ctx); initErr != nil && !isRedisError(initErr) {
+			cn.GetStateMachine().Transition(pool.StateClosed)
+			return initErr
 		}
 	}
 
+	// Set the connection initialization function for potential reconnections
+	// This must be set before transitioning to IDLE so that handoff/reauth can use it
+	cn.SetInitConnFunc(c.createInitConnFunc())
+
+	// Initialization succeeded - transition to IDLE state
+	// This marks the connection as initialized and ready for use
+	// NOTE: The connection is still owned by the calling goroutine at this point
+	// and won't be available to other goroutines until it's Put() back into the pool
+	cn.GetStateMachine().Transition(pool.StateIdle)
+
+	// Call OnConnect hook if configured
+	// The connection is in IDLE state but still owned by this goroutine
+	// If OnConnect needs to send commands, it can use the connection safely
 	if c.opt.OnConnect != nil {
-		return c.opt.OnConnect(ctx, conn)
+		if initErr = c.opt.OnConnect(ctx, conn); initErr != nil {
+			// OnConnect failed - transition to closed
+			cn.GetStateMachine().Transition(pool.StateClosed)
+			return initErr
+		}
 	}
 
 	return nil
@@ -456,6 +631,10 @@ func (c *baseClient) releaseConn(ctx context.Context, cn *pool.Conn, err error)
 	if isBadConn(err, false, c.opt.Addr) {
 		c.connPool.Remove(ctx, cn, err)
 	} else {
+		// process any pending push notifications before returning the connection to the pool
+		if err := c.processPushNotifications(ctx, cn); err != nil {
+			internal.Logger.Printf(ctx, "push: error processing pending notifications before releasing connection: %v", err)
+		}
 		c.connPool.Put(ctx, cn)
 	}
 }
@@ -497,16 +676,16 @@ func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
 	return lastErr
 }
 
-func (c *baseClient) assertUnstableCommand(cmd Cmder) bool {
+func (c *baseClient) assertUnstableCommand(cmd Cmder) (bool, error) {
 	switch cmd.(type) {
 	case *AggregateCmd, *FTInfoCmd, *FTSpellCheckCmd, *FTSearchCmd, *FTSynDumpCmd:
 		if c.opt.UnstableResp3 {
-			return true
+			return true, nil
 		} else {
-			panic("RESP3 responses for this command are disabled because they may still change. Please set the flag UnstableResp3 .  See the [README](https://github.com/redis/go-redis/blob/master/README.md) and the release notes for guidance.")
+			return false, fmt.Errorf("RESP3 responses for this command are disabled because they may still change. Please set the flag UnstableResp3. See the README and the release notes for guidance")
 		}
 	default:
-		return false
+		return false, nil
 	}
 }
 
@@ -519,6 +698,11 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool
 
 	retryTimeout := uint32(0)
 	if err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
+		// Process any pending push notifications before executing the command
+		if err := c.processPushNotifications(ctx, cn); err != nil {
+			internal.Logger.Printf(ctx, "push: error processing pending notifications before command: %v", err)
+		}
+
 		if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 			return writeCmd(wr, cmd)
 		}); err != nil {
@@ -527,10 +711,22 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool
 		}
 		readReplyFunc := cmd.readReply
 		// Apply unstable RESP3 search module.
-		if c.opt.Protocol != 2 && c.assertUnstableCommand(cmd) {
-			readReplyFunc = cmd.readRawReply
+		if c.opt.Protocol != 2 {
+			useRawReply, err := c.assertUnstableCommand(cmd)
+			if err != nil {
+				return err
+			}
+			if useRawReply {
+				readReplyFunc = cmd.readRawReply
+			}
 		}
-		if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), readReplyFunc); err != nil {
+		if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), func(rd *proto.Reader) error {
+			// To be sure there are no buffered push notifications, we process them before reading the reply
+			if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+				internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+			}
+			return readReplyFunc(rd)
+		}); err != nil {
 			if cmd.readTimeout() == nil {
 				atomic.StoreUint32(&retryTimeout, 1)
 			} else {
@@ -573,19 +769,76 @@ func (c *baseClient) context(ctx context.Context) context.Context {
 	return context.Background()
 }
 
+// createInitConnFunc creates a connection initialization function that can be used for reconnections.
+func (c *baseClient) createInitConnFunc() func(context.Context, *pool.Conn) error {
+	return func(ctx context.Context, cn *pool.Conn) error {
+		return c.initConn(ctx, cn)
+	}
+}
+
+// enableMaintNotificationsUpgrades initializes the maintnotifications upgrade manager and pool hook.
+// This function is called during client initialization.
+// will register push notification handlers for all maintenance upgrade events.
+// will start background workers for handoff processing in the pool hook.
+func (c *baseClient) enableMaintNotificationsUpgrades() error {
+	// Create client adapter
+	clientAdapterInstance := newClientAdapter(c)
+
+	// Create maintnotifications manager directly
+	manager, err := maintnotifications.NewManager(clientAdapterInstance, c.connPool, c.opt.MaintNotificationsConfig)
+	if err != nil {
+		return err
+	}
+	// Set the manager reference and initialize pool hook
+	c.maintNotificationsManagerLock.Lock()
+	c.maintNotificationsManager = manager
+	c.maintNotificationsManagerLock.Unlock()
+
+	// Initialize pool hook (safe to call without lock since manager is now set)
+	manager.InitPoolHook(c.dialHook)
+	return nil
+}
+
+func (c *baseClient) disableMaintNotificationsUpgrades() error {
+	c.maintNotificationsManagerLock.Lock()
+	defer c.maintNotificationsManagerLock.Unlock()
+
+	// Close the maintnotifications manager
+	if c.maintNotificationsManager != nil {
+		// Closing the manager will also shutdown the pool hook
+		// and remove it from the pool
+		c.maintNotificationsManager.Close()
+		c.maintNotificationsManager = nil
+	}
+	return nil
+}
+
 // Close closes the client, releasing any open resources.
 //
 // It is rare to Close a Client, as the Client is meant to be
 // long-lived and shared between many goroutines.
 func (c *baseClient) Close() error {
 	var firstErr error
+
+	// Close maintnotifications manager first
+	if err := c.disableMaintNotificationsUpgrades(); err != nil {
+		firstErr = err
+	}
+
 	if c.onClose != nil {
-		if err := c.onClose(); err != nil {
+		if err := c.onClose(); err != nil && firstErr == nil {
 			firstErr = err
 		}
 	}
-	if err := c.connPool.Close(); err != nil && firstErr == nil {
-		firstErr = err
+	if c.connPool != nil {
+		if err := c.connPool.Close(); err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	if c.pubSubPool != nil {
+		if err := c.pubSubPool.Close(); err != nil && firstErr == nil {
+			firstErr = err
+		}
 	}
 	return firstErr
 }
@@ -625,12 +878,19 @@ func (c *baseClient) generalProcessPipeline(
 		// Enable retries by default to retry dial errors returned by withConn.
 		canRetry := true
 		lastErr = c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
+			// Process any pending push notifications before executing the pipeline
+			if err := c.processPushNotifications(ctx, cn); err != nil {
+				internal.Logger.Printf(ctx, "push: error processing pending notifications before processing pipeline: %v", err)
+			}
 			var err error
 			canRetry, err = p(ctx, cn, cmds)
 			return err
 		})
 		if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) {
-			setCmdsErr(cmds, lastErr)
+			// The error should be set here only when failing to obtain the conn.
+			if !isRedisError(lastErr) {
+				setCmdsErr(cmds, lastErr)
+			}
 			return lastErr
 		}
 	}
@@ -640,6 +900,11 @@ func (c *baseClient) generalProcessPipeline(
 func (c *baseClient) pipelineProcessCmds(
 	ctx context.Context, cn *pool.Conn, cmds []Cmder,
 ) (bool, error) {
+	// Process any pending push notifications before executing the pipeline
+	if err := c.processPushNotifications(ctx, cn); err != nil {
+		internal.Logger.Printf(ctx, "push: error processing pending notifications before writing pipeline: %v", err)
+	}
+
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
 	}); err != nil {
@@ -648,7 +913,8 @@ func (c *baseClient) pipelineProcessCmds(
 	}
 
 	if err := cn.WithReader(c.context(ctx), c.opt.ReadTimeout, func(rd *proto.Reader) error {
-		return pipelineReadCmds(rd, cmds)
+		// read all replies
+		return c.pipelineReadCmds(ctx, cn, rd, cmds)
 	}); err != nil {
 		return true, err
 	}
@@ -656,8 +922,12 @@ func (c *baseClient) pipelineProcessCmds(
 	return false, nil
 }
 
-func pipelineReadCmds(rd *proto.Reader, cmds []Cmder) error {
+func (c *baseClient) pipelineReadCmds(ctx context.Context, cn *pool.Conn, rd *proto.Reader, cmds []Cmder) error {
 	for i, cmd := range cmds {
+		// To be sure there are no buffered push notifications, we process them before reading the reply
+		if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+		}
 		err := cmd.readReply(rd)
 		cmd.SetErr(err)
 		if err != nil && !isRedisError(err) {
@@ -672,6 +942,11 @@ func pipelineReadCmds(rd *proto.Reader, cmds []Cmder) error {
 func (c *baseClient) txPipelineProcessCmds(
 	ctx context.Context, cn *pool.Conn, cmds []Cmder,
 ) (bool, error) {
+	// Process any pending push notifications before executing the transaction pipeline
+	if err := c.processPushNotifications(ctx, cn); err != nil {
+		internal.Logger.Printf(ctx, "push: error processing pending notifications before transaction: %v", err)
+	}
+
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
 	}); err != nil {
@@ -684,12 +959,13 @@ func (c *baseClient) txPipelineProcessCmds(
 		// Trim multi and exec.
 		trimmedCmds := cmds[1 : len(cmds)-1]
 
-		if err := txPipelineReadQueued(rd, statusCmd, trimmedCmds); err != nil {
+		if err := c.txPipelineReadQueued(ctx, cn, rd, statusCmd, trimmedCmds); err != nil {
 			setCmdsErr(cmds, err)
 			return err
 		}
 
-		return pipelineReadCmds(rd, trimmedCmds)
+		// Read replies.
+		return c.pipelineReadCmds(ctx, cn, rd, trimmedCmds)
 	}); err != nil {
 		return false, err
 	}
@@ -697,7 +973,13 @@ func (c *baseClient) txPipelineProcessCmds(
 	return false, nil
 }
 
-func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder) error {
+// txPipelineReadQueued reads queued replies from the Redis server.
+// It returns an error if the server returns an error or if the number of replies does not match the number of commands.
+func (c *baseClient) txPipelineReadQueued(ctx context.Context, cn *pool.Conn, rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder) error {
+	// To be sure there are no buffered push notifications, we process them before reading the reply
+	if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+		internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+	}
 	// Parse +OK.
 	if err := statusCmd.readReply(rd); err != nil {
 		return err
@@ -705,6 +987,10 @@ func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder)
 
 	// Parse +QUEUED.
 	for _, cmd := range cmds {
+		// To be sure there are no buffered push notifications, we process them before reading the reply
+		if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+		}
 		if err := statusCmd.readReply(rd); err != nil {
 			cmd.SetErr(err)
 			if !isRedisError(err) {
@@ -713,6 +999,10 @@ func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder)
 		}
 	}
 
+	// To be sure there are no buffered push notifications, we process them before reading the reply
+	if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+		internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+	}
 	// Parse number of replies.
 	line, err := rd.ReadLine()
 	if err != nil {
@@ -746,15 +1036,61 @@ func NewClient(opt *Options) *Client {
 	if opt == nil {
 		panic("redis: NewClient nil options")
 	}
+	// clone to not share options with the caller
+	opt = opt.clone()
 	opt.init()
 
+	// Push notifications are always enabled for RESP3 (cannot be disabled)
+
 	c := Client{
 		baseClient: &baseClient{
 			opt: opt,
 		},
 	}
 	c.init()
-	c.connPool = newConnPool(opt, c.dialHook)
+
+	// Initialize push notification processor using shared helper
+	// Use void processor for RESP2 connections (push notifications not available)
+	c.pushProcessor = initializePushProcessor(opt)
+	// set opt push processor for child clients
+	c.opt.PushNotificationProcessor = c.pushProcessor
+
+	// Create connection pools
+	var err error
+	c.connPool, err = newConnPool(opt, c.dialHook)
+	if err != nil {
+		panic(fmt.Errorf("redis: failed to create connection pool: %w", err))
+	}
+	c.pubSubPool, err = newPubSubPool(opt, c.dialHook)
+	if err != nil {
+		panic(fmt.Errorf("redis: failed to create pubsub pool: %w", err))
+	}
+
+	if opt.StreamingCredentialsProvider != nil {
+		c.streamingCredentialsManager = streaming.NewManager(c.connPool, c.opt.PoolTimeout)
+		c.connPool.AddPoolHook(c.streamingCredentialsManager.PoolHook())
+	}
+
+	// Initialize maintnotifications first if enabled and protocol is RESP3
+	if opt.MaintNotificationsConfig != nil && opt.MaintNotificationsConfig.Mode != maintnotifications.ModeDisabled && opt.Protocol == 3 {
+		err := c.enableMaintNotificationsUpgrades()
+		if err != nil {
+			internal.Logger.Printf(context.Background(), "failed to initialize maintnotifications: %v", err)
+			if opt.MaintNotificationsConfig.Mode == maintnotifications.ModeEnabled {
+				/*
+					Design decision: panic here to fail fast if maintnotifications cannot be enabled when explicitly requested.
+					We choose to panic instead of returning an error to avoid breaking the existing client API, which does not expect
+					an error from NewClient. This ensures that misconfiguration or critical initialization failures are surfaced
+					immediately, rather than allowing the client to continue in a partially initialized or inconsistent state.
+					Clients relying on maintnotifications should be aware that initialization errors will cause a panic, and should
+					handle this accordingly (e.g., via recover or by validating configuration before calling NewClient).
+					This approach is only used when MaintNotificationsConfig.Mode is MaintNotificationsEnabled, indicating that maintnotifications
+					upgrades are required for correct operation. In other modes, initialization failures are logged but do not panic.
+				*/
+				panic(fmt.Errorf("failed to enable maintnotifications: %w", err))
+			}
+		}
+	}
 
 	return &c
 }
@@ -791,11 +1127,51 @@ func (c *Client) Options() *Options {
 	return c.opt
 }
 
+// GetMaintNotificationsManager returns the maintnotifications manager instance for monitoring and control.
+// Returns nil if maintnotifications are not enabled.
+func (c *Client) GetMaintNotificationsManager() *maintnotifications.Manager {
+	c.maintNotificationsManagerLock.RLock()
+	defer c.maintNotificationsManagerLock.RUnlock()
+	return c.maintNotificationsManager
+}
+
+// initializePushProcessor initializes the push notification processor for any client type.
+// This is a shared helper to avoid duplication across NewClient, NewFailoverClient, and NewSentinelClient.
+func initializePushProcessor(opt *Options) push.NotificationProcessor {
+	// Always use custom processor if provided
+	if opt.PushNotificationProcessor != nil {
+		return opt.PushNotificationProcessor
+	}
+
+	// Push notifications are always enabled for RESP3, disabled for RESP2
+	if opt.Protocol == 3 {
+		// Create default processor for RESP3 connections
+		return NewPushNotificationProcessor()
+	}
+
+	// Create void processor for RESP2 connections (push notifications not available)
+	return NewVoidPushNotificationProcessor()
+}
+
+// RegisterPushNotificationHandler registers a handler for a specific push notification name.
+// Returns an error if a handler is already registered for this push notification name.
+// If protected is true, the handler cannot be unregistered.
+func (c *Client) RegisterPushNotificationHandler(pushNotificationName string, handler push.NotificationHandler, protected bool) error {
+	return c.pushProcessor.RegisterHandler(pushNotificationName, handler, protected)
+}
+
+// GetPushNotificationHandler returns the handler for a specific push notification name.
+// Returns nil if no handler is registered for the given name.
+func (c *Client) GetPushNotificationHandler(pushNotificationName string) push.NotificationHandler {
+	return c.pushProcessor.GetHandler(pushNotificationName)
+}
+
 type PoolStats pool.Stats
 
 // PoolStats returns connection pool stats.
 func (c *Client) PoolStats() *PoolStats {
 	stats := c.connPool.Stats()
+	stats.PubSubStats = *(c.pubSubPool.Stats())
 	return (*PoolStats)(stats)
 }
 
@@ -830,13 +1206,31 @@ func (c *Client) TxPipeline() Pipeliner {
 func (c *Client) pubSub() *PubSub {
 	pubsub := &PubSub{
 		opt: c.opt,
-
-		newConn: func(ctx context.Context, channels []string) (*pool.Conn, error) {
-			return c.newConn(ctx)
+		newConn: func(ctx context.Context, addr string, channels []string) (*pool.Conn, error) {
+			cn, err := c.pubSubPool.NewConn(ctx, c.opt.Network, addr, channels)
+			if err != nil {
+				return nil, err
+			}
+			// will return nil if already initialized
+			err = c.initConn(ctx, cn)
+			if err != nil {
+				_ = cn.Close()
+				return nil, err
+			}
+			// Track connection in PubSubPool
+			c.pubSubPool.TrackConn(cn)
+			return cn, nil
 		},
-		closeConn: c.connPool.CloseConn,
+		closeConn: func(cn *pool.Conn) error {
+			// Untrack connection from PubSubPool
+			c.pubSubPool.UntrackConn(cn)
+			_ = cn.Close()
+			return nil
+		},
+		pushProcessor: c.pushProcessor,
 	}
 	pubsub.init()
+
 	return pubsub
 }
 
@@ -920,6 +1314,10 @@ func newConn(opt *Options, connPool pool.Pooler, parentHooks *hooksMixin) *Conn
 		c.hooksMixin = parentHooks.clone()
 	}
 
+	// Initialize push notification processor using shared helper
+	// Use void processor for RESP2 connections (push notifications not available)
+	c.pushProcessor = initializePushProcessor(opt)
+
 	c.cmdable = c.Process
 	c.statefulCmdable = c.Process
 	c.initHooks(hooks{
@@ -938,6 +1336,13 @@ func (c *Conn) Process(ctx context.Context, cmd Cmder) error {
 	return err
 }
 
+// RegisterPushNotificationHandler registers a handler for a specific push notification name.
+// Returns an error if a handler is already registered for this push notification name.
+// If protected is true, the handler cannot be unregistered.
+func (c *Conn) RegisterPushNotificationHandler(pushNotificationName string, handler push.NotificationHandler, protected bool) error {
+	return c.pushProcessor.RegisterHandler(pushNotificationName, handler, protected)
+}
+
 func (c *Conn) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
 	return c.Pipeline().Pipelined(ctx, fn)
 }
@@ -965,3 +1370,78 @@ func (c *Conn) TxPipeline() Pipeliner {
 	pipe.init()
 	return &pipe
 }
+
+// processPushNotifications processes all pending push notifications on a connection
+// This ensures that cluster topology changes are handled immediately before the connection is used
+// This method should be called by the client before using WithReader for command execution
+//
+// Performance optimization: Skip the expensive MaybeHasData() syscall if a health check
+// was performed recently (within 5 seconds). The health check already verified the connection
+// is healthy and checked for unexpected data (push notifications).
+func (c *baseClient) processPushNotifications(ctx context.Context, cn *pool.Conn) error {
+	// Only process push notifications for RESP3 connections with a processor
+	if c.opt.Protocol != 3 || c.pushProcessor == nil {
+		return nil
+	}
+
+	// Performance optimization: Skip MaybeHasData() syscall if health check was recent
+	// If the connection was health-checked within the last 5 seconds, we can skip the
+	// expensive syscall since the health check already verified no unexpected data.
+	// This is safe because:
+	// 0. lastHealthCheckNs is set in pool/conn.go:putConn() after a successful health check
+	// 1. Health check (connCheck) uses the same syscall (Recvfrom with MSG_PEEK)
+	// 2. If push notifications arrived, they would have been detected by health check
+	// 3. 5 seconds is short enough that connection state is still fresh
+	// 4. Push notifications will be processed by the next WithReader call
+	// used it is set on getConn, so we should use another timer (lastPutAt?)
+	lastHealthCheckNs := cn.LastPutAtNs()
+	if lastHealthCheckNs > 0 {
+		// Use pool's cached time to avoid expensive time.Now() syscall
+		nowNs := pool.GetCachedTimeNs()
+		if nowNs-lastHealthCheckNs < int64(5*time.Second) {
+			// Recent health check confirmed no unexpected data, skip the syscall
+			return nil
+		}
+	}
+
+	// Check if there is any data to read before processing
+	// This is an optimization on UNIX systems where MaybeHasData is a syscall
+	// On Windows, MaybeHasData always returns true, so this check is a no-op
+	if !cn.MaybeHasData() {
+		return nil
+	}
+
+	// Use WithReader to access the reader and process push notifications
+	// This is critical for maintnotifications to work properly
+	// NOTE: almost no timeouts are set for this read, so it should not block
+	// longer than necessary, 10us should be plenty of time to read if there are any push notifications
+	// on the socket.
+	return cn.WithReader(ctx, 10*time.Microsecond, func(rd *proto.Reader) error {
+		// Create handler context with client, connection pool, and connection information
+		handlerCtx := c.pushNotificationHandlerContext(cn)
+		return c.pushProcessor.ProcessPendingNotifications(ctx, handlerCtx, rd)
+	})
+}
+
+// processPendingPushNotificationWithReader processes all pending push notifications on a connection
+// This method should be called by the client in WithReader before reading the reply
+func (c *baseClient) processPendingPushNotificationWithReader(ctx context.Context, cn *pool.Conn, rd *proto.Reader) error {
+	// if we have the reader, we don't need to check for data on the socket, we are waiting
+	// for either a reply or a push notification, so we can block until we get a reply or reach the timeout
+	if c.opt.Protocol != 3 || c.pushProcessor == nil {
+		return nil
+	}
+
+	// Create handler context with client, connection pool, and connection information
+	handlerCtx := c.pushNotificationHandlerContext(cn)
+	return c.pushProcessor.ProcessPendingNotifications(ctx, handlerCtx, rd)
+}
+
+// pushNotificationHandlerContext creates a handler context for push notification processing
+func (c *baseClient) pushNotificationHandlerContext(cn *pool.Conn) push.NotificationHandlerContext {
+	return push.NotificationHandlerContext{
+		Client:   c,
+		ConnPool: c.connPool,
+		Conn:     cn, // Wrap in adapter for easier interface access
+	}
+}