fix(websocket): Raise IOException on closed connection instead of returning b"" (#1213)

* fix(websocket): Raise IOException on closed connection instead of returning b""

Fixes #1212

- WebSocket read() now raises IOException when connection is closed,
  allowing read_exactly() to immediately detect closure instead of
  retrying 100 times on empty bytes
- Yamux send_window_update() gracefully handles connection closure
  errors during window updates (data was already read successfully)
- Improved _extract_close_info() fallback for non-standard exceptions
- Added exception chaining (from e) in write() for better tracebacks
- Comprehensive test suite for closure handling scenarios

Co-authored-by: Cursor <cursoragent@cursor.com>

* refactor(exceptions): Replace fragile string matching with typed ConnectionClosedError

Add ConnectionClosedError(IOException) with structured close_code,
close_reason, and transport attributes.  The WebSocket transport now
raises this typed exception instead of a plain IOException with the
close info buried in the message string.  Yamux catches it by type
(`except ConnectionClosedError`) instead of parsing `str(e).lower()`
for closure keywords — eliminating a fragile pattern that would
silently break if anyone changed an error message.

A string-matching fallback is retained for transports that don't yet
raise ConnectionClosedError (e.g. TCP RawConnError).

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: acul71

libp2p/io/exceptions.py

@@ -7,6 +7,28 @@ class IOException(BaseLibp2pError):
     pass
 
 
+class ConnectionClosedError(IOException):
+    """
+    Raised when a connection is closed by the peer.
+
+    Carries structured close information so that upstream code can make
+    decisions based on exception type (``except ConnectionClosedError``)
+    rather than fragile string matching on the message.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        close_code: int | None = None,
+        close_reason: str = "",
+        transport: str = "",
+    ) -> None:
+        super().__init__(message)
+        self.close_code = close_code
+        self.close_reason = close_reason
+        self.transport = transport
+
+
 class IncompleteReadError(IOException):
     """Fewer bytes were read than requested."""
 

libp2p/stream_muxer/yamux/yamux.py

@@ -31,7 +31,9 @@
     ISecureConn,
 )
 from libp2p.io.exceptions import (
+    ConnectionClosedError,
     IncompleteReadError,
+    IOException,
 )
 from libp2p.io.utils import (
     read_exactly,
@@ -161,6 +163,11 @@ async def send_window_update(self, increment: int, skip_lock: bool = False) -> N
         param:skip_lock (bool): If True, skips acquiring window_lock.
         This should only be used when calling from a context
         that already holds the lock.
+
+        Note: This method gracefully handles connection closure errors.
+        If the connection is closed (e.g., peer closed WebSocket immediately
+        after sending data), the window update will fail silently, allowing
+        the read operation to complete successfully.
         """
         if increment <= 0:
             # If increment is zero or negative, skip sending update
@@ -182,7 +189,42 @@ async def _do_window_update() -> None:
                 self.stream_id,
                 increment,
             )
-            await self.conn.secured_conn.write(header)
+            try:
+                await self.conn.secured_conn.write(header)
+            except ConnectionClosedError as e:
+                # Typed exception from transports (e.g., WebSocket) that
+                # properly signal connection closure — handle gracefully.
+                # Connection may be closed by peer (e.g., WebSocket closed
+                # immediately after sending data, as seen with Nim).
+                # This is acceptable — the data was already read successfully.
+                logger.debug(
+                    f"Stream {self.stream_id}: Window update failed due to "
+                    f"connection closure (data was already read): {e}"
+                )
+                return
+            except (RawConnError, IOException) as e:
+                # Fallback for transports that don't yet raise
+                # ConnectionClosedError (e.g., TCP RawConnError).
+                error_str = str(e).lower()
+                if any(
+                    keyword in error_str
+                    for keyword in [
+                        "connection closed",
+                        "closed by peer",
+                        "connection is closed",
+                    ]
+                ):
+                    logger.debug(
+                        f"Stream {self.stream_id}: Window update failed due to "
+                        f"connection closure (data was already read): {e}"
+                    )
+                    return
+                # Re-raise if it's a different error we don't expect
+                logger.warning(
+                    f"Stream {self.stream_id}: Unexpected error during "
+                    f"window update: {e}"
+                )
+                raise
 
         if skip_lock:
             await _do_window_update()

libp2p/transport/websocket/connection.py

@@ -10,7 +10,7 @@
 from trio_websocket import WebSocketConnection
 
 from libp2p.io.abc import ReadWriteCloser
-from libp2p.io.exceptions import IOException
+from libp2p.io.exceptions import ConnectionClosedError, IOException
 
 logger = logging.getLogger(__name__)
 
