Pin AsyncConnection to the loop that first created its locks

The module docstring already warned that each AsyncConnection is
bound to the event loop it was first used on, but nothing enforced
it. Running the same instance on a second loop silently produced
asyncio's internal "got Future attached to a different loop"
RuntimeError. Capture a weakref to the loop on first _ensure_locks(),
raise ProgrammingError up front if a different loop later calls in,
and clear the pin on close() so a legitimate fixture-reuse cycle
still works.

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

Author: antoineleclair

src/dqlitedbapi/aio/connection.py

@@ -3,6 +3,7 @@
 import asyncio
 import contextlib
 import math
+import weakref
 from types import TracebackType
 from typing import Any
 
@@ -62,15 +63,41 @@ def __init__(
         # glue code before any loop exists).
         self._connect_lock: asyncio.Lock | None = None
         self._op_lock: asyncio.Lock | None = None
+        # Weak reference to the loop the locks were first bound to.
+        # Captured at first ``_ensure_locks()`` so subsequent use from a
+        # different event loop raises a clean ProgrammingError instead
+        # of asyncio's internal "got Future attached to a different
+        # loop" RuntimeError. Weakref avoids pinning a closed loop
+        # alive once the caller has moved on.
+        self._loop_ref: weakref.ref[asyncio.AbstractEventLoop] | None = None
         # PEP 249 optional extension; see Connection.messages.
         self.messages: list[tuple[type, Any]] = []
 
     def _ensure_locks(self) -> tuple[asyncio.Lock, asyncio.Lock]:
-        """Lazy-create the asyncio locks on the currently-running loop."""
+        """Lazy-create the asyncio locks on the currently-running loop.
+
+        Also pins the connection to that loop: subsequent calls from a
+        different loop raise ``ProgrammingError`` up front. The
+        underlying ``DqliteConnection`` protocol's StreamReader/Writer
+        is also loop-bound, so transparently rebinding is not safe;
+        fail fast with a clear message instead.
+        """
+        loop = asyncio.get_running_loop()
         if self._connect_lock is None:
+            self._loop_ref = weakref.ref(loop)
             self._connect_lock = asyncio.Lock()
-        if self._op_lock is None:
             self._op_lock = asyncio.Lock()
+        else:
+            bound = self._loop_ref() if self._loop_ref is not None else None
+            if bound is not loop:
+                raise ProgrammingError(
+                    "AsyncConnection was first used on a different event loop; "
+                    "AsyncConnection instances are loop-bound and cannot be "
+                    "reused across asyncio.run() invocations."
+                )
+        # ``_op_lock`` is created together with ``_connect_lock`` above;
+        # the assertion keeps mypy narrow without a runtime cost.
+        assert self._op_lock is not None
         return self._connect_lock, self._op_lock
 
     async def _ensure_connection(self) -> DqliteConnection:
@@ -130,6 +157,7 @@ async def close(self) -> None:
             # connect_lock.
             self._connect_lock = None
             self._op_lock = None
+            self._loop_ref = None
             return
         _, op_lock = self._ensure_locks()
         async with op_lock:
@@ -141,6 +169,7 @@ async def close(self) -> None:
         # re-check before it touches the now-None primitive.
         self._connect_lock = None
         self._op_lock = None
+        self._loop_id = None
 
     async def commit(self) -> None:
         """Commit any pending transaction.

tests/test_async_lock_binding.py

@@ -65,3 +65,68 @@ async def scenario() -> None:
             assert conn._op_lock is None
 
         asyncio.run(scenario())
+
+
+class TestLoopAffinityEnforcement:
+    """After the first _ensure_locks() call, the AsyncConnection is
+    pinned to that loop. Any subsequent use from a different loop must
+    raise a clean ProgrammingError instead of asyncio's internal
+    "got Future attached to a different loop" RuntimeError.
+    """
+
+    def test_cross_loop_use_raises_programming_error(self) -> None:
+        import asyncio
+
+        from dqlitedbapi.aio.connection import AsyncConnection
+        from dqlitedbapi.exceptions import ProgrammingError
+
+        conn = AsyncConnection("localhost:19001", database="x")
+
+        async def touch() -> None:
+            conn._ensure_locks()
+
+        asyncio.run(touch())
+
+        loop2 = asyncio.new_event_loop()
+        try:
+            with pytest.raises(ProgrammingError, match="loop"):
+                loop2.run_until_complete(touch())
+        finally:
+            loop2.close()
+
+    def test_same_loop_reuse_is_fine(self) -> None:
+        import asyncio
+
+        from dqlitedbapi.aio.connection import AsyncConnection
+
+        conn = AsyncConnection("localhost:19001", database="x")
+
+        async def touch() -> tuple[asyncio.Lock, asyncio.Lock]:
+            a1, b1 = conn._ensure_locks()
+            a2, b2 = conn._ensure_locks()
+            assert a1 is a2
+            assert b1 is b2
+            return a1, b1
+
+        asyncio.run(touch())
+
+    def test_close_clears_loop_pin(self) -> None:
+        """Close resets the pin so a subsequent asyncio.run on the
+        same object (e.g. test fixture reuse) works without tripping
+        the cross-loop guard.
+        """
+        import asyncio
+
+        from dqlitedbapi.aio.connection import AsyncConnection
+
+        conn = AsyncConnection("localhost:19001", database="x")
+
+        async def _scenario() -> None:
+            conn._ensure_locks()
+            await conn.close()
+
+        asyncio.run(_scenario())
+        assert conn._loop_ref is None
+
+
+import pytest  # noqa: E402