From 868aba366bb8058c23df96a0a091703620db9d63 Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:05:25 +0000 Subject: [PATCH] `tests`: rendezvous on driver names, not on arrival counts `test_nested_transactions_on_one_state_are_parallel` failed intermittently on the MacOS arm64 CI runner, and always by overshooting: `40 != 20` (every driver counted twice), `59 !== 20`, and once a 300s timeout. It never reproduced on Linux. The `Rendezvous` that the test met at counted invocations of `inner()`. Reboot guarantees that a transaction happens once, not that the code inside it is invoked once: when a root transaction aborts, the `ExternalContext` stub retries it, re-running `call_inner` and with it the nested `inner()` on the shared `COUNTER_ID`. On a slow runner such an abort is easy to come by, whether from a lock acquire that exceeds `LOCK_ACQUIRE_DEADLINE_DEFAULT` or from a participant raising `TransactionShouldRetryWithoutBackoff`, so the count climbed past `CONCURRENCY`. Counting invocations was also a weaker check than it looked: a driver counted twice can stand in for one that has not arrived at all, opening the meeting point while only 19 transactions really overlap. `Rendezvous` now collects the names of the drivers that have arrived and opens once it holds `expected` distinct ones, so a retried transaction arrives idempotently and the assertion stays exact. `inner` takes the driver's name in its request, because it runs on the peer's state, which is the same state for every driver, and so cannot work out who called it from its own context. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VTP67vX9mZusQwMrNCZw1r --- .../BUILD.bazel | 6 ++++ .../servicer.py | 36 +++++++++++++------ .../servicer_api.py | 9 ++++- .../test.py | 8 +++-- .../servicer.ts | 33 +++++++++++------ .../servicer_api.ts | 7 +++- .../test.ts | 6 +++- 7 files changed, 80 insertions(+), 25 deletions(-) diff --git a/tests/reboot/pydantic/concurrent_transactions_same_state/BUILD.bazel b/tests/reboot/pydantic/concurrent_transactions_same_state/BUILD.bazel index ca7c5e466..6d3e89729 100644 --- a/tests/reboot/pydantic/concurrent_transactions_same_state/BUILD.bazel +++ b/tests/reboot/pydantic/concurrent_transactions_same_state/BUILD.bazel @@ -28,6 +28,12 @@ py_library( py_test( name = "test_py", + # This test drives 20 transactions into contention on a single + # state, and a transaction that aborts under that contention retries + # from the top. On the MacOS arm64 CI runner a retry storm has taken + # it well past the 300s a `medium` test gets, from a usual ~15s, so + # this one gets the 900s of a `large` test to work in. + size = "large", srcs = ["test.py"], main = "test.py", deps = [ diff --git a/tests/reboot/pydantic/concurrent_transactions_same_state/servicer.py b/tests/reboot/pydantic/concurrent_transactions_same_state/servicer.py index b9f8da3eb..006169423 100644 --- a/tests/reboot/pydantic/concurrent_transactions_same_state/servicer.py +++ b/tests/reboot/pydantic/concurrent_transactions_same_state/servicer.py @@ -7,6 +7,7 @@ ) from tests.reboot.pydantic.concurrent_transactions_same_state.servicer_api import ( CountResponse, + DriverRequest, PeerRequest, ) from tests.reboot.pydantic.concurrent_transactions_same_state.servicer_api_rbt import ( @@ -17,23 +18,31 @@ class Rendezvous: - """A meeting point that only opens once `expected` callers have - arrived, so a caller can only get through if all of them are - inside at the same time. Serialized callers deadlock instead.""" + """A meeting point that only opens once callers with `expected` + distinct names have arrived, so a caller can only get through if + all of them are inside at the same time. Serialized callers + deadlock instead.""" def __init__(self) -> None: self.expected = 0 - self.arrived = 0 + self.arrived: set[str] = set() self.everyone_arrived = asyncio.Event() def reset(self, expected: int) -> None: self.expected = expected - self.arrived = 0 + self.arrived = set() self.everyone_arrived.clear() - async def arrive(self) -> None: - self.arrived += 1 - if self.arrived >= self.expected: + async def arrive(self, name: str) -> None: + # Names rather than a count, because a transaction that aborts + # is retried from the top and so runs a method it already ran + # again: Reboot promises that a transaction happens once, not + # that the code in it is invoked once. A count of invocations + # would climb past `expected`, and worse, could open the + # meeting point on one caller counted twice standing in for + # another that has not arrived at all. + self.arrived.add(name) + if len(self.arrived) >= self.expected: self.everyone_arrived.set() await self.everyone_arrived.wait() @@ -80,13 +89,20 @@ async def call_inner( context: TransactionContext, request: PeerRequest, ) -> None: - await Counter.ref(request.peer_id).inner(context) + # Every driver calls `inner` on the same peer, so the name of + # the state `inner` runs on says nothing about which driver is + # inside it; hand `inner` this driver's name instead. + await Counter.ref(request.peer_id).inner( + context, + driver_id=context.state_id, + ) async def inner( self, context: TransactionContext, + request: DriverRequest, ) -> CountResponse: - await rendezvous.arrive() + await rendezvous.arrive(request.driver_id) return CountResponse(count=self.state.count) async def call_parked_increment( diff --git a/tests/reboot/pydantic/concurrent_transactions_same_state/servicer_api.py b/tests/reboot/pydantic/concurrent_transactions_same_state/servicer_api.py index 577a03c23..4086c58a1 100644 --- a/tests/reboot/pydantic/concurrent_transactions_same_state/servicer_api.py +++ b/tests/reboot/pydantic/concurrent_transactions_same_state/servicer_api.py @@ -22,6 +22,10 @@ class PeerRequest(Model): peer_id: str = Field(tag=1) +class DriverRequest(Model): + driver_id: str = Field(tag=1) + + api = API( Counter=Type( state=CounterState, @@ -53,8 +57,11 @@ class PeerRequest(Model): response=None, mcp=None, ), + # Takes the name of the driver that called it, because it runs + # on the peer's state rather than the driver's and so cannot + # work that out from its own context. inner=Transaction( - request=None, + request=DriverRequest, response=CountResponse, mcp=None, ), diff --git a/tests/reboot/pydantic/concurrent_transactions_same_state/test.py b/tests/reboot/pydantic/concurrent_transactions_same_state/test.py index 3830a6ce1..93a3d4774 100644 --- a/tests/reboot/pydantic/concurrent_transactions_same_state/test.py +++ b/tests/reboot/pydantic/concurrent_transactions_same_state/test.py @@ -46,7 +46,11 @@ async def test_nested_transactions_on_one_state_are_parallel(self) -> None: shared mode, so many of them may be running inside one state at the same time. The rendezvous only opens once every one of them has arrived, so this can only finish if they really do - overlap.""" + overlap. + + The rendezvous counts the names it has seen rather than the + arrivals, so a driver whose transaction aborted and was + retried is still one arrival.""" driver_ids = [f"driver-{index}" for index in range(CONCURRENCY)] await asyncio.gather( *( @@ -66,7 +70,7 @@ async def test_nested_transactions_on_one_state_are_parallel(self) -> None: ) ) - self.assertEqual(rendezvous.arrived, CONCURRENCY) + self.assertEqual(len(rendezvous.arrived), CONCURRENCY) async def test_finishing_transaction_keeps_anothers_write(self) -> None: """A transaction's write survives another transaction on the diff --git a/tests/reboot/zod/concurrent_transactions_same_state/servicer.ts b/tests/reboot/zod/concurrent_transactions_same_state/servicer.ts index 45e975ca1..0976ae66c 100644 --- a/tests/reboot/zod/concurrent_transactions_same_state/servicer.ts +++ b/tests/reboot/zod/concurrent_transactions_same_state/servicer.ts @@ -29,23 +29,31 @@ class Gate { } } -// A meeting point that only opens once `expected` callers have -// arrived, so a caller can only get through if all of them are inside -// at the same time. Serialized callers deadlock instead. +// A meeting point that only opens once callers with `expected` +// distinct names have arrived, so a caller can only get through if all +// of them are inside at the same time. Serialized callers deadlock +// instead. class Rendezvous { expected = 0; - arrived = 0; + arrived = new Set(); #everyoneArrived = new Gate(); reset(expected: number): void { this.expected = expected; - this.arrived = 0; + this.arrived = new Set(); this.#everyoneArrived.reset(); } - async arrive(): Promise { - this.arrived += 1; - if (this.arrived >= this.expected) { + async arrive(name: string): Promise { + // Names rather than a count, because a transaction that aborts is + // retried from the top and so runs a method it already ran again: + // Reboot promises that a transaction happens once, not that the + // code in it is invoked once. A count of invocations would climb + // past `expected`, and worse, could open the meeting point on one + // caller counted twice standing in for another that has not + // arrived at all. + this.arrived.add(name); + if (this.arrived.size >= this.expected) { this.#everyoneArrived.open(); } await this.#everyoneArrived.opened; @@ -95,14 +103,19 @@ export class CounterServicer extends Counter.Servicer { context: TransactionContext, request: Counter.CallInnerRequest ): Promise { - await Counter.ref(request.peerId).inner(context, {}); + // Every driver calls `inner` on the same peer, so the name of the + // state `inner` runs on says nothing about which driver is inside + // it; hand `inner` this driver's name instead. + await Counter.ref(request.peerId).inner(context, { + driverId: context.stateId, + }); } async inner( context: TransactionContext, request: Counter.InnerRequest ): Promise { - await rendezvous.arrive(); + await rendezvous.arrive(request.driverId); return { count: this.state.count }; } diff --git a/tests/reboot/zod/concurrent_transactions_same_state/servicer_api.ts b/tests/reboot/zod/concurrent_transactions_same_state/servicer_api.ts index 8786e5880..a18dc8b1f 100644 --- a/tests/reboot/zod/concurrent_transactions_same_state/servicer_api.ts +++ b/tests/reboot/zod/concurrent_transactions_same_state/servicer_api.ts @@ -35,8 +35,13 @@ export const Counter = { }), response: z.void(), }), + // Takes the name of the driver that called it, because it runs + // on the peer's state rather than the driver's and so cannot work + // that out from its own context. inner: transaction({ - request: z.object({}), + request: z.object({ + driverId: z.string().meta({ tag: 1 }), + }), response: z.object({ count: z.number().meta({ tag: 1 }), }), diff --git a/tests/reboot/zod/concurrent_transactions_same_state/test.ts b/tests/reboot/zod/concurrent_transactions_same_state/test.ts index 62ec4fc36..ee77c988d 100644 --- a/tests/reboot/zod/concurrent_transactions_same_state/test.ts +++ b/tests/reboot/zod/concurrent_transactions_same_state/test.ts @@ -35,6 +35,10 @@ test("Concurrent transactions on one state", async (t) => { // so many of them may be running inside one state at the same time. // The rendezvous only opens once every one of them has arrived, so // this can only finish if they really do overlap. + // + // The rendezvous counts the names it has seen rather than the + // arrivals, so a driver whose transaction aborted and was retried is + // still one arrival. await t.test("nested transactions on one state are parallel", async (t) => { const { rbt, context } = await setUp(); @@ -58,7 +62,7 @@ test("Concurrent transactions on one state", async (t) => { ) ); - assert.equal(rendezvous.arrived, CONCURRENCY); + assert.equal(rendezvous.arrived.size, CONCURRENCY); }); // A transaction's write survives another transaction on the same