@@ -194,31 +194,40 @@ def _extract_close_info(self, e: Exception) -> tuple[int | None, str]:
             Tuple of (close_code, close_reason)
 
         """
-        # ConnectionClosed has a 'reason' attribute which is a CloseReason object
+        # ConnectionClosed has a 'reason' attribute which is a CloseReason object.
+        # Some exceptions (like mocks in tests) may have code/reason directly.
         close_reason_obj = getattr(e, "reason", None)
-        if close_reason_obj is not None:
+
+        # Check if reason is a CloseReason object (has 'code' attribute)
+        if close_reason_obj is not None and hasattr(close_reason_obj, "code"):
             close_code = getattr(close_reason_obj, "code", None)
             close_reason = (
                 getattr(close_reason_obj, "reason", None) or "Connection closed by peer"
             )
         else:
-            # Fallback if reason is not available
-            close_code = None
-            close_reason = "Connection closed by peer"
+            # Fallback: check if code and reason are directly on the exception
+            # (for mock exceptions in tests or other exception types)
+            close_code = getattr(e, "code", None)
+            close_reason = getattr(e, "reason", None) or "Connection closed by peer"
         return close_code, close_reason
 
     def _handle_connection_closed_exception(
         self, e: Exception, operation: str = "read"
-    ) -> IOException:
+    ) -> ConnectionClosedError:
         """
-        Handle a connection closure exception by creating an appropriate IOException.
+        Handle a connection closure exception by creating a ConnectionClosedError.
+
+        Returns a ``ConnectionClosedError`` (subclass of ``IOException``) that
+        carries the close code and reason as structured attributes.  This lets
+        upstream code (e.g. yamux) catch connection closures by *type* instead
+        of doing fragile string matching on the error message.
 
         Args:
             e: The exception that indicates connection closure
             operation: The operation that was being performed (read/write)
 
         Returns:
-            IOException with detailed information about the closure
+            ConnectionClosedError with close_code and close_reason attributes
 
         """
         self._closed = True
@@ -227,13 +236,17 @@ def _handle_connection_closed_exception(
             f"WebSocket connection closed during {operation}: "
             f"code={close_code}, reason={close_reason}"
         )
-        # Return IOException to be raised by caller
-        # This allows read_exactly() to immediately detect connection closure
-        # instead of returning empty bytes which would cause retries
-        return IOException(
+        # Return ConnectionClosedError (subclass of IOException) to be raised
+        # by caller.  This allows read_exactly() to immediately detect
+        # connection closure instead of returning empty bytes which would cause
+        # retries, and lets yamux catch the typed exception directly.
+        return ConnectionClosedError(
             f"WebSocket connection closed by peer during "
             f"{operation} operation: code={close_code}, "
-            f"reason={close_reason}."
+            f"reason={close_reason}.",
+            close_code=close_code,
+            close_reason=close_reason,
+            transport="websocket",
         )
 
     async def _start_keepalive(self) -> None:
@@ -284,8 +297,10 @@ async def read(self, n: int | None = None) -> bytes:
 
         """
         if self._closed:
-            # Return empty bytes to signal EOF (like TCP does)
-            return b""
+            # Raise IOException immediately when connection is closed.
+            # This allows read_exactly() to detect closure immediately
+            # instead of retrying up to 100 times on empty bytes.
+            raise IOException("Connection is closed")
 
         async with self._read_lock:
             try:
@@ -427,6 +442,9 @@ async def read(self, n: int | None = None) -> bytes:
                 # Re-raise IOException as-is (already has proper context)
                 raise
             except Exception as e:
+                # Handle connection closure missed by inner handlers
+                if self._is_connection_closed_exception(e):
+                    raise self._handle_connection_closed_exception(e, "read")
                 logger.error(f"WebSocket read failed: {e}")
                 raise IOException(f"Read failed: {str(e)}")
 
@@ -464,7 +482,7 @@ async def write(self, data: bytes) -> None:
                     raise self._handle_connection_closed_exception(e, "write")
                 logger.error(f"WebSocket write failed: {e}")
                 self._closed = True
-                raise IOException(f"Write failed: {str(e)}")
+                raise IOException(f"Write failed: {str(e)}") from e
 
     async def close(self) -> None:
         """

newsfragments/1212.bugfix.rst

@@ -0,0 +1 @@
+Fixed WebSocket transport to immediately raise ``IOException`` on connection closure instead of returning empty bytes, preventing retry loops in ``read_exactly()``. Also added graceful error handling in yamux ``send_window_update()`` for connections closed by peers during window updates.