Deflake concurrent_transactions_same_state by allowing retries

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 = [

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(

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,
             ),

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

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<string>();
   #everyoneArrived = new Gate();
 
   reset(expected: number): void {
     this.expected = expected;
-    this.arrived = 0;
+    this.arrived = new Set();
     this.#everyoneArrived.reset();
   }
 
-  async arrive(): Promise<void> {
-    this.arrived += 1;
-    if (this.arrived >= this.expected) {
+  async arrive(name: string): Promise<void> {
+    // 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<void> {
-    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<Counter.InnerResponse> {
-    await rendezvous.arrive();
+    await rendezvous.arrive(request.driverId);
     return { count: this.state.count };
   }
 

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 }),
       }),

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