Fix driver stuck RecursionError in COMMIT SUCCESS

The driver could get stuck when the result summary was so deeply nested that it
caused an `RecursionError`. It would also get stuck on certain protocol
violations in the same place (e.g., unknown PackStream tag). However, the
latter is much more unlikely to occur.

This patch will make the driver ditch the connection when it's not able to
properly read a message for unexpected reasons.

Author: robsdedude

src/neo4j/_async/io/_bolt.py

@@ -35,6 +35,7 @@
 from ..._exceptions import (
     BoltError,
     BoltHandshakeError,
+    SocketDeadlineExceededError,
 )
 from ..._io import BoltProtocolVersion
 from ..._meta import USER_AGENT
@@ -889,6 +890,15 @@ async def _set_defunct_write(self, error=None, silent=False):
     async def _set_defunct(self, message, error=None, silent=False):
         direct_driver = getattr(self.pool, "is_direct_pool", False)
         user_cancelled = isinstance(error, asyncio.CancelledError)
+        connection_failed = isinstance(
+            error,
+            (
+                ServiceUnavailable,
+                SessionExpired,
+                OSError,
+                SocketDeadlineExceededError,
+            ),
+        )
 
         if not (user_cancelled or self._closing):
             log_call = log.error
@@ -915,6 +925,12 @@ async def _set_defunct(self, message, error=None, silent=False):
         if user_cancelled:
             self.kill()
             raise error  # cancellation error should not be re-written
+        if not connection_failed:
+            # Something else but the connection failed
+            # => we're not sure which state we're in
+            # => ditch the connection and raise the error for user-awareness
+            await self.close()
+            raise error
         if not self._closing:
             # If we fail while closing the connection, there is no need to
             # remove the connection from the pool, nor to try to close the

src/neo4j/_async/io/_common.py

@@ -81,6 +81,15 @@ async def pop(self, hydration_hooks):
                 self._unpacker.unpack(hydration_hooks) for _ in range(size)
             ]
             return tag, fields
+        except Exception as error:
+            log.debug(
+                "[#%04X]  _: Failed to unpack response: %r",
+                self._local_port,
+                error,
+            )
+            self._broken = True
+            await AsyncUtil.callback(self.on_error, error)
+            raise
         finally:
             # Reset for new message
             self._unpacker.reset()

src/neo4j/_async/work/result.py

@@ -26,10 +26,6 @@
     warn_explicit,
 )
 
-
-if t.TYPE_CHECKING:
-    import typing_extensions as te
-
 from ..._api import (
     NotificationCategory,
     NotificationMinimumSeverity,
@@ -65,6 +61,7 @@
 
 if t.TYPE_CHECKING:
     import pandas  # type: ignore[import]
+    import typing_extensions as te
 
     from ..._addressing import Address
     from ...graph import Graph