fix: add transaction task-ownership tracking to prevent cross-task injection

When a connection is in a transaction, another asyncio task could call
execute() between the transaction owner's operations (when _in_use is
temporarily False), silently injecting operations into the transaction.

Track the owning task via asyncio.current_task() in the transaction()
context manager. _check_in_use() now rejects operations from tasks
that don't own the active transaction.

Fixes #100

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -81,6 +81,7 @@ def __init__(
         self._in_transaction = False
         self._in_use = False
         self._bound_loop: asyncio.AbstractEventLoop | None = None
+        self._tx_owner: asyncio.Task[Any] | None = None
 
     @property
     def address(self) -> str:
@@ -172,6 +173,14 @@ def _check_in_use(self) -> None:
                 "connection. DqliteConnection does not support concurrent coroutine "
                 "access. Use a ConnectionPool to manage multiple concurrent operations."
             )
+        if self._in_transaction and self._tx_owner is not None:
+            current = asyncio.current_task()
+            if current is not self._tx_owner:
+                raise InterfaceError(
+                    "Cannot perform operation: connection is in a transaction owned "
+                    "by another task. Each task should use its own connection from "
+                    "the pool."
+                )
 
     def _invalidate(self) -> None:
         """Mark the connection as broken after an unrecoverable error."""
@@ -286,9 +295,11 @@ async def transaction(self) -> AsyncIterator[None]:
             )
 
         self._in_transaction = True
+        self._tx_owner = asyncio.current_task()
         try:
             await self.execute("BEGIN")
         except BaseException:
+            self._tx_owner = None
             self._in_transaction = False
             raise
 
@@ -301,4 +312,5 @@ async def transaction(self) -> AsyncIterator[None]:
                 await self.execute("ROLLBACK")
             raise
         finally:
+            self._tx_owner = None
             self._in_transaction = False

tests/test_connection.py

@@ -899,3 +899,66 @@ async def use_conn():
             f"Expected InterfaceError, got {type(error_from_thread).__name__}: {error_from_thread}"
         )
         assert "event loop" in str(error_from_thread).lower()
+
+    async def test_other_task_rejected_during_transaction(self) -> None:
+        """Another task calling execute() during an active transaction must be rejected."""
+        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()
+
+        # Mock execute for the transaction owner (task A) — needs to work
+        a_inside_tx = asyncio.Event()
+
+        async def mock_execute_a(sql: str, params=None):
+            if sql not in ("BEGIN", "COMMIT", "ROLLBACK"):
+                a_inside_tx.set()
+                await asyncio.sleep(0)  # yield to let task B run
+            return (0, 0)
+
+        conn.execute = mock_execute_a  # type: ignore[assignment]
+
+        errors: list[Exception] = []
+
+        async def task_a():
+            async with conn.transaction():
+                await conn.execute("INSERT INTO t VALUES (1)")
+                await asyncio.sleep(0.1)
+
+        async def task_b():
+            await a_inside_tx.wait()
+            try:
+                # Use _check_in_use directly — this is what real execute() calls
+                conn._check_in_use()
+            except InterfaceError as e:
+                errors.append(e)
+
+        t_a = asyncio.create_task(task_a())
+        t_b = asyncio.create_task(task_b())
+
+        await asyncio.gather(t_a, t_b, return_exceptions=True)
+
+        assert len(errors) == 1, (
+            "Task B should have been rejected when trying to use a connection "
+            "that is in a transaction owned by task A"
+        )
+        assert isinstance(errors[0], InterfaceError)
+        assert "transaction" in str(errors[0]).lower()