fix(client): serialize statements with the same in-flight guard as commits

The mirror image of the commit race: while a statement awaited its RPC
the handle still read "open", so a concurrent commit could pass its
check, land on the server without the statement's write, and the
statement's late "unknown transaction" failure would overwrite the
recorded outcome. A statement now transitions the handle to an
in-flight state before awaiting, returning to open on success, so a
concurrent commit or second statement is refused with a clear error
before sending anything. Together with the committing state this makes
every operation on one handle strictly serialized.

Regression tests (seen failing before the fix): a commit during a slow
statement and a second concurrent statement are both refused, the
in-flight statement completes, and the handle works again afterwards.

Author: polaz

coordinode/coordinode/client.py

@@ -314,11 +314,12 @@ def is_open(self) -> bool:
     def _require_open(self, action: str) -> None:
         if self._state == "open":
             return
-        if self._state == "committing":
+        if self._state in ("committing", "executing"):
+            in_flight = "commit" if self._state == "committing" else "statement"
             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."
+                f"Cannot {action} this transaction: a {in_flight} on it is already in "
+                "flight. Concurrent operations on one transaction handle would race "
+                "its outcome; await the in-flight operation instead."
             )
         if self._state == "aborted":
             raise RuntimeError(
@@ -403,6 +404,11 @@ async def cypher(
         )
 
         self._require_open("run a statement in")
+        # Transition BEFORE the await, mirroring commit(): a concurrent
+        # commit slipping in while this statement is in flight could land
+        # without the statement's write, and the statement's late "unknown
+        # transaction" failure would then overwrite the real outcome.
+        self._state = "executing"
         req = ExecuteCypherRequest(
             query=query,
             parameters=dict_to_props(params or {}),
@@ -447,6 +453,7 @@ async def cypher(
             self._state = "aborted"
             self._spawn_cleanup()
             raise
+        self._state = "open"
         return _rows_to_dicts(resp)
 
     async def commit(self) -> int:

tests/unit/test_transactions.py

@@ -1254,3 +1254,56 @@ async def slow_commit(req, timeout=None):
             assert client._cypher_stub.ExecuteCypher.await_count == 0
 
         asyncio.run(_inner())
+
+
+class TestConcurrentStatementSerialization:
+    """The mirror image of the commit race: while a statement awaits its RPC
+    the handle must not accept a concurrent commit (the commit could land
+    without the statement's write, then the statement's failure would
+    overwrite `committed` with `aborted`) nor a second statement."""
+
+    def test_a_commit_during_a_statement_is_refused(self):
+        from unittest.mock import AsyncMock
+
+        async def _inner() -> None:
+            started = asyncio.Event()
+
+            async def slow_execute(req, timeout=None):
+                started.set()
+                await asyncio.sleep(0.05)
+                return _execute_response()
+
+            client = _async_client(ExecuteCypher=AsyncMock(side_effect=slow_execute))
+            tx = await client.begin_transaction()
+            stmt = asyncio.create_task(tx.cypher("CREATE (:A)"))
+            await started.wait()
+            with pytest.raises(RuntimeError, match="in flight"):
+                await tx.commit()
+            assert client._cypher_stub.CommitTransaction.await_count == 0
+            await stmt
+            # The handle is usable again once the statement resolved.
+            assert await tx.commit() == 7
+
+        asyncio.run(_inner())
+
+    def test_a_second_concurrent_statement_is_refused(self):
+        from unittest.mock import AsyncMock
+
+        async def _inner() -> None:
+            started = asyncio.Event()
+
+            async def slow_execute(req, timeout=None):
+                started.set()
+                await asyncio.sleep(0.05)
+                return _execute_response()
+
+            client = _async_client(ExecuteCypher=AsyncMock(side_effect=slow_execute))
+            tx = await client.begin_transaction()
+            stmt = asyncio.create_task(tx.cypher("CREATE (:A)"))
+            await started.wait()
+            with pytest.raises(RuntimeError, match="in flight"):
+                await tx.cypher("CREATE (:B)")
+            await stmt
+            assert client._cypher_stub.ExecuteCypher.await_count == 1
+
+        asyncio.run(_inner())