fix(tests): debug logger

Author: ndyakov

internal/pool/pool.go

@@ -392,7 +392,9 @@ func (p *ConnPool) Put(ctx context.Context, cn *Conn) {
 			}
 		}
 		// For non-RESP3 or data that is not a push notification, buffered data is unexpected
-		internal.Logger.Printf(ctx, "Conn has unread data")
+		internal.Logger.Printf(ctx, "Conn has unread data: %d bytes, closing it", cn.rd.Buffered())
+		repl, err := cn.rd.ReadReply()
+		internal.Logger.Printf(ctx, "Data: %v, ERR: %v", repl, err)
 		p.Remove(ctx, cn, BadConnError{})
 		return
 	}

push/processor.go

@@ -52,7 +52,7 @@ func (p *Processor) ProcessPendingNotifications(ctx context.Context, handlerCtx
 		return nil
 	}
 
-	for {
+	for rd.Buffered() > 0 {
 		// Check if there's data available to read
 		replyType, err := rd.PeekReplyType()
 		if err != nil {
@@ -130,36 +130,37 @@ func (v *VoidProcessor) UnregisterHandler(pushNotificationName string) error {
 // This avoids unnecessary buffer scanning overhead.
 func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, _ NotificationHandlerContext, rd *proto.Reader) error {
 	// read and discard all push notifications
-	if rd != nil {
-		for {
-			replyType, err := rd.PeekReplyType()
-			if err != nil {
-				// No more data available or error reading
-				break
-			}
+	if rd == nil {
+		return nil
+	}
+	for rd.Buffered() > 0 {
+		replyType, err := rd.PeekReplyType()
+		if err != nil {
+			// No more data available or error reading
+			break
+		}
 
-			// Only process push notifications (arrays starting with >)
-			if replyType != proto.RespPush {
-				break
-			}
-			// see if we should skip this notification
-			notificationName, err := rd.PeekPushNotificationName()
-			if err != nil {
+		// Only process push notifications (arrays starting with >)
+		if replyType != proto.RespPush {
+			break
+		}
+		// see if we should skip this notification
+		notificationName, err := rd.PeekPushNotificationName()
+		if err != nil {
+			break
+		}
+		if shouldSkipNotification(notificationName) {
+			// discard the notification
+			if err := rd.DiscardNext(); err != nil {
 				break
 			}
-			if shouldSkipNotification(notificationName) {
-				// discard the notification
-				if err := rd.DiscardNext(); err != nil {
-					break
-				}
-				continue
-			}
+			continue
+		}
 
-			// Read the push notification
-			_, err = rd.ReadReply()
-			if err != nil {
-				return nil
-			}
+		// Read the push notification
+		_, err = rd.ReadReply()
+		if err != nil {
+			return nil
 		}
 	}
 	return nil
@@ -170,7 +171,7 @@ func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, _ Notific
 func shouldSkipNotification(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