fix: invalidate connection after a failed COMMIT

A non-leader OperationalError from COMMIT leaves server-side state
ambiguous (maybe committed, maybe not, maybe still open). The previous
code attempted a suppressed ROLLBACK and then cleared _in_transaction,
letting the pool recycle a connection with unknown state.

Distinguish pre-COMMIT body failures (try ROLLBACK, keep the
connection) from COMMIT failures (invalidate so the pool discards it).

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -338,13 +338,23 @@ async def transaction(self) -> AsyncIterator[None]:
             self._in_transaction = False
             raise
 
+        commit_attempted = False
         try:
             yield
+            commit_attempted = True
             await self.execute("COMMIT")
         except BaseException:
-            # Swallow rollback failure; original exception is more important
-            with contextlib.suppress(BaseException):
-                await self.execute("ROLLBACK")
+            if commit_attempted:
+                # COMMIT was sent but failed. Server-side state is ambiguous
+                # (maybe committed, maybe still open, maybe rolled back). We
+                # cannot safely reuse this connection — invalidate so the
+                # pool discards it instead of recycling an unknown state.
+                self._invalidate()
+            else:
+                # Body raised before COMMIT; try to roll back. Swallow
+                # rollback errors; the original exception is more important.
+                with contextlib.suppress(BaseException):
+                    await self.execute("ROLLBACK")
             raise
         finally:
             self._tx_owner = None

tests/test_connection.py

@@ -821,6 +821,26 @@ async def test_close_on_pool_released_connection_is_noop(self, connected_connect
         # Would previously raise InterfaceError("returned to the pool").
         await conn.close()
 
+    async def test_commit_failure_invalidates_connection(self, connected_connection) -> None:
+        """A failed COMMIT leaves server-side state ambiguous; the connection
+        must be invalidated so the pool doesn't recycle it to another caller
+        in an unknown state.
+        """
+        from dqliteclient.exceptions import OperationalError as OpError
+        from dqlitewire.messages import FailureResponse, ResultResponse
+
+        conn, mock_reader, _ = connected_connection
+        mock_reader.read.side_effect = [
+            ResultResponse(last_insert_id=0, rows_affected=0).encode(),  # BEGIN
+            FailureResponse(code=1, message="disk I/O error").encode(),  # COMMIT
+            # ROLLBACK would also fail on an invalidated conn:
+            FailureResponse(code=1, message="should not reach").encode(),
+        ]
+        with pytest.raises(OpError):
+            async with conn.transaction():
+                pass
+        assert not conn.is_connected, "failed COMMIT must invalidate the connection"
+
     async def test_cross_event_loop_raises_interface_error(self) -> None:
         """Using a connection from a different event loop must raise InterfaceError."""
         import asyncio