diff --git a/changelog.d/19879.misc b/changelog.d/19879.misc
new file mode 100644
index 00000000000..be10ee05403
--- /dev/null
+++ b/changelog.d/19879.misc
@@ -0,0 +1 @@
+Update `HomeserverTestCase.get_success(...)` and friends to drive async Rust (Tokio runtime/thread pool).
diff --git a/synapse/api/auth_blocking.py b/synapse/api/auth_blocking.py
index 87918e15dca..1aed7f6de26 100644
--- a/synapse/api/auth_blocking.py
+++ b/synapse/api/auth_blocking.py
@@ -117,7 +117,7 @@ async def check_auth_blocking(
             # If the user is already part of the MAU cohort or a trial user
             if user_id:
                 timestamp = await self.store.user_last_seen_monthly_active(user_id)
-                if timestamp:
+                if timestamp is not None:
                     return
 
                 is_trial = await self.store.is_trial_user(user_id)
diff --git a/tests/federation/test_federation_sender.py b/tests/federation/test_federation_sender.py
index ced98a8b004..2a9c2f0fc41 100644
--- a/tests/federation/test_federation_sender.py
+++ b/tests/federation/test_federation_sender.py
@@ -37,6 +37,7 @@
 from synapse.storage.databases.main.events_worker import EventMetadata
 from synapse.types import JsonDict, ReadReceipt
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests.unittest import HomeserverTestCase
 
@@ -517,6 +518,24 @@ async def record_transaction(
         self.edus.extend(data["edus"])
         return {}
 
+    def wait_for_device_list_updates_to_be_sent(self) -> None:
+        """
+        Wait for the device list update EDU's to get pushed out over federation
+
+        For example, each login does a fire-and-forget (`LoginRestServlet` ->
+        `register_device` -> `notify_device_update` -> `handle_new_device_update` ->
+        `send_device_messages(hosts, immediate=False)`) which adds to the
+        `_DestinationWakeupQueue` which has a background process that sends depending on
+        how `federation_rr_transactions_per_room_per_second` is configured.
+
+        The default `federation_rr_transactions_per_room_per_second` is `50` (1s/50 ->
+        0.02s)
+        """
+        self.reactor.advance(
+            1.0
+            / self.hs.config.ratelimiting.federation_rr_transactions_per_room_per_second
+        )
+
     def test_send_device_updates(self) -> None:
         """Basic case: each device update should result in an EDU"""
         # create a device
@@ -527,9 +546,7 @@ def test_send_device_updates(self) -> None:
         self.assertEqual(len(self.edus), 1)
         stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D1", None)
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # a second call should produce no new device EDUs
         self.get_success(
@@ -568,7 +585,7 @@ def test_dont_send_device_updates_for_remote_users(self) -> None:
             )
         )
 
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # We shouldn't see an EDU for that update
         self.assertEqual(self.edus, [])
@@ -590,6 +607,8 @@ def test_upload_signatures(self) -> None:
         self.login(u1, "pass", device_id="D1")
         self.login(u1, "pass", device_id="D2")
 
+        self.wait_for_device_list_updates_to_be_sent()
+
         # expect two edus
         self.assertEqual(len(self.edus), 2)
         stream_id: int | None = None
@@ -600,9 +619,7 @@ def test_upload_signatures(self) -> None:
         device1_signing_key = self.generate_and_upload_device_signing_key(u1, "D1")
         device2_signing_key = self.generate_and_upload_device_signing_key(u1, "D2")
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect two more edus
         self.assertEqual(len(self.edus), 2)
@@ -637,9 +654,7 @@ def test_upload_signatures(self) -> None:
             e2e_handler.upload_signing_keys_for_user(u1, cross_signing_keys)
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect signing key update edu
         self.assertEqual(len(self.edus), 2)
@@ -662,9 +677,7 @@ def test_upload_signatures(self) -> None:
         )
         self.assertEqual(ret["failures"], {})
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect two edus, in one or two transactions. We don't know what order the
         # devices will be updated.
