fix: protect connect() with _in_use guard to prevent TOCTOU race

connect() was not protected by the _in_use re-entrance guard added in
#077. Two coroutines calling connect() on the same DqliteConnection
could both pass the `if self._protocol is not None: return` check,
both open TCP connections, and the second would overwrite
self._protocol, permanently leaking the first connection's socket.

Add _check_in_use() and _in_use flag to connect(), consistent with
the existing pattern in execute/fetch/fetchall/fetchval. This turns a
silent socket leak into a loud InterfaceError.

Closes #091

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -87,30 +87,39 @@ def is_connected(self) -> bool:
 
     async def connect(self) -> None:
         """Establish connection to the database."""
+        self._check_in_use()
         if self._protocol is not None:
             return
 
-        host, port = _parse_address(self._address)
-
-        try:
-            reader, writer = await asyncio.wait_for(
-                asyncio.open_connection(host, port),
-                timeout=self._timeout,
-            )
-        except TimeoutError as e:
-            raise DqliteConnectionError(f"Connection to {self._address} timed out") from e
-        except OSError as e:
-            raise DqliteConnectionError(f"Failed to connect to {self._address}: {e}") from e
-
-        self._protocol = DqliteProtocol(reader, writer, timeout=self._timeout)
-
+        self._in_use = True
         try:
-            await self._protocol.handshake()
-            self._db_id = await self._protocol.open_database(self._database)
-        except BaseException:
-            self._protocol.close()
-            self._protocol = None
-            raise
+            host, port = _parse_address(self._address)
+
+            try:
+                reader, writer = await asyncio.wait_for(
+                    asyncio.open_connection(host, port),
+                    timeout=self._timeout,
+                )
+            except TimeoutError as e:
+                raise DqliteConnectionError(
+                    f"Connection to {self._address} timed out"
+                ) from e
+            except OSError as e:
+                raise DqliteConnectionError(
+                    f"Failed to connect to {self._address}: {e}"
+                ) from e
+
+            self._protocol = DqliteProtocol(reader, writer, timeout=self._timeout)
+
+            try:
+                await self._protocol.handshake()
+                self._db_id = await self._protocol.open_database(self._database)
+            except BaseException:
+                self._protocol.close()
+                self._protocol = None
+                raise
+        finally:
+            self._in_use = False
 
     async def close(self) -> None:
         """Close the connection."""

tests/test_connection.py

@@ -675,3 +675,54 @@ async def second_execute():
 
         assert len(errors) == 1
         assert "another operation is in progress" in str(errors[0])
+
+    async def test_concurrent_connect_raises_interface_error(self) -> None:
+        """Two coroutines connecting the same object must raise InterfaceError."""
+        import asyncio
+
+        from dqliteclient.exceptions import InterfaceError
+
+        conn = DqliteConnection("localhost:9001")
+
+        gate = asyncio.Event()
+
+        async def slow_open(*args, **kwargs):
+            await gate.wait()
+            reader = AsyncMock()
+            writer = MagicMock()
+            writer.drain = AsyncMock()
+            writer.close = MagicMock()
+            writer.wait_closed = AsyncMock()
+            return reader, writer
+
+        errors: list[Exception] = []
+
+        async def first_connect():
+            with (
+                patch("asyncio.open_connection", side_effect=slow_open),
+                patch("dqliteclient.connection.DqliteProtocol") as MockProto,
+            ):
+                proto = MagicMock()
+                proto.handshake = AsyncMock()
+                proto.open_database = AsyncMock(return_value=1)
+                proto.close = MagicMock()
+                MockProto.return_value = proto
+                await conn.connect()
+
+        async def second_connect():
+            await asyncio.sleep(0)  # Let first_connect start
+            try:
+                await conn.connect()
+            except InterfaceError as e:
+                errors.append(e)
+
+        task1 = asyncio.create_task(first_connect())
+        task2 = asyncio.create_task(second_connect())
+
+        await asyncio.sleep(0)  # Let both start
+        gate.set()
+
+        await asyncio.gather(task1, task2, return_exceptions=True)
+
+        assert len(errors) == 1
+        assert "another operation is in progress" in str(errors[0])