fix(client): never swallow cancellation in cleanups, narrow sync interrupt settlement

Three fixes, each with a regression test seen failing first. Both
inline indeterminate cleanups (after a failed automatic commit, and on
the exceptional exit path) now let a cancellation that lands mid-RPC
propagate — asyncio does not re-inject a swallowed one — while handing
the interrupted cleanup to a detached retry. The context manager's
cancellation handler gains the aborted-with-unconfirmed-cleanup branch,
so a manual rollback() cancelled mid-RPC still gets its detached retry
before the scope is gone. And the sync statement wrapper's conservative
interrupt settlement is narrowed to real interruptions (non-Exception):
a local encoding failure sends no RPC and leaves the handle open,
matching the async contract.

Author: polaz

coordinode/coordinode/client.py

@@ -926,6 +926,11 @@ async def transaction(self) -> AsyncIterator[AsyncTransaction]:
                 tx._abandoned = True
                 if tx._state == "committing":
                     tx._spawn_cleanup()
+            elif tx._state == "aborted" and not tx._cleanup_confirmed:
+                # A manual rollback() cancelled mid-RPC: the request was
+                # interrupted, not answered, so the server may still hold the
+                # transaction. Retry detached before the scope is gone.
+                tx._spawn_cleanup()
             raise
         except BaseException:
             if tx.is_open:
@@ -942,8 +947,15 @@ async def transaction(self) -> AsyncIterator[AsyncTransaction]:
                 # best-effort request frees it in that case; if the commit
                 # applied, the server answers "unknown id" and nothing
                 # changes. The verdict stays indeterminate either way.
-                with suppress(asyncio.CancelledError):
+                try:
                     await tx._best_effort_rollback()
+                except asyncio.CancelledError:
+                    # Suppressing this would re-raise the block's error and
+                    # LOSE the cancellation — asyncio does not re-inject a
+                    # swallowed one. Spawn the detached retry and let it
+                    # propagate.
+                    tx._spawn_cleanup()
+                    raise
             elif tx._state in ("executing", "committing"):
                 # The block raised while an operation it started (in a
                 # background task) is still in flight. Raising over the
@@ -992,8 +1004,15 @@ async def transaction(self) -> AsyncIterator[AsyncTransaction]:
                 except BaseException:
                     if tx._state == "indeterminate":
                         # Same reasoning as above, for the automatic commit.
-                        with suppress(asyncio.CancelledError):
+                        try:
                             await tx._best_effort_rollback()
+                        except asyncio.CancelledError:
+                            # Suppressing this would re-raise the commit
+                            # error and LOSE the cancellation — asyncio does
+                            # not re-inject a swallowed one. Spawn the
+                            # detached retry and let it propagate.
+                            tx._spawn_cleanup()
+                            raise
                     raise
             elif tx._state == "indeterminate":
                 # A manual commit() inside the block failed ambiguously and
@@ -1586,15 +1605,19 @@ def cypher(
         """Run one statement inside this transaction. See :meth:`AsyncTransaction.cypher`."""
         try:
             return self._client._run(self._inner.cypher(query, params))  # type: ignore[no-any-return]
-        except BaseException:
+        except BaseException as exc:
             # An interruption raised INSIDE the stepping coroutine (Ctrl-C
             # delivered mid-call) completes the task before _run() can
             # cancel-and-drain it, so no async handler closes the handle.
             # Mirror commit()'s conservative settlement: no commit was sent,
             # so nothing can apply — the handle closes as aborted, with the
             # cleanup unconfirmed so a later rollback() frees the server
-            # side. An outcome the inner handler already decided is kept.
-            if self._inner._state in ("open", "executing"):
+            # side. Only for REAL interruptions (non-Exception): a local
+            # failure such as an unsupported parameter type raises an
+            # ordinary Exception before any RPC and must leave the handle
+            # open, exactly as the async path does. An outcome the inner
+            # handler already decided is kept either way.
+            if not isinstance(exc, Exception) and self._inner._state in ("open", "executing"):
                 self._inner._state = "aborted"
                 self._inner._cleanup_confirmed = False
             raise

tests/unit/test_transactions.py

@@ -1960,3 +1960,101 @@ async def ki(req, timeout=None):
         # must send the cleanup rather than trusting one that never ran.
         tx.rollback()
         assert client._async._cypher_stub.RollbackTransaction.await_count == 1
+
+
+class TestCancelledFailedCommitCleanupPropagates:
+    """When the automatic commit fails ambiguously and cancellation then
+    lands during the inline cleanup, the cancellation must propagate (a
+    swallowed one is simply lost — asyncio does not re-inject it) and the
+    interrupted cleanup must be retried detached."""
+
+    def test_cancellation_mid_failed_commit_cleanup_is_not_swallowed(self):
+        from unittest.mock import AsyncMock
+
+        async def _inner() -> None:
+            rollback_started = asyncio.Event()
+            calls = {"n": 0}
+
+            async def rollback(req, timeout=None):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    rollback_started.set()
+                    await asyncio.sleep(10)
+                return cypher_pb2.RollbackTransactionResponse()
+
+            client = _async_client(
+                CommitTransaction=AsyncMock(side_effect=_TransportError(grpc.StatusCode.DEADLINE_EXCEEDED)),
+                RollbackTransaction=AsyncMock(side_effect=rollback),
+            )
+
+            async def run_block() -> None:
+                async with client.transaction() as tx:
+                    await tx.cypher("CREATE (:A)")
+
+            t = asyncio.create_task(run_block())
+            await rollback_started.wait()
+            t.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await t
+            await client.close()  # drains the detached retry
+            assert calls["n"] == 2, "the interrupted cleanup was never retried"
+
+        asyncio.run(_inner())
+
+
+class TestCancelledManualRollbackGetsDetachedRetry:
+    """A manual rollback() cancelled mid-RPC leaves the handle aborted with
+    the cleanup unconfirmed; the context exit must register the detached
+    retry, or closing the client strands the server-side transaction until
+    the idle sweep."""
+
+    def test_cancelled_rollback_in_context_spawns_cleanup(self):
+        from unittest.mock import AsyncMock
+
+        async def _inner() -> None:
+            rollback_started = asyncio.Event()
+            calls = {"n": 0}
+
+            async def rollback(req, timeout=None):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    rollback_started.set()
+                    await asyncio.sleep(10)
+                return cypher_pb2.RollbackTransactionResponse()
+
+            client = _async_client(RollbackTransaction=AsyncMock(side_effect=rollback))
+
+            async def run_block() -> None:
+                async with client.transaction() as tx:
+                    await tx.cypher("CREATE (:A)")
+                    await tx.rollback()
+
+            t = asyncio.create_task(run_block())
+            await rollback_started.wait()
+            t.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await t
+            await client.close()  # drains the detached retry
+            assert calls["n"] == 2, "a cancelled manual rollback got no detached retry"
+
+        asyncio.run(_inner())
+
+
+class TestSyncLocalEncodingFailureKeepsTheHandleUsable:
+    """Sync mirror of the async encoding-failure contract: a parameter the
+    encoder rejects fails locally before any RPC, so the handle must stay
+    open — the interrupt settlement is for real interruptions only, not for
+    ordinary local exceptions."""
+
+    def test_a_bad_parameter_leaves_the_sync_transaction_open(self):
+        client = _sync_client()
+        tx = client.begin_transaction()
+        tx.cypher("CREATE (:A)")
+        with pytest.raises(Exception):
+            tx.cypher("CREATE (:B {v: $v})", {"v": object()})
+        assert client._async._cypher_stub.ExecuteCypher.await_count == 1, (
+            "the failing statement must not have reached the wire"
+        )
+        assert tx.is_open is True, "a local encoding failure aborted a usable transaction"
+        tx.rollback()
+        assert client._async._cypher_stub.RollbackTransaction.await_count == 1