fix: add event-loop affinity detection and thread-safety documentation

DqliteConnection now captures the event loop at connect() time and
verifies it in _check_in_use(). Using a connection from a different
event loop (e.g., from another OS thread running its own loop) raises
InterfaceError with a clear message instead of silently corrupting the
protocol stream.

Also adds thread-safety documentation to DqliteConnection, ConnectionPool,
and the module docstring, clearly stating that these classes are not
thread-safe and must be used within a single event loop.

Fixes #094

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

Author: antoineleclair

src/dqliteclient/__init__.py

@@ -1,4 +1,10 @@
-"""Async Python client for dqlite."""
+"""Async Python client for dqlite.
+
+Thread safety: connections and pools are NOT thread-safe. All operations
+must be performed within a single asyncio event loop. To submit work from
+other threads, use ``asyncio.run_coroutine_threadsafe()``. Free-threaded
+Python (no-GIL) is not supported.
+"""
 
 from dqliteclient.cluster import ClusterClient
 from dqliteclient.connection import DqliteConnection

src/dqliteclient/connection.py

@@ -47,7 +47,15 @@ def _parse_address(address: str) -> tuple[str, int]:
 
 
 class DqliteConnection:
-    """High-level async connection to a dqlite database."""
+    """High-level async connection to a dqlite database.
+
+    Thread safety: this class is NOT thread-safe. All operations must be
+    performed within a single asyncio event loop. Do not share instances
+    across OS threads or event loops. To submit work from other threads,
+    use ``asyncio.run_coroutine_threadsafe()`` — the coroutines execute
+    safely in the event loop thread. Free-threaded Python (no-GIL) is
+    not supported.
+    """
 
     def __init__(
         self,
@@ -72,6 +80,7 @@ def __init__(
         self._db_id: int | None = None
         self._in_transaction = False
         self._in_use = False
+        self._bound_loop: asyncio.AbstractEventLoop | None = None
 
     @property
     def address(self) -> str:
@@ -89,6 +98,7 @@ async def connect(self) -> None:
         if self._protocol is not None:
             return
 
+        self._bound_loop = asyncio.get_running_loop()
         self._in_use = True
         try:
             host, port = _parse_address(self._address)
@@ -138,7 +148,19 @@ def _ensure_connected(self) -> tuple[DqliteProtocol, int]:
         return self._protocol, self._db_id
 
     def _check_in_use(self) -> None:
-        """Raise if another coroutine is using this connection."""
+        """Raise on misuse: wrong event loop or concurrent coroutine access."""
+        if self._bound_loop is not None:
+            try:
+                current_loop = asyncio.get_running_loop()
+            except RuntimeError:
+                raise InterfaceError(
+                    "DqliteConnection must be used from within an async context."
+                ) from None
+            if current_loop is not self._bound_loop:
+                raise InterfaceError(
+                    "DqliteConnection is bound to a different event loop. "
+                    "Do not share connections across event loops or OS threads."
+                )
         if self._in_use:
             raise InterfaceError(
                 "Cannot perform operation: another operation is in progress on this "

src/dqliteclient/pool.py

@@ -12,7 +12,15 @@
 
 
 class ConnectionPool:
-    """Connection pool with automatic leader detection."""
+    """Connection pool with automatic leader detection.
+
+    Thread safety: this class is NOT thread-safe. All operations must be
+    performed within a single asyncio event loop. Do not share pool
+    instances across OS threads or event loops. To submit work from other
+    threads, use ``asyncio.run_coroutine_threadsafe()`` — the coroutines
+    execute safely in the event loop thread. Free-threaded Python (no-GIL)
+    is not supported.
+    """
 
     def __init__(
         self,

tests/test_connection.py

@@ -842,3 +842,60 @@ async def tx_b():
             f"got {type(errors[0]).__name__}: {errors[0]}"
         )
         assert "Nested" in str(errors[0]) or "nested" in str(errors[0])
+
+    async def test_cross_event_loop_raises_interface_error(self) -> None:
+        """Using a connection from a different event loop must raise InterfaceError."""
+        import asyncio
+        import threading
+
+        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, ResultResponse, 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()
+
+        # Now try to use the connection from a different event loop in another thread
+        error_from_thread: Exception | None = None
+
+        def run_in_other_loop():
+            nonlocal error_from_thread
+
+            async def use_conn():
+                # Provide a fresh response so the operation could succeed
+                # if the guard doesn't catch it
+                mock_reader.read.side_effect = [
+                    ResultResponse(last_insert_id=0, rows_affected=0).encode(),
+                ]
+                await conn.execute("SELECT 1")
+
+            try:
+                asyncio.run(use_conn())
+            except InterfaceError as e:
+                error_from_thread = e
+            except Exception as e:
+                error_from_thread = e
+
+        thread = threading.Thread(target=run_in_other_loop)
+        thread.start()
+        thread.join(timeout=5)
+
+        assert error_from_thread is not None, "Expected InterfaceError from cross-loop access"
+        assert isinstance(error_from_thread, InterfaceError), (
+            f"Expected InterfaceError, got {type(error_from_thread).__name__}: {error_from_thread}"
+        )
+        assert "event loop" in str(error_from_thread).lower()