@@ -689,9 +702,7 @@ def test_delete_devices(self) -> None:
         self.login("user", "pass", device_id="D2")
         self.login("user", "pass", device_id="D3")
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect three edus
         self.assertEqual(len(self.edus), 3)
@@ -702,9 +713,7 @@ def test_delete_devices(self) -> None:
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect three edus, in an unknown order
         self.assertEqual(len(self.edus), 3)
@@ -730,15 +739,18 @@ def test_unreachable_server(self) -> None:
         # create devices
         u1 = self.register_user("user", "pass")
         self.login("user", "pass", device_id="D1")
+        # Wait some time in between each device list update as we want each of them to
+        # be attempted to be sent in their own transaction
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D2")
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D3")
+        self.reactor.advance(Duration(seconds=1).as_secs())
 
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         self.assertGreaterEqual(mock_send_txn.call_count, 4)
 
@@ -748,9 +760,7 @@ def test_unreachable_server(self) -> None:
             self.hs.get_federation_sender().send_device_messages(["host2"])
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # for each device, there should be a single update
         self.assertEqual(len(self.edus), 3)
@@ -777,16 +787,20 @@ def test_prune_outbound_device_pokes1(self) -> None:
         # create devices
         u1 = self.register_user("user", "pass")
         self.login("user", "pass", device_id="D1")
+        # Wait some time in between each device list update as we want each of them to
+        # be attempted to be sent in their own transaction
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D2")
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D3")
+        self.reactor.advance(Duration(seconds=1).as_secs())
 
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
+        # Ensure that we tried sending the device list update EDU's out
         self.assertGreaterEqual(mock_send_txn.call_count, 4)
 
         # run the prune job
@@ -801,9 +815,7 @@ def test_prune_outbound_device_pokes1(self) -> None:
             self.hs.get_federation_sender().send_device_messages(["host2"])
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # there should be a single update for this user.
         self.assertEqual(len(self.edus), 1)
@@ -835,15 +847,17 @@ def test_prune_outbound_device_pokes2(self) -> None:
         mock_send_txn.side_effect = AssertionError("fail")
 
         self.login("user", "pass", device_id="D2")
+        # Wait some time in between each device list update as we want each of them to
+        # be attempted to be sent in their own transaction
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D3")
-
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.reactor.advance(Duration(seconds=1).as_secs())
 
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
+        self.wait_for_device_list_updates_to_be_sent()
+
         self.assertGreaterEqual(mock_send_txn.call_count, 3)
 
         # run the prune job
@@ -858,9 +872,7 @@ def test_prune_outbound_device_pokes2(self) -> None:
             self.hs.get_federation_sender().send_device_messages(["host2"])
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # ... and we should get a single update for this user.
         self.assertEqual(len(self.edus), 1)
diff --git a/tests/rest/client/sliding_sync/test_sliding_sync.py b/tests/rest/client/sliding_sync/test_sliding_sync.py
index 2fd18f0e545..db1d4ac9c91 100644
--- a/tests/rest/client/sliding_sync/test_sliding_sync.py
+++ b/tests/rest/client/sliding_sync/test_sliding_sync.py
@@ -564,9 +564,16 @@ def test_wait_for_sync_token(self) -> None:
         )
         # Block for 10 seconds to make `notifier.wait_for_stream_token(from_token)`
         # timeout
+        #
+        # First, block for *almost* 10 seconds to make sure we are
+        # `notifier.wait_for_stream_token(from_token)`
         with self.assertRaises(TimedOutException):
             channel.await_result(timeout_ms=9900)
-        channel.await_result(timeout_ms=200)
+        # Then wait for the rest of the 10 second timeout, 9900 + 500 > 10000
+        #
+        # `notifier.wait_for_stream_token(from_token)` only checks every 500ms so we
+        # need to match that in order to make sure we hit the wake-up for sure.
+        channel.await_result(timeout_ms=500)
         self.assertEqual(channel.code, 200, channel.json_body)
 
         # We expect the next `pos` in the result to be the same as what we requested
diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py
index c3bfdf7c8d4..3173a90681b 100644
--- a/tests/rest/client/test_delayed_events.py
+++ b/tests/rest/client/test_delayed_events.py
@@ -26,6 +26,7 @@
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests import unittest
 from tests.server import FakeChannel
@@ -90,6 +91,10 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
             room=self.room_id, user=self.user2_user_id, tok=self.user2_access_token
         )
 
+        # Advance enough time where any requests we made during `prepare(...)` doesn't
+        # affect the rate-limits in the test itself
+        self.reactor.advance(Duration(days=1).as_secs())
+
     def test_delayed_events_empty_on_startup(self) -> None:
         self.assertListEqual([], self._get_delayed_events())
 
diff --git a/tests/rest/client/test_msc4388_rendezvous.py b/tests/rest/client/test_msc4388_rendezvous.py
index e99aeee7a05..f9b7f578f63 100644
--- a/tests/rest/client/test_msc4388_rendezvous.py
+++ b/tests/rest/client/test_msc4388_rendezvous.py
@@ -175,13 +175,11 @@ def test_rendezvous_public(self) -> None:
         self.assertEqual(channel.code, 200)
         rendezvous_id = channel.json_body["id"]
         sequence_token = channel.json_body["sequence_token"]
-        expires_in_ms = channel.json_body["expires_in_ms"]
-        self.assertGreater(expires_in_ms, 0)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         session_endpoint = rz_endpoint + f"/{rendezvous_id}"
 
         # We can get the data back
