fix(peek): non-blocking peek

Author: ndyakov

internal/pool/conn.go

@@ -58,6 +58,10 @@ func (cn *Conn) SetNetConn(netConn net.Conn) {
 	cn.bw.Reset(netConn)
 }
 
+func (cn *Conn) GetNetConn() net.Conn {
+	return cn.netConn
+}
+
 func (cn *Conn) Write(b []byte) (int, error) {
 	return cn.netConn.Write(b)
 }

internal/pool/pool.go

@@ -384,6 +384,8 @@ func (p *ConnPool) Put(ctx context.Context, cn *Conn) {
 	if cn.rd.Buffered() > 0 {
 		// Check if this might be push notification data
 		if p.cfg.Protocol == 3 {
+			// we know that there is something in the buffer, so peek at the next reply type without
+			// the potential to block
 			if replyType, err := cn.rd.PeekReplyType(); err == nil && replyType == proto.RespPush {
 				// For push notifications, we allow some buffered data
 				// The client will process these notifications before using the connection
@@ -546,6 +548,8 @@ func (p *ConnPool) isHealthyConn(cn *Conn) bool {
 		// However, push notification processing is now handled by the client
 		// before WithReader to ensure proper context is available to handlers
 		if err == errUnexpectedRead && p.cfg.Protocol == 3 {
+			// we know that there is something in the buffer, so peek at the next reply type without
+			// the potential to block
 			if replyType, err := cn.rd.PeekReplyType(); err == nil && replyType == proto.RespPush {
 				// For RESP3 connections with push notifications, we allow some buffered data
 				// The client will process these notifications before using the connection

push/processor.go

@@ -2,6 +2,7 @@ package push
 
 import (
 	"context"
+	"time"
 
 	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/proto"
@@ -51,8 +52,19 @@ func (p *Processor) ProcessPendingNotifications(ctx context.Context, handlerCtx
 	if rd == nil {
 		return nil
 	}
+	conn := handlerCtx.Conn
+	if conn == nil {
+		return nil
+	}
+	netConn := handlerCtx.Conn.GetNetConn()
+	if netConn == nil {
+		return nil
+	}
 
 	for {
+		// Set a short read deadline to check for available data
+		// otherwise we may block on Peek if there is no data available
+		netConn.SetReadDeadline(time.Now().Add(1))
 		// Check if there's data available to read
 		replyType, err := rd.PeekReplyType()
 		if err != nil {
@@ -104,6 +116,7 @@ func (p *Processor) ProcessPendingNotifications(ctx context.Context, handlerCtx
 		}
 	}
 
+	netConn.SetReadDeadline(time.Time{})
 	return nil
 }
 
@@ -133,12 +146,23 @@ func (v *VoidProcessor) UnregisterHandler(pushNotificationName string) error {
 // ProcessPendingNotifications for VoidProcessor does nothing since push notifications
 // are only available in RESP3 and this processor is used for RESP2 connections.
 // This avoids unnecessary buffer scanning overhead.
-func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, _ NotificationHandlerContext, rd *proto.Reader) error {
+func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, handlerCtx NotificationHandlerContext, rd *proto.Reader) error {
 	// read and discard all push notifications
 	if rd == nil {
 		return nil
 	}
+	conn := handlerCtx.Conn
+	if conn == nil {
+		return nil
+	}
+	netConn := handlerCtx.Conn.GetNetConn()
+	if netConn == nil {
+		return nil
+	}
 	for {
+		// Set a short read deadline to check for available data
+		netConn.SetReadDeadline(time.Now().Add(1))
+		// Check if there's data available to read
 		replyType, err := rd.PeekReplyType()
 		if err != nil {
 			// No more data available or error reading
@@ -166,6 +190,7 @@ func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, _ Notific
 			return nil
 		}
 	}
+	netConn.SetReadDeadline(time.Time{})
 	return nil
 }
 
@@ -174,7 +199,7 @@ func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, _ Notific
 func willHandleNotificationInClient(notificationType string) bool {
 	switch notificationType {
 	// Pub/Sub notifications - handled by pub/sub system
-	case "message",     // Regular pub/sub message
+	case "message", // Regular pub/sub message
 		"pmessage",     // Pattern pub/sub message
 		"subscribe",    // Subscription confirmation
 		"unsubscribe",  // Unsubscription confirmation

redis.go

@@ -733,13 +733,6 @@ func (c *baseClient) txPipelineProcessCmds(
 	}
 
 	if err := cn.WithReader(c.context(ctx), c.opt.ReadTimeout, 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 {
-			// Log the error but don't fail the command execution
-			// Push notification processing errors shouldn't break normal Redis operations
-			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
-		}
-
 		statusCmd := cmds[0].(*StatusCmd)
 		// Trim multi and exec.
 		trimmedCmds := cmds[1 : len(cmds)-1]