Fix two transaction hangs, and other CI flakes

.github/workflows/build_and_test.yml

@@ -351,10 +351,19 @@ jobs:
         QUERY='tests(//tests/...)'
 
         # 2) Exclude these tags by default for MacOS.
+        #
+        # A test carries `platform-independent` when it exercises no
+        # behavior that could differ between Linux and MacOS, or when
+        # whatever platform-dependent behavior it does exercise is also
+        # exercised by a `medium`-or-smaller test that still runs here.
+        # Such a test spends time on our slowest and most expensive
+        # runners for signal that the Linux jobs — which keep running
+        # every test — already give us.
         QUERY="$QUERY \
         except attr(\"tags\",\"requires-docker\",//tests/...) \
         except attr(\"tags\",\"macos_not_supported\",//tests/...) \
         except attr(\"tags\",\"requires-linux-x86\",//tests/...) \
+        except attr(\"tags\",\"platform-independent\",//tests/...) \
         except attr(\"tags\",\"manual\",//tests/...)"
 
         # 3) If this is a PR labelled "reboot-release", also drop flaky

reboot/aio/state_managers.py

@@ -1547,6 +1547,12 @@ class Lock:
       transaction. Passing `deadline=None` means wait forever or until
       the task is cancelled.
 
+    - A waiter that has given up, i.e., whose `deadline` elapsed or
+      whose task was cancelled, is dropped rather than granted (see
+      `_Waiter.abandoned()`). Granting to one would leave the lock
+      held by nobody, and every later acquire on the state it guards
+      would then fail or block forever.
+
     The lock is not `asyncio.Task` aware because it is meant to be
     used across multiple different calls on the same state, i.e.,
     across transactions. Instead, it tracks shared as a count and
@@ -1567,6 +1573,21 @@ def __init__(self, mode: Lock.Mode) -> None:
                 asyncio.get_event_loop().create_future()
             )
 
+        def abandoned(self) -> bool:
+            """`True` once this waiter can no longer take the hold it
+            is queued for, because its acquire deadline elapsed or its
+            task was cancelled. It is still on the queue only because
+            it has not been resumed yet to remove itself. Granting to
+            it would mark the lock held with nobody left to release
+            it, leaving the state this lock guards unusable for the
+            life of the process, so such a waiter must be dropped.
+
+            A waiter that is granted is removed from the queue at the
+            same time its future is resolved, and nothing else
+            resolves a queued waiter's future, so a waiter that is
+            still queued and whose future is resolved has given up."""
+            return self.future.done()
+
     def __init__(self) -> None:
         # Number of shared holders.
         self._shared: int = 0
@@ -1707,6 +1728,17 @@ def _remove_waiter(self, waiter: Lock._Waiter) -> None:
         except ValueError:
             pass
 