-        # Advances clock by 100ms
         channel = self.make_request(
             "GET",
             session_endpoint,
@@ -191,10 +189,9 @@ def test_rendezvous_public(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=bar")
         self.assertEqual(channel.json_body["sequence_token"], sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 100)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can update the data
-        # Advances clock by 100ms
         channel = self.make_request(
             "PUT",
             session_endpoint,
@@ -207,7 +204,6 @@ def test_rendezvous_public(self) -> None:
         new_sequence_token = channel.json_body["sequence_token"]
 
         # If we try to update it again with the old etag, it should fail
-        # Advances clock by 100ms
         channel = self.make_request(
             "PUT",
             session_endpoint,
@@ -221,7 +217,6 @@ def test_rendezvous_public(self) -> None:
         )
 
         # We should get the updated data
-        # Advances clock by 100ms
         channel = self.make_request(
             "GET",
             session_endpoint,
@@ -231,7 +226,7 @@ def test_rendezvous_public(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=baz")
         self.assertEqual(channel.json_body["sequence_token"], new_sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 400)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can delete the data
         channel = self.make_request(
@@ -376,8 +371,7 @@ def test_rendezvous_requires_authentication(self) -> None:
         self.assertEqual(channel.code, 200)
         rendezvous_id = channel.json_body["id"]
         sequence_token = channel.json_body["sequence_token"]
-        expires_in_ms = channel.json_body["expires_in_ms"]
-        self.assertEqual(expires_in_ms, 120000)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         session_endpoint = rz_endpoint + f"/{rendezvous_id}"
 
@@ -391,7 +385,7 @@ def test_rendezvous_requires_authentication(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=bar")
         self.assertEqual(channel.json_body["sequence_token"], sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 100)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can update the data without authentication
         channel = self.make_request(
@@ -414,7 +408,7 @@ def test_rendezvous_requires_authentication(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=baz")
         self.assertEqual(channel.json_body["sequence_token"], new_sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 300)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can delete the data without authentication
         channel = self.make_request(
diff --git a/tests/rest/client/test_sticky_events.py b/tests/rest/client/test_sticky_events.py
index a6e704fe8c2..015315a307b 100644
--- a/tests/rest/client/test_sticky_events.py
+++ b/tests/rest/client/test_sticky_events.py
@@ -58,7 +58,7 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         # Create a room
         self.room_id = self.helper.create_room_as(self.user_id, tok=self.token)
 
-    def _assert_event_sticky_for(self, event_id: str, sticky_ttl: int) -> None:
+    def _assert_event_sticky_for(self, event_id: str) -> None:
         channel = self.make_request(
             "GET",
             f"/rooms/{self.room_id}/event/{event_id}",
@@ -75,10 +75,10 @@ def _assert_event_sticky_for(self, event_id: str, sticky_ttl: int) -> None:
             event["unsigned"],
             f"No {EventUnsignedContentFields.STICKY_TTL} field in {event_id}; event not sticky: {event}",
         )
-        self.assertEqual(
+        self.assertGreater(
             event["unsigned"][EventUnsignedContentFields.STICKY_TTL],
-            sticky_ttl,
-            f"{event_id} had an unexpected sticky TTL: {event}",
+            0,
+            f"{event_id} had an unexpected sticky TTL (expected some value greater than 0): {event}",
         )
 
     def _assert_event_not_sticky(self, event_id: str) -> None:
@@ -112,17 +112,15 @@ def test_sticky_event_via_event_endpoint(self) -> None:
 
         # If we request the event immediately, it will still have
         # 1 minute of stickiness
-        # The other 100 ms is advanced in FakeChannel.await_result.
-        self._assert_event_sticky_for(event_id, 59_900)
+        self._assert_event_sticky_for(event_id)
 
-        # But if we advance time by 59.799 seconds...
-        # we will get the event on its last millisecond of stickiness
-        # The other 100 ms is advanced in FakeChannel.await_result.
-        self.reactor.advance(59.799)
-        self._assert_event_sticky_for(event_id, 1)
+        # But if we advance time by 59 seconds...
+        # we will get the event on its last second of stickiness
+        self.reactor.advance(Duration(seconds=59).as_secs())
+        self._assert_event_sticky_for(event_id)
 
         # Advancing time any more, the event is no longer sticky
-        self.reactor.advance(0.001)
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self._assert_event_not_sticky(event_id)
 
 
diff --git a/tests/rest/client/test_sync_sticky_events.py b/tests/rest/client/test_sync_sticky_events.py
index 7a38debdb95..2e52a0ad167 100644
--- a/tests/rest/client/test_sync_sticky_events.py
+++ b/tests/rest/client/test_sync_sticky_events.py
@@ -96,10 +96,9 @@ def test_single_sticky_event_appears_in_initial_sync(self) -> None:
             sticky_event_id,
             f"Sticky event {sticky_event_id} not found in sync timeline",
         )
-        self.assertEqual(
+        self.assertGreater(
             timeline_events[-1]["unsigned"][EventUnsignedContentFields.STICKY_TTL],
-            # The other 100 ms is advanced in FakeChannel.await_result.
-            59_900,
+            0,
         )
 
         self.assertNotIn(
@@ -129,7 +128,6 @@ def test_sticky_event_beyond_timeline_in_initial_sync(self) -> None:
         # that the /sync will get.
         regular_event_ids = []
         for i in range(10):
-            # (Note: each one advances time by 100ms)
             response = self.helper.send(
                 room_id=self.room_id,
                 body=f"regular message {i}",
@@ -138,7 +136,6 @@ def test_sticky_event_beyond_timeline_in_initial_sync(self) -> None:
             regular_event_ids.append(response["event_id"])
 
         # Send another sticky event
-        # (Note: this advances time by 100ms)
         second_sticky_response = self.helper.send_sticky_event(
             self.room_id,
             EventTypes.Message,
@@ -185,10 +182,9 @@ def test_sticky_event_beyond_timeline_in_initial_sync(self) -> None:
             f"Expected exactly 1 item in sticky events section, got {sticky_events}",
         )
         self.assertEqual(sticky_events[0]["event_id"], first_sticky_event_id)
-        self.assertEqual(
-            # The 'missing' 1100 ms were elapsed when sending events
+        self.assertGreater(
             sticky_events[0]["unsigned"][EventUnsignedContentFields.STICKY_TTL],
-            58_800,
+            0,
         )
 
         # Assertions for the second sticky event: should be only in timeline section
@@ -197,10 +193,9 @@ def test_sticky_event_beyond_timeline_in_initial_sync(self) -> None:
             second_sticky_event_id,
             f"Second sticky event {second_sticky_event_id} not found in sync timeline",
         )
-        self.assertEqual(
+        self.assertGreater(
             timeline_events[-1]["unsigned"][EventUnsignedContentFields.STICKY_TTL],
-            # The other 100 ms is advanced in FakeChannel.await_result.
-            59_900,
+            0,
         )
         # (sticky section: we already checked it only has 1 item and
         # that item was the first above)
diff --git a/tests/server.py b/tests/server.py
index ab4aae02185..15a7661c345 100644
--- a/tests/server.py
+++ b/tests/server.py
@@ -101,6 +101,7 @@
 from synapse.storage.prepare_database import prepare_database
 from synapse.types import ISynapseReactor, JsonDict
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 from synapse.util.json import json_encoder
 
 from tests.utils import (
@@ -301,15 +302,102 @@ def transport(self) -> "FakeChannel":
     def await_result(self, timeout_ms: int = 1000) -> None:
         """
         Wait until the request is finished.
+
+        Advances the Twisted reactor clock by 0.1s and suspending execution of the
+        Python thread (to allow other threads to do work) in a loop until we see a
+        result. We timeout when both the Twisted reactor clock has been advanced enough
+        AND we've done at-least 100 iterations (round-trips for other threads to get
+        work done).
+
+        The loop 1) allows `clock.call_later` scheduled callbacks to run if they are
+        scheduled to run now and 2) will also allow other threads to make progress. This
+        could be things spawned on the Twisted reactor threadpool or Tokio runtime
+        (async Rust code).
+
+        Args:
+            timeout_ms: The Twisted reactor time we wait until we raise a `TimedOutException`
         """
-        end_time = self._reactor.seconds() + timeout_ms / 1000.0
+        timeout = Duration(milliseconds=timeout_ms)
+
+        # TODO: Why?
         self._reactor.run()
 
+        # First, run anything that's scheduled now before we start looping and advancing
+        # non-zero time increments.
+        #
+        # Without this, if some request handler had some database queries followed by
+        # `self.hs.get_clock().sleep(Duration(seconds=1))`, and called
+        # `channel.await_result(timeout_ms=1000)`, it wouldn't be called because the
+        # first `self._reactor.advance(0.1)` would be first spent driving the database
+        # queries, and only leaving 0.9s remaining (0.1s shy of the sleep finishing) so
+        # the request would timeout.
+        #
+        # The goal is to remove the foot-guns and having to think about this for the
+        # standard cases.
+        #
+        # FIXME: Ideally, we'd advance by `0` but there is a handful of tests that
+        # assume that time advances in between requests and many requests complete from
+        # a single advance. Second best, we'd just advance by minuscule amount of time
+        # (`CLOCK_SCHEDULE_EPSILON`) but some tests assume at-least a millisecond in
+        # between as our timestamps are often recorded at the millisecond granularity
+        # (`origin_server_ts`, etc). It's a balance between test convenience of this
+        # helper and materializing test expectations so we may never fix this.
+        self._reactor.advance(Duration(milliseconds=1).as_secs())
+
+        # We only count the looping time (record the start after we advance once above)
+        start_time_seconds = self._reactor.seconds()
+        loop_count = 0
         while not self.is_finished():
-            if self._reactor.seconds() > end_time:
+            if (
+                # Exceeded the Twisted reactor time timeout
+                #
+                # We use `>=` for the reactor time condition as it's possible we advance
+                # exactly the `timeout` amount and we don't want to get stuck in an
+                # infinite loop
+                self._reactor.seconds() >= start_time_seconds + timeout.as_secs()
+                # 100 loops is arbitrary. This also makes the assumption that any work
+                # on other threads will finish before we give up after sleeping ~0.1s of
+                # real-time (100 * 0.001).
+                and loop_count > 100
+            ):
                 raise TimedOutException("Timed out waiting for request to finish.")
 
-            self._reactor.advance(0.1)
+            # Suspend execution of this thread to allow other threads to do work. This
+            # could be things spawned on the Twisted reactor threadpool or Tokio thread
+            # pool (async Rust code).
+            #
+            # Note: Python has a default thread switch interval (5ms for cpython) (see
+            # `sys.setswitchinterval(interval)`) but we still want this here as we're
+            # able to preempt and cause the thread context switch to happen faster.
+            # Also, without any real-time sleeping, this function would complete before
+            # the 5ms switch ever happened.
+            #
+            # After a few cycles, we use `time.sleep(0.001)` instead of `time.sleep(0)`
+            # to avoid tightlooping on the main thread (CPU 100%) because it's wasteful
+            # and may starve out other threads. 10 is arbitrary but many cases will have
+            # none or only a few round-trips so we can just try to go as fast as
+            # possible.
+            if loop_count < 10:
+                time.sleep(0)
+            else:
+                time.sleep(0.001)
+
+            # Advance the Twisted reactor and run any scheduled callbacks
+            #
+            # Don't advance the Twisted reactor clock further than the timeout duration
+            # as someone should increase the timeout if they expect things to take
+            # longer.
+            if self._reactor.seconds() < start_time_seconds + timeout.as_secs():
+                self._reactor.advance(0.1)
+            else:
+                # But we want to still keep running whatever might be getting scheduled
+                # to run now.
+                #
+                # For example from other threads, they may have scheduled something on
+                # the reactor to run (like `reactor.callFromThread(...)`)
+                self._reactor.advance(0)
+
+            loop_count += 1
 
     def extract_cookies(self, cookies: MutableMapping[str, str]) -> None:
         """Process the contents of any Set-Cookie headers in the response
diff --git a/tests/storage/test_client_ips.py b/tests/storage/test_client_ips.py
index bd68f2aaa1e..fd138335f0e 100644
--- a/tests/storage/test_client_ips.py
+++ b/tests/storage/test_client_ips.py
@@ -782,7 +782,11 @@ def _runtest(
                 device_id=device_id,
                 ip=expected_ip,
                 user_agent="Mozzila pizza",
-                last_seen=123456100,
+                # Note: The extra 1ms is from `make_request(...)` -> `await_result(...)`
+                #
+                # FIXME: This test shouldn't care about this internal detail (don't
+                # assert exact timing)
+                last_seen=123456001,
             ),
             r,
         )
diff --git a/tests/unittest.py b/tests/unittest.py
index 7dccff81c0f..202f7120ef0 100644
--- a/tests/unittest.py
+++ b/tests/unittest.py
@@ -773,7 +773,7 @@ def _wait_for_deferred(
             #
             # Note: Python has a default thread switch interval (5ms for cpython) (see
             # `sys.setswitchinterval(interval)`) but we still want this here as we're
-            # able to preempt and cause the thread context swtich to happen faster.
+            # able to preempt and cause the thread context switch to happen faster.
             # Also, without any real-time sleeping, this function would complete before
             # the 5ms switch ever happened.
             #
@@ -781,7 +781,7 @@ def _wait_for_deferred(
             # to avoid tightlooping on the main thread (CPU 100%) because it's wasteful
             # and may starve out other threads. 10 is arbitrary but many cases will have
             # none or only a few round-trips so we can just try to go as fast as
-            # posssible.
+            # possible.
             if loop_count < 10:
                 time.sleep(0)
             else: