From 0a6e03565c17aba305cb723b332ed197e3b0bf3a Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Sun, 13 Sep 2026 19:14:33 +0000 Subject: [PATCH 1/2] Presume a deadlock only with a holder that held through a grace period Wait-die's check ran each time a grace period elapsed and judged whoever held the lock at that instant, with no memory of earlier checks. It could not tell an older transaction that had held the lock the whole time the waiter waited from a lock that had changed hands several times and happened to be held by an older transaction just then. So on a busy state a younger transaction queued behind a run of short older holders waited past one grace period, saw an older holder at its check, and died with `TransactionShouldRetry`, reason `PRESUMED_DEADLOCK`, although the lock was turning over and it was about to be granted. The hook for a lock wait now snapshots the lock's holders as the wait begins and, each time a grace period elapses, judges only the holders present both now and at the previous snapshot. A holder in both has held the lock for a whole grace period while the waiter waited, which is what a deadlock looks like, and the presumption applies as before. A holder in only the current set arrived since the last check, so the lock is moving and the waiter keeps waiting. `claim_ownership` does the same with the owner id chain, running the sibling check only when the chain is unchanged since the previous grace period. Detection latency for a real deadlock is unchanged: the first snapshot is taken as the wait begins, so a holder stuck on the waiter is in both sets at the first check. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01EK2ybbxpNK5L7S5exgJ8fq --- reboot/aio/state_managers.py | 124 ++++++++++++++++++++-------- tests/reboot/transaction_tests.py | 131 +++++++++++++++++++++++++++++- 2 files changed, 220 insertions(+), 35 deletions(-) diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index 74fa62e1..a8be60f2 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -712,8 +712,21 @@ async def claim_ownership( transaction_ids = context.transaction_ids assert transaction_ids is not None + # Presume a deadlock only with an owner that has owned this + # state for a whole grace period while we waited; an owner + # that changed in between is a sign the state is moving + # between nested transactions, and we keep waiting. + previous_owner_ids = list(self.owner_ids) + def on_grace() -> None: - self._abort_if_presumed_sibling_deadlock(transaction_ids) + nonlocal previous_owner_ids + current_owner_ids = list(self.owner_ids) + if current_owner_ids == previous_owner_ids: + self._abort_if_presumed_sibling_deadlock( + transaction_ids, + current_owner_ids, + ) + previous_owner_ids = current_owner_ids await self.wait_ownership( lambda: self.is_claimable_by(transaction_ids), @@ -742,25 +755,28 @@ def on_grace() -> None: def _abort_if_presumed_sibling_deadlock( self, transaction_ids: list[uuid.UUID], + owner_ids: list[uuid.UUID], ) -> None: """Raises `SystemAborted(NestedTransactionShouldRetry(...))` - when the nested transaction owning this state is an older - sibling of a nested transaction among `transaction_ids`, so - that the younger sibling rolls back, ownership of what it - claimed returns to the transaction that started it, and the - older sibling proceeds: the younger of two siblings waiting - on each other's states is always the one to go. A state - owned by a younger sibling, or by a descendant of the - caller, keeps the caller waiting. + when `owner_ids`, the nested transactions that have owned + this state for a whole grace period while a call of + `transaction_ids` waited, end in an older sibling of a + nested transaction among `transaction_ids`, so that the + younger sibling rolls back, ownership of what it claimed + returns to the transaction that started it, and the older + sibling proceeds: the younger of two siblings waiting on + each other's states is always the one to go. A state owned + by a younger sibling, or by a descendant of the caller, + keeps the caller waiting. """ nested_transaction_id = presumed_deadlocked_nested_transaction( transaction_ids, - self.owner_ids, + owner_ids, ) if nested_transaction_id is None: return level = transaction_ids.index(nested_transaction_id) - owner_id = self.owner_ids[level] + owner_id = owner_ids[level] grace_ms = TRANSACTION_DEADLOCK_GRACE // timedelta(milliseconds=1) message = ( f"Nested transaction {nested_transaction_id} waited longer " @@ -2850,23 +2866,67 @@ def __init__( def latest_timestamp_ms(self) -> Optional[int]: return self._latest_timestamp_ms + def _presume_deadlock_on_grace( + self, + state_type: StateTypeName, + state_ref: StateRef, + transaction: StateManager.Transaction, + ) -> Callable[[], None]: + """The `on_grace` hook for `transaction` waiting on the lock of + `(state_type, state_ref)`: each time a grace period elapses it + presumes a deadlock with, and aborts `transaction` for, an + older holder that has held the lock since the previous grace + period elapsed, or since the wait began. A holder that arrived + in between is not a deadlock but the lock changing hands, and + `transaction`, queued behind it, keeps waiting. + + TODO: when every state in a cycle lives on this server, the + deadlock can be proven rather than presumed, and at once: record + on each participant the state it is waiting on (here and in + `claim_ownership`), and when a wait begins walk from the lock's + holders through what they wait on, and so on, looking for a + path back to `transaction`. A cycle found that way aborts its + youngest transaction immediately, with no grace period and no + false positive; a wait that leaves this server ends the walk + and falls back to the grace check. Most cycles in tests and in + small deployments are local, so this would resolve them in + microseconds instead of a grace period. + """ + lock = self._locks[state_type][state_ref] + previous_holders = set(lock.holders) + + def on_grace() -> None: + nonlocal previous_holders + current_holders = set(lock.holders) + self._abort_if_presumed_deadlock( + state_type, + state_ref, + transaction, + current_holders & previous_holders, + ) + previous_holders = current_holders + + return on_grace + def _abort_if_presumed_deadlock( self, state_type: StateTypeName, state_ref: StateRef, transaction: StateManager.Transaction, + holders: set[StateManager.Transaction], ) -> None: """Raises `SystemAborted(TransactionShouldRetry(...))`, with - reason `PRESUMED_DEADLOCK`, when the lock on `(state_type, - state_ref)` is held by a transaction older than `transaction`, - so that `transaction` aborts and the older one proceeds: the - younger of two transactions waiting on each other's states is - always the one to go. A lock held only by younger transactions, - or by a plain reader or writer outside any transaction (which - never waits on anything and so does not identify itself to the - lock), keeps `transaction` waiting. + reason `PRESUMED_DEADLOCK`, when `holders`, the transactions + that have held the lock on `(state_type, state_ref)` for a + whole grace period while `transaction` waited, include one + older than `transaction`, so that `transaction` aborts and the + older one proceeds: the younger of two transactions waiting on + each other's states is always the one to go. A lock held only + by younger transactions, or by a plain reader or writer outside + any transaction (which never waits on anything and so does not + identify itself to the lock), keeps `transaction` waiting. """ - for holder in self._locks[state_type][state_ref].holders: + for holder in holders: # An upgrader is itself among the holders through the # shared hold it upgrades from. if holder is transaction: @@ -4276,18 +4336,15 @@ async def _upgrade_lock( """ assert transaction.mode == Lock.Mode.SHARED - def on_grace() -> None: - self._abort_if_presumed_deadlock( - state_type, - state_ref, - transaction, - ) - await self._locks[state_type][state_ref].upgrade( deadline=LOCK_ACQUIRE_DEADLINE_DEFAULT, transaction=transaction, grace=TRANSACTION_DEADLOCK_GRACE, - on_grace=on_grace, + on_grace=self._presume_deadlock_on_grace( + state_type, + state_ref, + transaction, + ), ) transaction.mode = Lock.Mode.EXCLUSIVE @@ -5441,12 +5498,11 @@ async def _transaction_participant_start( # success / failure we resolve `transaction.acquired_lock` # so concurrent callers on the same transaction can # proceed (or propagate our failure). - def on_grace() -> None: - self._abort_if_presumed_deadlock( - state_type, - state_ref, - transaction, - ) + on_grace = self._presume_deadlock_on_grace( + state_type, + state_ref, + transaction, + ) try: if transaction.mode == Lock.Mode.SHARED: diff --git a/tests/reboot/transaction_tests.py b/tests/reboot/transaction_tests.py index 10687555..b08963af 100644 --- a/tests/reboot/transaction_tests.py +++ b/tests/reboot/transaction_tests.py @@ -30,7 +30,12 @@ ) from reboot.aio.internals.contextvars import Servicing, _servicing from reboot.aio.resolvers import NoResolver -from reboot.aio.state_managers import Lock, SidecarStateManager, StateManager +from reboot.aio.state_managers import ( + TRANSACTION_DEADLOCK_GRACE, + Lock, + SidecarStateManager, + StateManager, +) from reboot.aio.stubs import ( NestedTransactionUnaryRetriedCall, Stub, @@ -744,6 +749,130 @@ async def Transaction( (_, first_exited), (second_entered, _) = runs self.assertGreaterEqual(second_entered, first_exited) + async def test_waiting_behind_a_moving_lock_is_not_a_deadlock( + self, + ) -> None: + """A younger transaction queued behind a run of older + transactions that each hold a state's lock briefly waits far + longer than the grace period, yet is never presumed + deadlocked: at each grace check the holder is a different + transaction from the one at the previous check, which means the + lock is changing hands, not stuck. Only a holder that has held + the lock for a whole grace period is a presumed deadlock. + """ + hold = 4 * TRANSACTION_DEADLOCK_GRACE / 5 + + class TargetServicer(GeneralServicer): + + def authorizer(self): + return allow() + + async def ConstructorWriter( + self, + context: WriterContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + return GeneralResponse() + + # A root that calls the target's exclusive transaction. + async def ConstructorTransaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + await General.ref(request.content["target"]).Transaction( + context, + content={"hold": request.content["hold"]}, + ) + return GeneralResponse() + + # The target's transaction holds its lock for `hold`. + async def Transaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + await asyncio.sleep(float(request.content["hold"])) + return GeneralResponse() + + # Record every `TransactionShouldRetry` the client parses, to + # prove that nothing was presumed deadlocked. + transaction_should_retry = UnaryRetriedCall._transaction_should_retry + should_retries: list[errors_pb2.TransactionShouldRetry] = [] + + async def mock_transaction_should_retry(unary_retried_call): + should_retry = await transaction_should_retry(unary_retried_call) + if should_retry is not None: + should_retries.append(should_retry) + return should_retry + + with mock.patch( + 'reboot.aio.stubs.UnaryRetriedCall._transaction_should_retry', + mock_transaction_should_retry, + ): + await self.rbt.up( + Application(servicers=[TargetServicer]), + # Run each body once, so that the holders queue in the + # order the roots were started. + effect_validation=EffectValidation.DISABLED, + ) + context = self.rbt.create_external_context(name=self.id()) + + await General.ConstructorWriter(context, 'target') + + # Four older roots, started in order so that each queues + # behind the previous one, together hold the target's lock + # for several grace periods. + older = [] + for index in range(4): + older.append( + asyncio.create_task( + General.ConstructorTransaction( + context, + f'older-{index}', + content={ + "target": "target", + "hold": str(hold.total_seconds()), + }, + ) + ) + ) + # Let this root reach the target before the next starts, + # so that the target's queue is in root order. + await asyncio.sleep(0.05) + + # The youngest root queues last and holds the lock for no + # time at all once it gets it. + started = time.monotonic() + await General.ConstructorTransaction( + context, + 'youngest', + content={ + "target": "target", + "hold": "0" + }, + ) + waited = time.monotonic() - started + await asyncio.gather(*older) + + # The youngest waited out more than one grace period behind the + # older holders, and nothing was presumed deadlocked. A retry + # for `RESTART_DETECTED` can still happen right after the server + # starts, when a transaction's id predates the server's recovery + # timestamp; that is unrelated to the lock. + self.assertGreater(waited, TRANSACTION_DEADLOCK_GRACE.total_seconds()) + self.assertEqual( + [ + should_retry for should_retry in should_retries + if should_retry.reason == + errors_pb2.TransactionShouldRetry.PRESUMED_DEADLOCK + ], + [], + ) + async def test_shared_transactions_on_one_state_overlap(self) -> None: """Two transactions declared `shared`, each nested in its own root so that neither carries an idempotency key, take the same From e34a7dc9efd396bd95d6d55ce7cee154cdb40389 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Sun, 13 Sep 2026 20:01:19 +0000 Subject: [PATCH 2/2] Abort the second upgrader as a presumed deadlock carrying its age Two shared holders that both try to upgrade are a deadlock by construction: each waits for the other's shared hold to go. The lock already refused the second upgrade at once, but with a plain `Unavailable`, so the aborted transaction's retry carried no age, was the youngest again, and backed off first as if the cause were load. The lock now refuses it with `TransactionShouldRetry`, reason `PRESUMED_DEADLOCK`, carrying the transaction's age, and logs which transaction it lost to, the same way a deadlock found after the grace period is reported; a caller that passed no participant is still refused with `Unavailable`. The second to ask is still the one to go; choosing the younger would mean aborting the upgrader already parked in the lock from outside, which is the wound-wait machinery this design leaves out. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01EK2ybbxpNK5L7S5exgJ8fq --- reboot/aio/state_managers.py | 48 ++++++++--- tests/reboot/state_manager_tests.py | 44 +++++++++- tests/reboot/transaction_tests.py | 125 ++++++++++++++++++++++++++++ 3 files changed, 206 insertions(+), 11 deletions(-) diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index a8be60f2..f1856ff9 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -1711,9 +1711,12 @@ class Lock: promoted to `exclusive`. A caller can upgrade and skip other `exclusive` waiters, preserving the upgrading transaction's shared-consistent view of state. However, at most one upgrade - may be pending per lock; a second `upgrade(...)` raises - `SystemAborted(Unavailable())` immediately to avoid the deadlock - where two shared holders both want to upgrade. + may be pending per lock; a second `upgrade(...)` aborts + immediately, since two shared holders that both want to upgrade + are each waiting for the other's shared hold to go, which is a + deadlock by construction: with `SystemAborted(TransactionShouldRetry)` + carrying the transaction's age when the upgrader passed its + participant, and `SystemAborted(Unavailable())` otherwise. - `downgrade()` is the inverse: a caller that holds `exclusive` is demoted to `shared`, granting any queued `shared` waiters that @@ -2088,15 +2091,38 @@ async def _wait( if upgrade: # If we've already got a upgrader that is waiting we fail # fast because otherwise each upgrader would sit on their - # shared hold and we'd deadlock. + # shared hold and we'd deadlock. A transaction is asked to + # retry the way a presumed deadlock asks it, carrying its + # age, so that its retry skips the backoff and is not the + # youngest again; a caller without a participant is asked + # to retry as `Unavailable`. if self._upgrader is not None: + if transaction is None: + raise SystemAborted( + Unavailable(), + message=( + "Cannot upgrade shared lock to exclusive: " + "another transaction is already upgrading " + "the same state; retry the transaction." + ), + ) + pending = self._upgrader.transaction + message = ( + f"Transaction {transaction.root_id} cannot upgrade its " + "shared hold to exclusive: " + ( + f"transaction {pending.root_id} (age {pending.age})" + if pending is not None else "another transaction" + ) + " is already upgrading the same state, and each " + "would wait for the other's shared hold to go; " + "aborting so that it proceeds. Retry required." + ) + logger.warning(message) raise SystemAborted( - Unavailable(), - message=( - "Cannot upgrade shared lock to exclusive: " - "another transaction is already upgrading " - "the same state; retry the transaction." + TransactionShouldRetry( + reason=TransactionShouldRetry.PRESUMED_DEADLOCK, + retry_age=str(transaction.age), ), + message=message, ) self._upgrader = waiter else: @@ -4332,7 +4358,9 @@ async def _upgrade_lock( `(state_type, state_ref)` to exclusive, aborting `transaction` with `TransactionShouldRetry` (presumed deadlock) instead if it has waited longer than the grace period on a shared holder that - is older. + is older, or at once if another transaction's upgrade is + already pending, since the two would each wait for the other's + shared hold to go. """ assert transaction.mode == Lock.Mode.SHARED diff --git a/tests/reboot/state_manager_tests.py b/tests/reboot/state_manager_tests.py index b8db8840..c3cb1c0e 100644 --- a/tests/reboot/state_manager_tests.py +++ b/tests/reboot/state_manager_tests.py @@ -1112,7 +1112,8 @@ async def test_second_upgrade_fast_fails(self) -> None: await asyncio.sleep(0) self.assertFalse(first_upgrade_task.done()) - # The second upgrade attempt must fail immediately. + # The second upgrade attempt must fail immediately; without a + # participant it is asked to retry as `Unavailable`. with self.assertRaises(SystemAborted) as aborted: await lock.upgrade(deadline=None) self.assertEqual(type(aborted.exception.error), Unavailable) @@ -1123,6 +1124,47 @@ async def test_second_upgrade_fast_fails(self) -> None: await asyncio.wait_for(first_upgrade_task, timeout=1.0) lock.release_exclusive() + async def test_second_upgrade_by_a_transaction_retries_with_its_age( + self, + ) -> None: + """A transaction whose upgrade meets another pending upgrade is + asked to retry the way a presumed deadlock asks it, carrying its + age, since the two would each wait for the other's shared hold. + """ + lock = Lock() + # Stand-ins for the participants; the lock only reports them. + first = unittest.mock.Mock(spec=StateManager.Transaction) + first.root_id = uuid7(timestamp_ms=1000) + first.age = first.root_id + second = unittest.mock.Mock(spec=StateManager.Transaction) + second.root_id = uuid7(timestamp_ms=2000) + second.age = second.root_id + await lock.acquire_shared(deadline=None, transaction=first) + await lock.acquire_shared(deadline=None, transaction=second) + + first_upgrade_task = asyncio.create_task( + lock.upgrade(deadline=None, transaction=first) + ) + await asyncio.sleep(0) + self.assertFalse(first_upgrade_task.done()) + + with self.assertRaises(SystemAborted) as aborted: + await lock.upgrade(deadline=None, transaction=second) + error = aborted.exception.error + assert isinstance(error, TransactionShouldRetry) + self.assertEqual( + error.reason, TransactionShouldRetry.PRESUMED_DEADLOCK + ) + self.assertEqual(error.retry_age, str(second.age)) + assert aborted.exception.message is not None + self.assertIn(str(first.root_id), aborted.exception.message) + + # The second still holds shared; releasing it lets the first + # upgrade complete, and then the first releases exclusive. + lock.release_shared(transaction=second) + await asyncio.wait_for(first_upgrade_task, timeout=1.0) + lock.release_exclusive(transaction=first) + async def test_downgrade_when_sole_holder(self) -> None: lock = Lock() await lock.acquire_exclusive(deadline=None) diff --git a/tests/reboot/transaction_tests.py b/tests/reboot/transaction_tests.py index b08963af..c46d8a6b 100644 --- a/tests/reboot/transaction_tests.py +++ b/tests/reboot/transaction_tests.py @@ -946,6 +946,131 @@ async def SharedTransaction( self.assertTrue(both_entered.is_set()) + async def test_second_upgrader_retries_carrying_its_age(self) -> None: + """Two `shared` transactions on one state, each nested in its own + root, both write the state and so both upgrade. The two would + each wait for the other's shared hold to go, so the second to + upgrade aborts at once with `TransactionShouldRetry`, reason + `PRESUMED_DEADLOCK`, carrying its age, and its retry succeeds. + """ + # Each body waits for the other to have entered too, so that + # both hold the state shared when they go to write it. + both_entered = asyncio.Event() + entered = 0 + + class TargetServicer(GeneralServicer): + + def authorizer(self): + return allow() + + async def ConstructorWriter( + self, + context: WriterContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + return GeneralResponse() + + # A root that calls the target's shared transaction. + async def ConstructorTransaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + await General.ref(request.content["target"] + ).SharedTransaction(context) + return GeneralResponse() + + # Reads its state alongside the other, then writes it. + async def SharedTransaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + nonlocal entered + entered += 1 + if entered == 2: + both_entered.set() + await asyncio.wait_for(both_entered.wait(), timeout=10) + state.content["writes"] = str( + int(state.content.get("writes", "0")) + 1 + ) + return GeneralResponse() + + async def Reader( + self, + context: ReaderContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + return GeneralResponse(content=state.content) + + # Record every `TransactionShouldRetry` the client parses. + transaction_should_retry = UnaryRetriedCall._transaction_should_retry + should_retries: list[tuple[UnaryRetriedCall, + errors_pb2.TransactionShouldRetry]] = [] + + async def mock_transaction_should_retry(unary_retried_call): + should_retry = await transaction_should_retry(unary_retried_call) + if should_retry is not None: + should_retries.append((unary_retried_call, should_retry)) + return should_retry + + with mock.patch( + 'reboot.aio.stubs.UnaryRetriedCall._transaction_should_retry', + mock_transaction_should_retry, + ): + await self.rbt.up( + Application(servicers=[TargetServicer]), + # Run each body once, so that exactly two bodies meet at + # the barrier and exactly two upgrades are attempted. + effect_validation=EffectValidation.DISABLED, + ) + context = self.rbt.create_external_context(name=self.id()) + + await General.ConstructorWriter(context, 'target') + # Retries after the server starts are unrelated to the lock. + should_retries.clear() + + started = time.monotonic() + await asyncio.gather( + General.ConstructorTransaction( + context, + 'root-1', + content={"target": "target"}, + ), + General.ConstructorTransaction( + context, + 'root-2', + content={"target": "target"}, + ), + ) + elapsed = time.monotonic() - started + + # Both writes landed, one of them on a retry. + target = await General.ref('target').Reader(context) + self.assertEqual(target.content["writes"], "2") + + # The conflict was resolved at once rather than by the lock + # deadline, by the second upgrader asking to retry for a + # presumed deadlock with its age carried back on the retry. + self.assertLess(elapsed, 15) + deaths = [ + (call, should_retry) + for call, should_retry in should_retries + if should_retry.reason == + errors_pb2.TransactionShouldRetry.PRESUMED_DEADLOCK + ] + self.assertGreater(len(deaths), 0) + for call, should_retry in deaths: + self.assertNotEqual(should_retry.retry_age, '') + self.assertIn( + (TRANSACTION_RETRY_AGE_HEADER, should_retry.retry_age), + call._metadata, + ) + async def test_retry_carries_the_age_of_the_first_attempt(self) -> None: """A call that aborts with a `TransactionShouldRetry` whose reason skips backoff learns the transaction's age from the