fix: roll back open transactions when returning connections to pool

When a connection with _in_transaction=True is returned to the pool
(via _release() or acquire()'s exception handler), the pool now issues
ROLLBACK before recycling. If the ROLLBACK fails, the connection is
destroyed instead of being returned dirty.

This prevents a connection with an open transaction from being handed
to the next caller, who would unknowingly operate inside someone
else's stale transaction.

Also adds a public in_transaction property to DqliteConnection.

Fixes #101

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -92,6 +92,11 @@ def is_connected(self) -> bool:
         """Check if connected."""
         return self._protocol is not None
 
+    @property
+    def in_transaction(self) -> bool:
+        """Check if a transaction is active."""
+        return self._in_transaction
+
     async def connect(self) -> None:
         """Establish connection to the database."""
         self._check_in_use()

src/dqliteclient/pool.py

@@ -148,7 +148,12 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
         except BaseException:
             if conn.is_connected and not self._closed:
                 # Connection is healthy — user code raised a non-connection error.
-                # Return it to the pool instead of destroying it.
+                # Roll back any open transaction, then return to pool.
+                if not await self._reset_connection(conn):
+                    with contextlib.suppress(Exception):
+                        await conn.close()
+                    self._size -= 1
+                    raise
                 try:
                     self._pool.put_nowait(conn)
                 except asyncio.QueueFull:
@@ -165,17 +170,38 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
         else:
             await self._release(conn)
 
+    async def _reset_connection(self, conn: DqliteConnection) -> bool:
+        """Roll back any open transaction before returning to pool.
+
+        Returns True if the connection is clean and can be reused,
+        False if it should be destroyed.
+        """
+        if conn._in_transaction is True:
+            try:
+                await conn.execute("ROLLBACK")
+            except Exception:
+                return False
+            conn._in_transaction = False
+        return True
+
     async def _release(self, conn: DqliteConnection) -> None:
         """Return a connection to the pool or close it."""
         if self._closed:
             await conn.close()
             self._size -= 1
-        else:
-            try:
-                self._pool.put_nowait(conn)
-            except asyncio.QueueFull:
+            return
+
+        if not await self._reset_connection(conn):
+            with contextlib.suppress(Exception):
                 await conn.close()
-                self._size -= 1
+            self._size -= 1
+            return
+
+        try:
+            self._pool.put_nowait(conn)
+        except asyncio.QueueFull:
+            await conn.close()
+            self._size -= 1
 
     async def execute(self, sql: str, params: Sequence[Any] | None = None) -> tuple[int, int]:
         """Execute a SQL statement using a pooled connection."""

tests/test_pool.py

@@ -568,6 +568,78 @@ async def test_pool_has_no_in_use_set(self) -> None:
             "Pool._in_use set should have been removed — it was dead code (written but never read)"
         )
 
+    async def test_release_rolls_back_open_transaction(self) -> None:
+        """Returning a connection with _in_transaction=True must issue ROLLBACK."""
+        pool = ConnectionPool(["localhost:9001"], max_size=1)
+
+        mock_conn = MagicMock()
+        mock_conn.is_connected = True
+        mock_conn.connect = AsyncMock()
+        mock_conn.close = AsyncMock()
+        mock_conn._in_transaction = False
+        mock_conn._in_use = False
+        mock_conn._bound_loop = None
+        mock_conn._check_in_use = MagicMock()
+
+        call_log: list[str] = []
+
+        async def mock_execute(sql, params=None):
+            call_log.append(sql)
+            return (0, 0)
+
+        mock_conn.execute = mock_execute
+
+        with patch.object(pool._cluster, "connect", return_value=mock_conn):
+            await pool.initialize()
+
+        # Simulate: user enters transaction but the context manager exit
+        # somehow leaves _in_transaction=True (e.g., a bug or raw BEGIN)
+        async with pool.acquire() as conn:
+            conn._in_transaction = True
+
+        # The pool should have issued ROLLBACK before returning to queue
+        assert "ROLLBACK" in call_log, (
+            f"Pool should issue ROLLBACK for dirty connections, calls: {call_log}"
+        )
+        # And the flag should be reset
+        assert not mock_conn._in_transaction
+
+        await pool.close()
+
+    async def test_release_destroys_connection_if_rollback_fails(self) -> None:
+        """If ROLLBACK fails on dirty release, connection must be destroyed."""
+        pool = ConnectionPool(["localhost:9001"], max_size=1)
+
+        mock_conn = MagicMock()
+        mock_conn.is_connected = True
+        mock_conn.connect = AsyncMock()
+        mock_conn.close = AsyncMock()
+        mock_conn._in_transaction = False
+        mock_conn._in_use = False
+        mock_conn._bound_loop = None
+        mock_conn._check_in_use = MagicMock()
+
+        async def failing_execute(sql, params=None):
+            if "ROLLBACK" in sql:
+                raise Exception("connection lost")
+            return (0, 0)
+
+        mock_conn.execute = failing_execute
+
+        with patch.object(pool._cluster, "connect", return_value=mock_conn):
+            await pool.initialize()
+
+        initial_size = pool._size
+
+        async with pool.acquire() as conn:
+            conn._in_transaction = True
+
+        # ROLLBACK failed, so connection should be destroyed
+        mock_conn.close.assert_called()
+        assert pool._size == initial_size - 1
+
+        await pool.close()
+
 
 class TestConnectionPoolIntegration:
     """Integration tests requiring mocked connections."""