+    def _remove_upgrader(self, waiter: Lock._Waiter) -> None:
+        """Empties the upgrade slot, but only while `waiter` is the
+        waiter occupying it. A waiter that has given up may already
+        have been dropped from the slot, and by the time it cleans up
+        after itself another holder may have registered an upgrade of
+        its own there. That one must be left in place: emptying the
+        slot would leave it registered nowhere, and a waiter nothing
+        holds a reference to is a waiter nothing can ever grant."""
+        if self._upgrader is waiter:
+            self._upgrader = None
+
     @overload
     async def _wait(
         self,
@@ -1765,7 +1797,7 @@ async def _wait(
             await asyncio.wait_for(waiter.future, timeout=timeout)
         except asyncio.TimeoutError:
             if upgrade:
-                self._upgrader = None
+                self._remove_upgrader(waiter)
             else:
                 self._remove_waiter(waiter)
             raise SystemAborted(
@@ -1783,12 +1815,18 @@ async def _wait(
                 waiter.future.done() and not waiter.future.cancelled() and
                 waiter.future.exception() is None
             ):
-                if mode == Lock.Mode.SHARED:
+                if upgrade:
+                    # An upgrade whose caller is interrupted leaves
+                    # that caller holding the shared hold it upgrades
+                    # from, so hand that back rather than releasing
+                    # the exclusive hold outright.
+                    self.downgrade()
+                elif mode == Lock.Mode.SHARED:
                     self.release_shared()
                 else:
                     self.release_exclusive()
             elif upgrade:
-                self._upgrader = None
+                self._remove_upgrader(waiter)
             else:
                 self._remove_waiter(waiter)
             raise
@@ -1811,15 +1849,21 @@ def _maybe_grant_next(self, *, released: Lock.Mode) -> None:
             # other remaining shared holders, bypassing any queued
             # exclusive waiters.
             assert released == Lock.Mode.SHARED
-            if self._shared == 1:
+            if self._upgrader.abandoned():
+                # The upgrader has given up, so forget it and grant as
+                # if it had never asked. It keeps the shared hold it
+                # upgrades from and releases that itself.
+                self._upgrader = None
+            elif self._shared == 1:
                 upgrader = self._upgrader
                 self._upgrader = None
                 self._shared = 0
                 self._exclusive = True
-                if not upgrader.future.done():
-                    upgrader.future.set_result(None)
-            # While we have an upgrader nothing else gets granted.
-            return
+                upgrader.future.set_result(None)
+                return
+            else:
+                # While we have an upgrader nothing else gets granted.
+                return
 
         # Nothing to do if we released a shared hold but still have
         # shared holders.
@@ -1833,6 +1877,11 @@ def _maybe_grant_next(self, *, released: Lock.Mode) -> None:
         # Grant waiters in FIFO ordering.
         while self._waiters:
             waiter = self._waiters[0]
+            if waiter.abandoned():
+                # Drop the waiter and try the next one rather than
+                # taking the lock on its behalf (see `abandoned()`).
+                self._waiters.pop(0)
+                continue
             if waiter.mode == Lock.Mode.EXCLUSIVE:
                 # We may have granted a shared waiter below if the
                 # _first_ waiter was not exclusive and thus we've now
@@ -1841,14 +1890,12 @@ def _maybe_grant_next(self, *, released: Lock.Mode) -> None:
                     return
                 self._waiters.pop(0)
                 self._exclusive = True
-                if not waiter.future.done():
-                    waiter.future.set_result(None)
+                waiter.future.set_result(None)
                 return
             # The mode is `Lock.Mode.SHARED`.
             self._waiters.pop(0)
             self._shared += 1
-            if not waiter.future.done():
-                waiter.future.set_result(None)
+            waiter.future.set_result(None)
             # Continue granting a cohort of shared waiters up until
             # the first exclusive waiter.
 
@@ -5347,13 +5394,43 @@ async def _transaction_coordinator_abort(
         participants: Participants,
         coordinator_state_ref: Optional[StateRef],
     ):
-        """Aborts a transaction for which we are the coordinator: (1) cleans
-        up the "preparing" record that may be in the database; (2)
-        best-effort sends `Abort` RPCs to all participants; and (3)
-        resolves the `_coordinator_participants` future to `None` so
-        any participant "watch control loops" observe the abort, then
-        deletes the future entry.
+        """Aborts a transaction for which we are the coordinator: (1)
+        resolves the `_coordinator_participants` future to `None` and
+        drops its entry so any participant "watch control loops"
+        observe the abort; (2) cleans up the "preparing" record that
+        may be in the database; and (3) best-effort sends `Abort` RPCs
+        to all participants.
         """
+        # To indicate that the transaction has aborted we resolve the
+        # future to `None`, so a participant `Watch` that already passed
+        # the "is it still in the map" check and is awaiting the future
+        # observes the abort rather than finding itself still listed to
+        # commit. Both statements are synchronous and come before
+        # anything that can be interrupted, so from here on every
+        # watcher has an answer no matter where a cancellation lands.
+        #
+        # Answering before the database cleanup below lands is safe,
+        # because an abort is not something a coordinator records: its
+        # durable outcomes are "preparing" and "absent", and neither
+        # says a transaction committed. A coordinator only gets here
+        # while the transaction is not durably prepared -- either its
+        # `_transaction_coordinator_complete()` did not return, or it
+        # is re-preparing, which asserts as much -- so there is no
+        # committed outcome for an early answer to contradict. Should
+        # this process die before the cleanup lands, recovery finds
+        # "preparing" and re-prepares, the participants that have
+        # already aborted report that they never prepared, and the
+        # transaction aborts again.
+        #
+        # Not expecting the future to ever be cancelled, see:
+        # https://github.com/reboot-dev/mono/issues/3241
+        assert not self._coordinator_participants[transaction_id].cancelled()
+        self._coordinator_participants[transaction_id].set_result(None)
+
+        # Remove transaction so that participants "watch control loop"
+        # will determine that the transaction has aborted!
+        del self._coordinator_participants[transaction_id]
+
         # Clean up the "preparing" record that may be in the database
         # when the coordinator stored the transaction participants
         # (and `preparing=True`). The database write may or may not
@@ -5418,21 +5495,6 @@ async def abort(state_type: StateTypeName, state_ref: StateRef):
             for (state_type, state_ref) in participants.should_prepare()
         )
 
-        # To indicate that the transaction has aborted we resolve the
-        # future to `None`, so a participant `Watch` that already passed
-        # the "is it still in the map" check and is awaiting the future
-        # observes the abort rather than finding itself still listed to
-        # commit.
-        #
-        # Not expecting the future to ever be cancelled, see:
-        # https://github.com/reboot-dev/mono/issues/3241
-        assert not self._coordinator_participants[transaction_id].cancelled()
-        self._coordinator_participants[transaction_id].set_result(None)
-
-        # Remove transaction so that participants "watch control loop"
-        # will determine that the transaction has aborted!
-        del self._coordinator_participants[transaction_id]
-
     async def _transaction_participant_watch(
         self,
         application_id: ApplicationId,

tests/reboot/agents/pydantic_ai/BUILD.bazel

@@ -6,6 +6,10 @@ py_test(
     timeout = "long",
     srcs = [":agent_tests.py"],
     main = "agent_tests.py",
+    # This exercises the `pydantic-ai` integration against a mocked
+    # model (`FunctionModel`) on the in-process `Reboot()` test
+    # harness; neither has platform-dependent behavior.
+    tags = ["platform-independent"],
     deps = [
         requirement("mcp"),
         requirement("pydantic-ai-slim"),

tests/reboot/cli/init/BUILD.bazel

@@ -96,7 +96,15 @@ sh_test(
     # Additionally, in all our example tests, we initiate LocalEnvoy on the
     # default secure port 9991. We require 'exclusive' to not run multiple
     # tests, that use the same port, concurrently.
+    #
+    # Installing the locally built Reboot npm packages, `rbt generate`
+    # and `npx rbt dev run` are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room-nodejs:test`, and the `rbt
+    # init` flow by `init_sh_test_python310`; both keep running
+    # there, and what is left is the content of the Node.js
+    # templates.
     tags = [
         "exclusive",
+        "platform-independent",
     ],
 )

tests/reboot/examples/agent-wiki/BUILD.bazel

@@ -30,11 +30,6 @@ osx_env.update({
 
 sh_test_in_working_directory(
     name = "test",
-    # MacOS runners need the full Bazel "large" budget because
-    # `uv sync` cold-fetches the locked dependency tree (and
-    # possibly a managed Python interpreter) into the test
-    # sandbox, which exceeds the default "medium" 300s timeout.
-    size = "large",
     data = test_packages + [
         ":test_packages",
         # The actual test script will be given access to all of
@@ -51,7 +46,16 @@ sh_test_in_working_directory(
     # In all our example tests, we initiate LocalEnvoy on the
     # default secure port 9991. We require 'exclusive' to not run
     # multiple tests, that use the same port, concurrently.
-    tags = ["exclusive"],
+    #
+    # The `uv sync`, `rbt generate`, `pytest` and `rbt dev run
+    # --terminate-after-health-check` steps are the same ones that
+    # `//tests/reboot/examples/chat-room:test` runs, and that test
+    # keeps running on MacOS; only the application code in between
+    # differs, so this adds no MacOS-specific coverage.
+    tags = [
+        "exclusive",
+        "platform-independent",
+    ],
     working_directory = "reboot/examples/agent-wiki/",
 )
 
@@ -79,5 +83,12 @@ sh_test_in_working_directory(
     ],
     env = frontend_env,
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/agent-wiki/",
 )

tests/reboot/examples/ai-chat-counter-dashboard/BUILD.bazel

@@ -37,5 +37,12 @@ sh_test_in_working_directory(
     # is generated from are copied into the test sandbox; the script
     # `cd`s into `frontend/` itself.
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/ai-chat-counter-dashboard/",
 )

tests/reboot/examples/ai-chat-counter/BUILD.bazel

@@ -37,5 +37,12 @@ sh_test_in_working_directory(
     # is generated from are copied into the test sandbox; the script
     # `cd`s into `frontend/` itself.
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/ai-chat-counter/",
 )

tests/reboot/examples/bank-pydantic/BUILD.bazel

@@ -97,7 +97,16 @@ sh_test_in_working_directory(
     # from are copied into the test sandbox; the script `cd`s into
     # `frontend/mobile/` itself.
     script = "frontend/mobile/.tests/react_native_test.sh",
-    tags = ["exclusive"],
+    # `npx tsc --noEmit` produces the same result on MacOS as on
+    # Linux; generating the mobile client is exercised on MacOS by
+    # `//tests/reboot/cli/generate:mobile_out_specified_tests_py`,
+    # and installing the locally built Reboot npm packages by
+    # `//tests/reboot/examples/chat-room-nodejs:test`; both keep
+    # running there.
+    tags = [
+        "exclusive",
+        "platform-independent",
+    ],
     working_directory = "reboot/examples/bank-pydantic/",
 )
 
@@ -135,5 +144,12 @@ sh_test_in_working_directory(
     ],
     env = frontend_env,
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/bank-pydantic/",
 )

tests/reboot/examples/bank-zod/BUILD.bazel

@@ -70,5 +70,12 @@ sh_test_in_working_directory(
     ],
     env = general_env,
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/bank-zod/",
 )

tests/reboot/examples/bank/BUILD.bazel

@@ -72,5 +72,12 @@ sh_test_in_working_directory(
     ],
     env = frontend_env,
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/bank/",
 )

tests/reboot/examples/boutique/BUILD.bazel

@@ -80,5 +80,12 @@ sh_test_in_working_directory(
     ],
     env = frontend_env,
     script = "frontend/.tests/type_check.sh",
+    # `tsc` and the frontend bundler produce the same result on MacOS
+    # as on Linux, and the `rbt generate` and `npm install` steps on
+    # the way there are exercised on MacOS by
+    # `//tests/reboot/examples/chat-room:test` and
+    # `//tests/reboot/examples/chat-room-nodejs:test`, which keep
+    # running there.
+    tags = ["platform-independent"],
     working_directory = "reboot/examples/boutique/",
 )