fix(client): refuse concurrent operations while a commit is in flight

Two tasks committing the same handle could both pass the open-check;
the server applies one and rejects the other for the consumed handle,
and if that rejection arrived last it overwrote `committed` with
`aborted`, telling the caller nothing was applied and inviting a
duplicate retry. The handle now transitions to a `committing` state
BEFORE awaiting the RPC, so a concurrent commit or statement is
refused with a clear error before it sends anything; the state settles
to the real outcome when the first commit resolves. An interruption at
the synchronous boundary treats a parked `committing` like `open` and
records indeterminate.

Regression tests (seen failing before the fix): a second concurrent
commit and a statement during a commit are both refused with exactly
one RPC sent and the true outcome preserved.

Author: polaz

coordinode/coordinode/client.py

@@ -314,6 +314,12 @@ def is_open(self) -> bool:
     def _require_open(self, action: str) -> None:
         if self._state == "open":
             return
+        if self._state == "committing":
+            raise RuntimeError(
+                f"Cannot {action} this transaction: its commit is already in flight. "
+                "Concurrent operations on one transaction handle would race the "
+                "commit's outcome; await it instead."
+            )
         if self._state == "aborted":
             raise RuntimeError(
                 f"Cannot {action} this transaction: an earlier failure closed it on the "
@@ -467,6 +473,11 @@ async def commit(self) -> int:
         )
 
         self._require_open("commit")
+        # Transition BEFORE the await: a concurrent operation on this handle
+        # would otherwise pass its own open-check while the commit is in
+        # flight, and the loser's "unknown transaction" rejection would
+        # overwrite the real outcome (committed) with a claimed abort.
+        self._state = "committing"
         try:
             resp = await self._client._cypher_stub.CommitTransaction(
                 CommitTransactionRequest(transaction_id=self._id), timeout=self._client._timeout
@@ -1371,11 +1382,12 @@ def commit(self) -> int:
         except BaseException:
             # An interruption at the loop boundary (Ctrl-C, SystemExit) never
             # reaches the async handlers, so without this the handle would
-            # read "open" while the server may already have applied the
-            # writes — inviting the duplicate retry the indeterminate state
-            # exists to prevent. An outcome the inner handler already decided
-            # (aborted, indeterminate, committed) is kept.
-            if self._inner._state == "open":
+            # read "open" (or stay parked mid-"committing") while the server
+            # may already have applied the writes — inviting the duplicate
+            # retry the indeterminate state exists to prevent. An outcome the
+            # inner handler already decided (aborted, indeterminate,
+            # committed) is kept.
+            if self._inner._state in ("open", "committing"):
                 self._inner._state = "indeterminate"
             raise
 

tests/unit/test_transactions.py

@@ -1206,3 +1206,51 @@ async def slow_rollback(req, timeout=None):
             assert cleanup_done.is_set(), "cancelling close() killed the cleanup"
 
         asyncio.run(_inner())
+
+
+class TestConcurrentCommitSerialization:
+    """Two tasks committing the same handle must not race the state machine:
+    the second must be refused BEFORE it sends anything, or the loser's
+    "unknown transaction" rejection would overwrite `committed` with
+    `aborted` and invite a duplicate retry."""
+
+    def test_a_second_concurrent_commit_is_refused_not_raced(self):
+        from unittest.mock import AsyncMock
+
+        async def _inner() -> None:
+            async def slow_commit(req, timeout=None):
+                await asyncio.sleep(0.05)
+                return cypher_pb2.CommitTransactionResponse(applied_index=7)
+
+            client = _async_client(CommitTransaction=AsyncMock(side_effect=slow_commit))
+            tx = await client.begin_transaction()
+            first = asyncio.create_task(tx.commit())
+            await asyncio.sleep(0.01)  # first commit is now awaiting the RPC
+            with pytest.raises(RuntimeError, match="in flight"):
+                await tx.commit()
+            assert await first == 7
+            assert client._cypher_stub.CommitTransaction.await_count == 1
+            # The handle records the real outcome, untouched by the refusal.
+            with pytest.raises(RuntimeError, match="already committed"):
+                await tx.cypher("RETURN 1")
+
+        asyncio.run(_inner())
+
+    def test_a_statement_during_a_commit_is_refused(self):
+        from unittest.mock import AsyncMock
+
+        async def _inner() -> None:
+            async def slow_commit(req, timeout=None):
+                await asyncio.sleep(0.05)
+                return cypher_pb2.CommitTransactionResponse(applied_index=7)
+
+            client = _async_client(CommitTransaction=AsyncMock(side_effect=slow_commit))
+            tx = await client.begin_transaction()
+            first = asyncio.create_task(tx.commit())
+            await asyncio.sleep(0.01)
+            with pytest.raises(RuntimeError, match="in flight"):
+                await tx.cypher("CREATE (:Late)")
+            assert await first == 7
+            assert client._cypher_stub.ExecuteCypher.await_count == 0
+
+        asyncio.run(_inner())