Clean up raw_conn if eager connect fails in the async dialect

DqliteDialect_aio.connect() calls self.loaded_dbapi.connect() to
build an AsyncConnection, then drives its eager TCP connect via
await_only(raw_conn.connect()). If that eager step raised, raw_conn
was already constructed (holding references to loop locks and
partially-initialised state) and leaked until GC. In
SQLAlchemy's connection-pool context a handful of those per leader
flap added up fast.

Wrap the eager connect in try/except BaseException, close raw_conn
inside a narrow suppress(Exception), then re-raise. Using
BaseException means cancellation during connect (parent
asyncio.timeout(), TaskGroup cancel) also triggers cleanup and
still propagates.

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

Author: antoineleclair

src/sqlalchemydqlite/aio.py

@@ -253,10 +253,25 @@ def connect(self, *cargs: Any, **cparams: Any) -> Any:
         """Create and wrap an async connection.
 
         Eagerly establishes the TCP connection so errors surface at
-        connect-time rather than on the first query.
+        connect-time rather than on the first query. If that eager
+        connect raises, the ``raw_conn`` object is already constructed
+        and holds references to loop locks / partially-initialised
+        state — without explicit cleanup it leaks until GC and can
+        linger on the event loop it was bound to. Call ``close()`` on
+        ``BaseException`` so cancellation (e.g. a parent
+        ``asyncio.timeout()`` firing during connect) also cleans up,
+        then re-raise unchanged. ``close()`` is documented as
+        idempotent and safe to call even when no TCP connection
+        landed, so the suppression around it is narrow
+        (``Exception`` only; cancellation signals still propagate).
         """
         raw_conn = self.loaded_dbapi.connect(*cargs, **cparams)
-        await_only(raw_conn.connect())
+        try:
+            await_only(raw_conn.connect())
+        except BaseException:
+            with contextlib.suppress(Exception):
+                await_only(raw_conn.close())
+            raise
         return AsyncAdaptedConnection(raw_conn)
 
     def get_driver_connection(self, connection: Any) -> Any:

tests/test_async_adapter.py

@@ -242,6 +242,67 @@ def test_close_propagates_programming_error_from_rollback(self) -> None:
             adapted.close()
 
 
+class TestAioDialectConnectCleanup:
+    """DqliteDialect_aio.connect() eagerly establishes the TCP
+    connection. If that raises, raw_conn (already constructed) must
+    be closed — otherwise the partial connection leaks. Cleanup must
+    also fire on cancellation so a parent asyncio.timeout() firing
+    mid-connect doesn't leave the raw_conn dangling.
+    """
+
+    def test_close_is_called_when_eager_connect_fails(self) -> None:
+        import pytest
+
+        from dqliteclient.exceptions import DqliteConnectionError
+        from sqlalchemydqlite.aio import DqliteDialect_aio
+
+        dialect = DqliteDialect_aio()
+
+        raw_conn = MagicMock()
+        raw_conn.connect = MagicMock(side_effect=DqliteConnectionError("probe failed"))
+        raw_conn.close = MagicMock()
+
+        dbapi = MagicMock()
+        dbapi.connect = MagicMock(return_value=raw_conn)
+        dialect.loaded_dbapi = dbapi
+
+        with (
+            patch("sqlalchemydqlite.aio.await_only", side_effect=_run_sync),
+            pytest.raises(DqliteConnectionError, match="probe failed"),
+        ):
+            dialect.connect()
+
+        raw_conn.close.assert_called_once()
+
+    def test_close_is_called_on_cancellation(self) -> None:
+        """CancelledError during eager connect must trigger cleanup
+        AND still propagate — callers rely on structured-concurrency
+        signal fidelity."""
+        import asyncio
+
+        import pytest
+
+        from sqlalchemydqlite.aio import DqliteDialect_aio
+
+        dialect = DqliteDialect_aio()
+
+        raw_conn = MagicMock()
+        raw_conn.connect = MagicMock(side_effect=asyncio.CancelledError())
+        raw_conn.close = MagicMock()
+
+        dbapi = MagicMock()
+        dbapi.connect = MagicMock(return_value=raw_conn)
+        dialect.loaded_dbapi = dbapi
+
+        with (
+            patch("sqlalchemydqlite.aio.await_only", side_effect=_run_sync),
+            pytest.raises(asyncio.CancelledError),
+        ):
+            dialect.connect()
+
+        raw_conn.close.assert_called_once()
+
+
 class TestAioAllExports:
     """Adapter classes are part of the supported public surface."""