fix: add re-entrance guard to prevent concurrent coroutine access

DqliteConnection now detects when two coroutines attempt to use the
same connection concurrently and raises InterfaceError immediately,
matching the behavior of asyncpg and other mature async DB clients.

The dqlite wire protocol is strictly request-response with no
pipelining. Without this guard, concurrent coroutines would interleave
their sends and reads on the shared StreamWriter/StreamReader, causing
response mix-ups, ProtocolErrors, or silent data corruption.

Each public method (execute, fetch, fetchall, fetchval) sets an
_in_use flag before the protocol operation and clears it after. If
a second coroutine tries to use the connection while _in_use is True,
InterfaceError is raised with a clear message directing users to use
ConnectionPool for concurrent operations.

Also adds InterfaceError to the exceptions module and public API.

Closes #077
Also resolves #079 (_in_transaction flag race) and #084 (concurrent
connect TOCTOU) since the guard prevents concurrent access entirely.

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

Author: antoineleclair

src/dqliteclient/__init__.py

@@ -6,6 +6,7 @@
     ClusterError,
     DqliteConnectionError,
     DqliteError,
+    InterfaceError,
     OperationalError,
     ProtocolError,
 )
@@ -23,6 +24,7 @@
     "MemoryNodeStore",
     "DqliteError",
     "DqliteConnectionError",
+    "InterfaceError",
     "ProtocolError",
     "ClusterError",
     "OperationalError",

src/dqliteclient/connection.py

@@ -6,7 +6,12 @@
 from contextlib import asynccontextmanager
 from typing import Any
 
-from dqliteclient.exceptions import DqliteConnectionError, OperationalError, ProtocolError
+from dqliteclient.exceptions import (
+    DqliteConnectionError,
+    InterfaceError,
+    OperationalError,
+    ProtocolError,
+)
 from dqliteclient.protocol import DqliteProtocol
 
 # dqlite error codes that indicate a leader change (SQLite extended error codes)
@@ -68,6 +73,7 @@ def __init__(
         self._protocol: DqliteProtocol | None = None
         self._db_id: int | None = None
         self._in_transaction = False
+        self._in_use = False
 
     @property
     def address(self) -> str:
@@ -127,6 +133,15 @@ def _ensure_connected(self) -> tuple[DqliteProtocol, int]:
             raise DqliteConnectionError("Not connected")
         return self._protocol, self._db_id
 
+    def _check_in_use(self) -> None:
+        """Raise if another coroutine is using this connection."""
+        if self._in_use:
+            raise InterfaceError(
+                "Cannot perform operation: another operation is in progress on this "
+                "connection. DqliteConnection does not support concurrent coroutine "
+                "access. Use a ConnectionPool to manage multiple concurrent operations."
+            )
+
     def _invalidate(self) -> None:
         """Mark the connection as broken after an unrecoverable error."""
         if self._protocol is not None:
@@ -141,7 +156,9 @@ async def execute(self, sql: str, params: Sequence[Any] | None = None) -> tuple[
 
         Returns (last_insert_id, rows_affected).
         """
+        self._check_in_use()
         protocol, db_id = self._ensure_connected()
+        self._in_use = True
         try:
             return await protocol.exec_sql(db_id, sql, params)
         except (DqliteConnectionError, ProtocolError):
@@ -154,10 +171,14 @@ async def execute(self, sql: str, params: Sequence[Any] | None = None) -> tuple[
         except BaseException:
             self._invalidate()
             raise
+        finally:
+            self._in_use = False
 
     async def fetch(self, sql: str, params: Sequence[Any] | None = None) -> list[dict[str, Any]]:
         """Execute a query and return results as list of dicts."""
+        self._check_in_use()
         protocol, db_id = self._ensure_connected()
+        self._in_use = True
         try:
             columns, rows = await protocol.query_sql(db_id, sql, params)
         except (DqliteConnectionError, ProtocolError):
@@ -170,11 +191,15 @@ async def fetch(self, sql: str, params: Sequence[Any] | None = None) -> list[dic
         except BaseException:
             self._invalidate()
             raise
+        finally:
+            self._in_use = False
         return [dict(zip(columns, row, strict=True)) for row in rows]
 
     async def fetchall(self, sql: str, params: Sequence[Any] | None = None) -> list[list[Any]]:
         """Execute a query and return results as list of lists."""
+        self._check_in_use()
         protocol, db_id = self._ensure_connected()
+        self._in_use = True
         try:
             _, rows = await protocol.query_sql(db_id, sql, params)
         except (DqliteConnectionError, ProtocolError):
@@ -187,6 +212,8 @@ async def fetchall(self, sql: str, params: Sequence[Any] | None = None) -> list[
         except BaseException:
             self._invalidate()
             raise
+        finally:
+            self._in_use = False
         return rows
 
     async def fetchone(
@@ -198,7 +225,9 @@ async def fetchone(
 
     async def fetchval(self, sql: str, params: Sequence[Any] | None = None) -> Any:
         """Execute a query and return the first column of the first row."""
+        self._check_in_use()
         protocol, db_id = self._ensure_connected()
+        self._in_use = True
         try:
             _, rows = await protocol.query_sql(db_id, sql, params)
         except (DqliteConnectionError, ProtocolError):
@@ -211,6 +240,8 @@ async def fetchval(self, sql: str, params: Sequence[Any] | None = None) -> Any:
         except BaseException:
             self._invalidate()
             raise
+        finally:
+            self._in_use = False
         if rows and rows[0]:
             return rows[0][0]
         return None

src/dqliteclient/exceptions.py

@@ -13,6 +13,10 @@ class ProtocolError(DqliteError):
     """Protocol-level error."""
 
 
+class InterfaceError(DqliteError):
+    """Misuse of the client interface (e.g. concurrent access on a connection)."""
+
+
 class ClusterError(DqliteError):
     """Cluster-related error (leader not found, etc)."""
 

tests/test_connection.py

@@ -617,3 +617,61 @@ async def test_not_leader_error_invalidates_connection(self) -> None:
 
         # Connection should be invalidated after a leader error
         assert not conn.is_connected
+
+    async def test_concurrent_coroutines_raises_interface_error(self) -> None:
+        """Two coroutines using the same connection must raise InterfaceError."""
+        import asyncio
+
+        from dqliteclient.exceptions import InterfaceError
+
+        conn = DqliteConnection("localhost:9001")
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        from dqlitewire.messages import DbResponse, WelcomeResponse
+
+        responses = [
+            WelcomeResponse(heartbeat_timeout=15000).encode(),
+            DbResponse(db_id=1).encode(),
+        ]
+        mock_reader.read.side_effect = responses
+
+        with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)):
+            await conn.connect()
+
+        # Make execute hang so two coroutines overlap
+        first_entered = asyncio.Event()
+
+        async def slow_exec_sql(db_id, sql, params=None):
+            first_entered.set()
+            await asyncio.sleep(10)
+            return (0, 1)
+
+        conn._protocol.exec_sql = AsyncMock(side_effect=slow_exec_sql)  # type: ignore[union-attr]
+
+        errors: list[Exception] = []
+
+        async def first_execute():
+            await conn.execute("INSERT INTO t VALUES (1)")
+
+        async def second_execute():
+            await first_entered.wait()
+            try:
+                await conn.execute("INSERT INTO t VALUES (2)")
+            except InterfaceError as e:
+                errors.append(e)
+
+        task1 = asyncio.create_task(first_execute())
+        task2 = asyncio.create_task(second_execute())
+
+        await task2  # second should raise InterfaceError
+        task1.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task1
+
+        assert len(errors) == 1
+        assert "another operation is in progress" in str(errors[0])