Skip to content

Commit 4a8ce8a

Browse files
committed
fix(client): contest abandoned commits, retriable rollbacks, interrupt settlement
Four fixes, each with a regression test seen failing first. Cancellation landing during the normal-exit indeterminate cleanup now propagates (with a detached retry spawned) instead of being swallowed into a successful-looking exit. A block that raises while a background commit is in flight now contests it with the detached best-effort rollback: a landed commit cannot be retracted, so the server race decides the outcome instead of the commit silently applying despite the exception. An explicit rollback() whose request is lost in transit settles the handle as aborted with the cleanup unconfirmed — still unusable for statements, but a later rollback() can retry once connectivity recovers, instead of being rejected as already finished (the existing lost-rollback test's expected message updated for that deliberate contract change). And the sync statement wrapper mirrors commit()'s BaseException settlement, so an interrupt raised inside the stepping coroutine closes the handle as aborted-and-cleanable rather than parking it in "executing" forever.
1 parent 867c140 commit 4a8ce8a

2 files changed

Lines changed: 186 additions & 9 deletions

File tree

coordinode/coordinode/client.py

Lines changed: 50 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -603,9 +603,20 @@ async def rollback(self) -> None:
603603
# The failure still propagates, so they know the request did not
604604
# arrive.
605605
self._state = "rolled_back"
606-
await self._client._cypher_stub.RollbackTransaction(
607-
RollbackTransactionRequest(transaction_id=self._id), timeout=self._client._timeout
608-
)
606+
try:
607+
await self._client._cypher_stub.RollbackTransaction(
608+
RollbackTransactionRequest(transaction_id=self._id), timeout=self._client._timeout
609+
)
610+
except BaseException:
611+
# The request may never have arrived: the discard promise still
612+
# holds (no commit was ever sent), but the server may hold the
613+
# transaction until its idle sweep. "aborted" with the cleanup
614+
# unconfirmed keeps the handle unusable for statements while a
615+
# later rollback() can still retry once connectivity recovers;
616+
# the failure propagates so the caller knows it did not land.
617+
self._state = "aborted"
618+
self._cleanup_confirmed = False
619+
raise
609620

610621

611622
class AsyncCoordinodeClient:
@@ -907,8 +918,14 @@ async def transaction(self) -> AsyncIterator[AsyncTransaction]:
907918
# An operation started in a background task is still in
908919
# flight while this scope unwinds; it cannot be awaited or
909920
# cancelled from here, so the straggler is marked to hand the
910-
# transaction to cleanup when it completes.
921+
# transaction to cleanup when it completes. An in-flight
922+
# COMMIT is additionally contested with a detached rollback:
923+
# a successful commit cannot be retracted afterwards, so the
924+
# only honest shot at the rollback-on-cancellation contract
925+
# is letting the server race decide which request wins.
911926
tx._abandoned = True
927+
if tx._state == "committing":
928+
tx._spawn_cleanup()
912929
raise
913930
except BaseException:
914931
if tx.is_open:
@@ -934,8 +951,14 @@ async def transaction(self) -> AsyncIterator[AsyncTransaction]:
934951
# be awaited or cancelled from here — so the straggler is
935952
# marked to hand the transaction to cleanup when it
936953
# completes, instead of returning the handle to "open" with
937-
# nobody left to use it.
954+
# nobody left to use it. An in-flight COMMIT is additionally
955+
# contested with a detached rollback: a successful commit
956+
# cannot be retracted afterwards, so the only honest shot at
957+
# the rollback-on-exception contract is letting the server
958+
# race decide which request wins.
938959
tx._abandoned = True
960+
if tx._state == "committing":
961+
tx._spawn_cleanup()
939962
raise
940963
else:
941964
if tx._state in ("executing", "committing"):
@@ -978,8 +1001,15 @@ async def transaction(self) -> AsyncIterator[AsyncTransaction]:
9781001
# never have reached the server, leaving the transaction open
9791002
# there. Same bounded best-effort cleanup as the exception
9801003
# path; the verdict stays indeterminate.
981-
with suppress(asyncio.CancelledError):
1004+
try:
9821005
await tx._best_effort_rollback()
1006+
except asyncio.CancelledError:
1007+
# Swallowing the cancellation here would turn a cancelled
1008+
# exit into a successful-looking one — propagate it, and
1009+
# hand the interrupted cleanup to a detached retry so the
1010+
# server transaction is still freed.
1011+
tx._spawn_cleanup()
1012+
raise
9831013

9841014
async def vector_search(
9851015
self,
@@ -1554,7 +1584,20 @@ def cypher(
15541584
params: dict[str, PyValue] | None = None,
15551585
) -> list[dict[str, Any]]:
15561586
"""Run one statement inside this transaction. See :meth:`AsyncTransaction.cypher`."""
1557-
return self._client._run(self._inner.cypher(query, params)) # type: ignore[no-any-return]
1587+
try:
1588+
return self._client._run(self._inner.cypher(query, params)) # type: ignore[no-any-return]
1589+
except BaseException:
1590+
# An interruption raised INSIDE the stepping coroutine (Ctrl-C
1591+
# delivered mid-call) completes the task before _run() can
1592+
# cancel-and-drain it, so no async handler closes the handle.
1593+
# Mirror commit()'s conservative settlement: no commit was sent,
1594+
# so nothing can apply — the handle closes as aborted, with the
1595+
# cleanup unconfirmed so a later rollback() frees the server
1596+
# side. An outcome the inner handler already decided is kept.
1597+
if self._inner._state in ("open", "executing"):
1598+
self._inner._state = "aborted"
1599+
self._inner._cleanup_confirmed = False
1600+
raise
15581601

15591602
def commit(self) -> int:
15601603
"""Apply every buffered write as one unit. See :meth:`AsyncTransaction.commit`."""

tests/unit/test_transactions.py

Lines changed: 136 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -706,7 +706,8 @@ def test_a_lost_rollback_still_closes_the_transaction(self):
706706
"""The request may not have landed, so the server may still hold the
707707
transaction until the idle sweep. What is certain is that no commit was
708708
ever sent, so nothing can apply: the discard promise holds and the
709-
handle must not stay usable."""
709+
handle must not stay usable — though it stays RETRIABLE for rollback,
710+
so the closed state reads as aborted, not rolled back."""
710711
from unittest.mock import AsyncMock
711712

712713
async def _inner() -> None:
@@ -717,7 +718,7 @@ async def _inner() -> None:
717718
with pytest.raises(_TransportError):
718719
await tx.rollback()
719720
assert tx.is_open is False
720-
with pytest.raises(RuntimeError, match="already rolled back"):
721+
with pytest.raises(RuntimeError, match="earlier failure closed it"):
721722
await tx.cypher("CREATE (:Person)")
722723

723724
asyncio.run(_inner())
@@ -1826,3 +1827,136 @@ async def hang(req, timeout=None):
18261827
await client.close()
18271828

18281829
asyncio.run(_inner())
1830+
1831+
1832+
class TestCancelledNormalExitCleanupPropagates:
1833+
"""Cancellation landing while the normal-exit indeterminate cleanup is
1834+
awaiting the server must propagate, not be swallowed into a
1835+
successful-looking exit — and the interrupted cleanup must be retried
1836+
detached so the server transaction is still freed."""
1837+
1838+
def test_cancellation_mid_cleanup_is_not_swallowed(self):
1839+
from contextlib import suppress as ctx_suppress
1840+
from unittest.mock import AsyncMock
1841+
1842+
async def _inner() -> None:
1843+
rollback_started = asyncio.Event()
1844+
calls = {"n": 0}
1845+
1846+
async def rollback(req, timeout=None):
1847+
calls["n"] += 1
1848+
if calls["n"] == 1:
1849+
rollback_started.set()
1850+
await asyncio.sleep(10)
1851+
return cypher_pb2.RollbackTransactionResponse()
1852+
1853+
client = _async_client(
1854+
CommitTransaction=AsyncMock(side_effect=_TransportError(grpc.StatusCode.DEADLINE_EXCEEDED)),
1855+
RollbackTransaction=AsyncMock(side_effect=rollback),
1856+
)
1857+
1858+
async def run_block() -> None:
1859+
async with client.transaction() as tx:
1860+
await tx.cypher("CREATE (:A)")
1861+
with ctx_suppress(grpc.RpcError):
1862+
await tx.commit()
1863+
1864+
t = asyncio.create_task(run_block())
1865+
await rollback_started.wait()
1866+
t.cancel()
1867+
with pytest.raises(asyncio.CancelledError):
1868+
await t
1869+
await client.close() # drains the detached retry
1870+
assert calls["n"] == 2, "the interrupted cleanup was never retried"
1871+
1872+
asyncio.run(_inner())
1873+
1874+
1875+
class TestAbandonedInFlightCommitIsContested:
1876+
"""A block that starts commit() in a background task and then raises has
1877+
asked for a rollback it can no longer perform itself: the exit must send
1878+
the best-effort rollback to CONTEST the in-flight commit, so the server
1879+
race decides the outcome instead of the commit silently applying despite
1880+
the exception."""
1881+
1882+
def test_exceptional_exit_sends_a_contesting_rollback(self):
1883+
from unittest.mock import AsyncMock
1884+
1885+
async def _inner() -> None:
1886+
commit_in_flight = asyncio.Event()
1887+
release_commit = asyncio.Event()
1888+
1889+
async def gated_commit(req, timeout=None):
1890+
commit_in_flight.set()
1891+
await release_commit.wait()
1892+
return cypher_pb2.CommitTransactionResponse(applied_index=7)
1893+
1894+
client = _async_client(CommitTransaction=AsyncMock(side_effect=gated_commit))
1895+
bg = None
1896+
with pytest.raises(ValueError):
1897+
async with client.transaction() as tx:
1898+
await tx.cypher("CREATE (:A)")
1899+
bg = asyncio.create_task(tx.commit())
1900+
await commit_in_flight.wait()
1901+
raise ValueError("boom")
1902+
await asyncio.sleep(0.02) # the detached contesting rollback runs
1903+
assert client._cypher_stub.RollbackTransaction.await_count == 1, (
1904+
"an abandoned in-flight commit was left uncontested"
1905+
)
1906+
release_commit.set()
1907+
await bg # the fake server lets the commit win; that is its call
1908+
await client.close()
1909+
1910+
asyncio.run(_inner())
1911+
1912+
1913+
class TestFailedExplicitRollbackStaysRetriable:
1914+
"""An explicit rollback() whose request is lost in transit must not leave
1915+
the handle permanently rolled_back: the server may still hold the
1916+
transaction, so once connectivity recovers a later rollback() has to be
1917+
able to retry instead of being rejected as already finished."""
1918+
1919+
def test_rollback_can_be_retried_after_a_transport_failure(self):
1920+
from unittest.mock import AsyncMock
1921+
1922+
async def _inner() -> None:
1923+
rollback = AsyncMock(
1924+
side_effect=[
1925+
_TransportError(grpc.StatusCode.UNAVAILABLE),
1926+
cypher_pb2.RollbackTransactionResponse(),
1927+
]
1928+
)
1929+
client = _async_client(RollbackTransaction=rollback)
1930+
tx = await client.begin_transaction()
1931+
with pytest.raises(_TransportError):
1932+
await tx.rollback()
1933+
assert tx.is_open is False, "a failed rollback left the handle usable"
1934+
await tx.rollback() # connectivity recovered; the retry must go through
1935+
assert client._cypher_stub.RollbackTransaction.await_count == 2
1936+
await client.close()
1937+
1938+
asyncio.run(_inner())
1939+
1940+
1941+
class TestSyncStatementInterruptedInsideTheTask:
1942+
"""KeyboardInterrupt raised while the statement coroutine itself is
1943+
stepping completes the task before _run() can cancel-and-drain it, so no
1944+
async handler closes the handle; the sync wrapper must then settle the
1945+
outcome conservatively instead of leaving the handle parked in
1946+
"executing" forever."""
1947+
1948+
def test_interrupted_statement_closes_the_handle(self):
1949+
from unittest.mock import AsyncMock
1950+
1951+
async def ki(req, timeout=None):
1952+
raise KeyboardInterrupt
1953+
1954+
client = _sync_client(ExecuteCypher=AsyncMock(side_effect=ki))
1955+
tx = client.begin_transaction()
1956+
with pytest.raises(KeyboardInterrupt):
1957+
tx.cypher("CREATE (:A)")
1958+
assert tx._inner._state == "aborted", "the interruption left the handle parked in-flight"
1959+
# The server may still hold the transaction; an explicit rollback
1960+
# must send the cleanup rather than trusting one that never ran.
1961+
tx.rollback()
1962+
assert client._async._cypher_stub.RollbackTransaction.await_count == 1

0 commit comments

Comments
 (0)