Fixing error handling of connection buffer purging of closed connecton. Enabling troubleshooting logging for maint notifications e2e tests. (#3971)

* Fixing error handling of connection buffer purging of closed connection. Enabling troubleshooting logging for maint notifications e2e tests.

* Fix race condition in RESP3 parser buffer purge (#3972)

* Initial plan

* Combine None check with try-except for buffer.purge() to handle race condition

Co-authored-by: petyaslavova <194077574+petyaslavova@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: petyaslavova <194077574+petyaslavova@users.noreply.github.com>

---------

Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: petyaslavova <194077574+petyaslavova@users.noreply.github.com>

Author: petyaslavova

redis/_parsers/resp3.py

@@ -29,17 +29,23 @@ def handle_pubsub_push_response(self, response):
         return response
 
     def read_response(self, disable_decoding=False, push_request=False):
-        pos = self._buffer.get_pos() if self._buffer else None
+        pos = self._buffer.get_pos() if self._buffer is not None else None
         try:
             result = self._read_response(
                 disable_decoding=disable_decoding, push_request=push_request
             )
         except BaseException:
-            if self._buffer:
+            if self._buffer is not None:
                 self._buffer.rewind(pos)
             raise
         else:
-            self._buffer.purge()
+            if self._buffer is not None:
+                try:
+                    self._buffer.purge()
+                except AttributeError:
+                    # Buffer may have been set to None by another thread after
+                    # the check above; result is still valid so we don't raise
+                    pass
             return result
 
     def _read_response(self, disable_decoding=False, push_request=False):

redis/cluster.py

@@ -1573,8 +1573,13 @@ def _execute_command(self, target_node, *args, **kwargs):
                 e.connection = target_node
                 raise
             except (ConnectionError, TimeoutError) as e:
-                # if we have a connection, use it, otherwise use the target node
-                # object which contains the host and port info
+                if is_debug_log_enabled():
+                    socket_address = self._extracts_socket_address(connection)
+                    args_log_str = truncate_text(" ".join(map(safe_str, args)))
+                    logger.debug(
+                        f"{type(e).__name__} received for command {args_log_str}, on node {target_node.name}, "
+                        f"and connection: {connection} using local socket address: {socket_address}, error: {e}"
+                    )
                 # this is used to report the metrics based on host and port info
                 e.connection = connection if connection else target_node
 

tests/test_scenario/test_maint_notifications.py

@@ -45,6 +45,10 @@
     datefmt="%Y-%m-%d %H:%M:%S:%f",
 )
 
+# Set DEBUG level for specific redis-py loggers
+logging.getLogger("redis.maint_notifications").setLevel(logging.DEBUG)
+logging.getLogger("redis.cluster").setLevel(logging.DEBUG)
+
 BIND_TIMEOUT = 60
 MIGRATE_TIMEOUT = 60
 FAILOVER_TIMEOUT = 15