fix: skip ROLLBACK when the socket is already dead

_reset_connection issued ROLLBACK blindly, so a half-closed TCP socket
(LB idle reset, peer crash, etc.) took self._timeout (default 10 s) to
discover — paid on every pool release through a dead connection.

Add a cheap pre-write liveness check: if writer.transport.is_closing()
or reader.at_eof() return True, bail early with False. Use isinstance
bool checks so mocked attributes don't false-trigger.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqliteclient/pool.py

@@ -11,6 +11,30 @@
 from dqliteclient.exceptions import DqliteConnectionError
 
 
+def _socket_looks_dead(conn: DqliteConnection) -> bool:
+    """Best-effort local detection of a half-closed TCP socket.
+
+    Returns True only on an affirmative bool signal from the transport or
+    reader. Mocked / missing attributes default to False (assume alive) so
+    the check never produces a false positive against well-behaved peers.
+    """
+    protocol = conn._protocol
+    if protocol is None:
+        return True
+    writer = getattr(protocol, "_writer", None)
+    reader = getattr(protocol, "_reader", None)
+    transport = getattr(writer, "transport", None) if writer is not None else None
+    try:
+        closing = transport.is_closing() if transport is not None else False
+    except Exception:
+        closing = False
+    try:
+        eof = reader.at_eof() if reader is not None else False
+    except Exception:
+        eof = False
+    return (isinstance(closing, bool) and closing) or (isinstance(eof, bool) and eof)
+
+
 class ConnectionPool:
     """Connection pool with automatic leader detection.
 
@@ -181,6 +205,11 @@ async def _reset_connection(self, conn: DqliteConnection) -> bool:
         False if it should be destroyed.
         """
         if conn._in_transaction:
+            # Cheap pre-write liveness check: if the transport is already
+            # closing or the reader has seen EOF, ROLLBACK would stall on
+            # _read_data until self._timeout. Bail fast instead.
+            if _socket_looks_dead(conn):
+                return False
             try:
                 await conn.execute("ROLLBACK")
             except BaseException:

tests/test_pool.py

@@ -692,6 +692,50 @@ async def exec_or_cancel(db_id, sql, params=None):
 
         await pool.close()
 
+    async def test_reset_connection_skips_rollback_on_dead_socket(self) -> None:
+        """If the transport is already closing / EOF, _reset_connection must
+        return False without waiting for a read timeout. Otherwise every
+        release through a half-closed socket costs self._timeout seconds.
+        """
+        import asyncio
+
+        from dqliteclient.connection import DqliteConnection
+
+        pool = ConnectionPool(["localhost:9001"], max_size=1, timeout=10.0)
+
+        conn = DqliteConnection("127.0.0.1:9001")
+        conn._protocol = MagicMock()
+        conn._db_id = 1
+        conn._bound_loop = asyncio.get_running_loop()
+        conn._in_transaction = True
+
+        # Reader has seen EOF / transport is closing.
+        dead_reader = MagicMock()
+        dead_reader.at_eof = MagicMock(return_value=True)
+        dead_transport = MagicMock()
+        dead_transport.is_closing = MagicMock(return_value=True)
+        dead_writer = MagicMock()
+        dead_writer.transport = dead_transport
+        conn._protocol._reader = dead_reader
+        conn._protocol._writer = dead_writer
+
+        exec_called = False
+
+        async def track_exec(*args, **kwargs):
+            nonlocal exec_called
+            exec_called = True
+            # Never resolves — we'd pay self._timeout if reached.
+            await asyncio.sleep(1000)
+            return (0, 0)
+
+        conn._protocol.exec_sql = track_exec
+        conn._protocol.close = MagicMock()
+        conn._protocol.wait_closed = AsyncMock()
+
+        result = await asyncio.wait_for(pool._reset_connection(conn), timeout=1.0)
+        assert result is False
+        assert not exec_called, "ROLLBACK must not be sent on a dead socket"
+
     async def test_reset_connection_returns_false_on_cancelled_error(self) -> None:
         """_reset_connection must return False (not raise) when ROLLBACK is cancelled."""
         import asyncio