Update HomeserverTestCase.get_success(...) and friends to drive async Rust (Tokio runtime/thread pool) (#19871)

This means you can use `get_success(...)` anywhere regardless
of what kind of work needs to be done.

Spawning from adding some more async Rust things in
https://github.com/element-hq/synapse/pull/19846 and wanting something
more standard instead of the custom `till_deferred_has_result(...)` that
has crept in to a few files.

Alternative to https://github.com/element-hq/synapse/pull/19867 spurred
on by [this
comment](https://github.com/element-hq/synapse/pull/19867#discussion_r3441774685)
from @erikjohnston


### How does this work?

Previously, `get_success(...)` just ran in a hot-loop advancing the
Twisted reactor clock which didn't give any time for other threads to do
some work or acquire the GIL if necessary (whenever there is a hand-off
from Rust to Python, we need the GIL).

Now, `get_success(...)` loops until we see a result (until we hit the
~0.1s real-time timeout). In the loop, we call
[`time.sleep(0)`](https://docs.python.org/3/library/time.html#time.sleep)
which will "Suspend execution of the calling thread [...]" (CPU and GIL)
to allow other threads to do some work. Then like before, we advance the
Twisted reactor clock to run any scheduled callbacks which includes
anything the other threads may have scheduled.


### Does this slow down the entire test suite?

Seems just as fast as before. There is minutes variance in what we had
before and after but both are within the same range of each other.

(see PR for actual before/after timings)

changelog.d/19871.misc

@@ -0,0 +1 @@
+Update `HomeserverTestCase.get_success(...)` and friends to drive async Rust (Tokio runtime/thread pool).

synapse/http/client.py

@@ -87,8 +87,7 @@ from synapse.logging.opentracing import set_tag, start_active_span, tags
 from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import ISynapseReactor, StrSequence
 from synapse.util.async_helpers import timeout_deferred
-from synapse.util.clock import Clock
-from synapse.util.duration import Duration
+from synapse.util.clock import CLOCK_SCHEDULE_EPSILON, Clock
 from synapse.util.json import json_decoder
 
 if TYPE_CHECKING:
@@ -163,11 +162,6 @@ def _is_ip_blocked(
     return False
 
 
-# The delay used by the scheduler to schedule tasks "as soon as possible", while
-# still allowing other tasks to run between runs.
-_EPSILON = Duration(microseconds=1)
-
-
 def _make_scheduler(clock: Clock) -> Callable[[Callable[[], object]], IDelayedCall]:
     """Makes a schedular suitable for a Cooperator using the given reactor.
 
@@ -176,7 +170,7 @@ def _make_scheduler(clock: Clock) -> Callable[[Callable[[], object]], IDelayedCa
 
     def _scheduler(x: Callable[[], object]) -> IDelayedCall:
         return clock.call_later(
-            _EPSILON,
+            CLOCK_SCHEDULE_EPSILON,
             x,
         )
 

synapse/util/clock.py

@@ -62,6 +62,19 @@ this setting won't inherit the log level from the parent logger.
 logging.setLoggerClass(original_logger_class)
 
 
+CLOCK_SCHEDULE_EPSILON = Duration(microseconds=1)
+"""
+The smallest value we can use that will schedule tasks "as soon as possible", while
+still allowing other tasks to run between runs.
+
+This should be a non-zero value as the Twisted Reactor API does not specify how calls
+get scheduled. If we used `0`, a weird reactor implementation could run it immediately
+or run it any order with the other calls that are scheduled now.
+
+We want the semantics of run this in the "next reactor iteration".
+"""
+
+
 def _try_wakeup_deferred(d: Deferred) -> None:
     """Try to wake up a deferred, but ignore any exceptions raised by the
     callback. This is useful when we want to wake up a deferred that may have

tests/app/test_homeserver_shutdown.py

@@ -76,6 +76,13 @@ class HomeserverCleanShutdownTestCase(HomeserverTestCase):
 
         self.get_success(shutdown())
 
+        # XXX: There can be a few already dispatched database queries (from normal
+        # background tasks in Synapse) and the threadless `ThreadPool` that we use in
+        # tests uses *untracked* clock calls to pass database results back so `shutdown`
+        # doesn't cancel those calls. This is a quirk of our test infrastructure
+        # (threadless `ThreadPool`) so this kind of "hack" is fine.
+        self.reactor.advance(0)
+
         # Cleanup the internal reference in our test case
         del self.hs
 
@@ -106,7 +113,7 @@ class HomeserverCleanShutdownTestCase(HomeserverTestCase):
         # Pump the background updates by a single iteration, just to ensure any extra
         # resources it uses have been started.
         store = weakref.proxy(self.hs.get_datastores().main)
-        self.get_success(store.db_pool.updates.do_next_background_update(False), by=0.1)
+        self.get_success(store.db_pool.updates.do_next_background_update(False))
 
         hs_ref = weakref.ref(self.hs)
 
@@ -127,6 +134,13 @@ class HomeserverCleanShutdownTestCase(HomeserverTestCase):
 
         self.get_success(shutdown())
 
+        # XXX: There can be a few already dispatched database queries (from normal
+        # background tasks in Synapse) and the threadless `ThreadPool` that we use in
+        # tests uses *untracked* clock calls to pass database results back so `shutdown`
+        # doesn't cancel those calls. This is a quirk of our test infrastructure
+        # (threadless `ThreadPool`) so this kind of "hack" is fine.
+        self.reactor.advance(0)
+
         # Cleanup the internal reference in our test case
         del self.hs
 

tests/crypto/test_keyring.py

@@ -499,7 +499,7 @@ class ServerKeyFetcherTestCase(unittest.HomeserverTestCase):
         res = key_json[testverifykey_id]
         self.assertIsNotNone(res)
         assert res is not None
-        self.assertEqual(res.added_ts, self.reactor.seconds() * 1000)
+        self.assertEqual(res.added_ts, self.clock.time_msec())
         self.assertEqual(res.valid_until_ts, VALID_UNTIL_TS)
 
         # we expect it to be encoded as canonical json *before* it hits the db
@@ -614,7 +614,7 @@ class PerspectivesKeyFetcherTestCase(unittest.HomeserverTestCase):
         res = key_json[testverifykey_id]
         self.assertIsNotNone(res)
         assert res is not None
-        self.assertEqual(res.added_ts, self.reactor.seconds() * 1000)
+        self.assertEqual(res.added_ts, self.clock.time_msec())
         self.assertEqual(res.valid_until_ts, VALID_UNTIL_TS)
 
         self.assertEqual(res.key_json, canonicaljson.encode_canonical_json(response))
@@ -732,7 +732,7 @@ class PerspectivesKeyFetcherTestCase(unittest.HomeserverTestCase):
         res = key_json[testverifykey_id]
         self.assertIsNotNone(res)
         assert res is not None
-        self.assertEqual(res.added_ts, self.reactor.seconds() * 1000)
+        self.assertEqual(res.added_ts, self.clock.time_msec())
         self.assertEqual(res.valid_until_ts, VALID_UNTIL_TS)
 
         self.assertEqual(res.key_json, canonicaljson.encode_canonical_json(response))

tests/handlers/test_federation.py

@@ -357,7 +357,6 @@ class FederationTestCase(unittest.FederatingHomeserverTestCase):
                 event.room_version,
             ),
             exc=LimitExceededError,
-            by=0.5,
         )
 
     def _build_and_send_join_event(

tests/handlers/test_oauth_delegation.py

@@ -21,15 +21,13 @@
 
 import json
 import threading
-import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Any, ClassVar, Coroutine, Generator, TypeVar, Union
+from typing import Any, ClassVar, TypeVar
 from unittest.mock import AsyncMock, Mock
 from urllib.parse import parse_qs
 
 from parameterized.parameterized import parameterized_class
 
-from twisted.internet.defer import Deferred, ensureDeferred
 from twisted.internet.testing import MemoryReactor
 
 from synapse.api.auth.mas import MasDelegatedAuth
@@ -204,31 +202,6 @@ class MasAuthDelegation(HomeserverTestCase):
     def device_scope(self) -> str:
         return self.device_scope_prefix + DEVICE
 
-    def till_deferred_has_result(
-        self,
-        awaitable: Union[
-            "Coroutine[Deferred[Any], Any, T]",
-            "Generator[Deferred[Any], Any, T]",
-            "Deferred[T]",
-        ],
-    ) -> "Deferred[T]":
-        """Wait until a deferred has a result.
-
-        This is useful because the Rust HTTP client will resolve the deferred
-        using reactor.callFromThread, which are only run when we call
-        reactor.advance.
-        """
-        deferred = ensureDeferred(awaitable)
-        tries = 0
-        while not deferred.called:
-            time.sleep(0.1)
-            self.reactor.advance(0)
-            tries += 1
-            if tries > 100:
-                raise Exception("Timed out waiting for deferred to resolve")
-
-        return deferred
-
     def default_config(self) -> dict[str, Any]:
         config = super().default_config()
         config["public_baseurl"] = BASE_URL
@@ -278,11 +251,7 @@ class MasAuthDelegation(HomeserverTestCase):
             "expires_in": 60,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.user.to_string(), USER_ID)
         self.assertEqual(requester.device_id, DEVICE)
@@ -301,11 +270,7 @@ class MasAuthDelegation(HomeserverTestCase):
             "username": USERNAME,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.user.to_string(), USER_ID)
         self.assertEqual(requester.device_id, DEVICE)
@@ -326,9 +291,7 @@ class MasAuthDelegation(HomeserverTestCase):
         }
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             InvalidClientTokenError,
         )
         self.assertEqual(failure.value.code, 401)
@@ -343,9 +306,7 @@ class MasAuthDelegation(HomeserverTestCase):
         }
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             AuthError,
         )
         # This is a 500, it should never happen really
@@ -361,9 +322,7 @@ class MasAuthDelegation(HomeserverTestCase):
         }
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             InvalidClientTokenError,
         )
         self.assertEqual(failure.value.code, 401)
@@ -372,9 +331,7 @@ class MasAuthDelegation(HomeserverTestCase):
         self.server.introspection_response = {}
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             SynapseError,
         )
         self.assertEqual(failure.value.code, 503)
@@ -389,11 +346,7 @@ class MasAuthDelegation(HomeserverTestCase):
             "device_id": DEVICE,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.device_id, DEVICE)
 
@@ -406,11 +359,7 @@ class MasAuthDelegation(HomeserverTestCase):
             "expires_in": 60,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.user.to_string(), USER_ID)
         self.assertTrue(self.get_success(self._auth.is_server_admin(requester)))
@@ -435,17 +384,15 @@ class MasAuthDelegation(HomeserverTestCase):
         request.requestHeaders.getRawHeaders = mock_getRawHeaders()
 
         # The first CS-API request causes a successful introspection
-        self.get_success(
-            self.till_deferred_has_result(self._auth.get_user_by_req(request))
-        )
+        self.get_success(self._auth.get_user_by_req(request))
         self.assertEqual(self.server.calls, 1)
 
         # Sleep for 60 seconds so the token expires.
         self.reactor.advance(60.0)
 
         # Now the CS-API request fails because the token expired
-        self.assertFailure(
-            self.till_deferred_has_result(self._auth.get_user_by_req(request)),
+        self.get_failure(
+            self._auth.get_user_by_req(request),
             InvalidClientTokenError,
         )
         # Ensure another introspection request was not sent

tests/handlers/test_oidc.py

@@ -960,7 +960,7 @@ class OidcHandlerTestCase(HomeserverTestCase):
         # advance the clock a bit before we start, so we aren't working with zero
         # timestamps.
         self.reactor.advance(1000)
-        start_time = self.reactor.seconds()
+        start_time_s = int(self.reactor.seconds())
         ret = self.get_success(self.provider._exchange_code(code, code_verifier=""))
 
         self.assertEqual(ret, token)
@@ -981,8 +981,8 @@ class OidcHandlerTestCase(HomeserverTestCase):
         self.assertEqual(claims["aud"], ISSUER)
         self.assertEqual(claims["iss"], "DEFGHI")
         self.assertEqual(claims["sub"], CLIENT_ID)
-        self.assertEqual(claims["iat"], start_time)
-        self.assertGreater(claims["exp"], start_time)
+        self.assertEqual(claims["iat"], start_time_s)
+        self.assertGreater(claims["exp"], start_time_s)
 
         # check the rest of the POSTed data
         self.assertEqual(args["grant_type"], ["authorization_code"])

tests/handlers/test_presence.py

@@ -951,8 +951,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
         self.get_success(
             worker_presence_handler.user_syncing(
                 self.user_id, self.device_id, True, PresenceState.ONLINE
-            ),
-            by=0.1,
+            )
         )
 
         # Check that if we wait a while without telling the handler the user has
@@ -1270,8 +1269,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
                 "dev-1",
                 affect_presence=dev_1_state != PresenceState.OFFLINE,
                 presence_state=dev_1_state,
-            ),
-            by=0.01,
+            )
         )
 
         # 2. Wait half the idle timer.
@@ -1285,8 +1283,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
                 "dev-2",
                 affect_presence=dev_2_state != PresenceState.OFFLINE,
                 presence_state=dev_2_state,
-            ),
-            by=0.01,
+            )
         )
 
         # 4. Assert the expected presence state.
@@ -1311,8 +1308,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
                     "dev-3",
                     affect_presence=True,
                     presence_state=PresenceState.ONLINE,
-                ),
-                by=0.01,
+                )
             ):
                 pass
 
@@ -1507,8 +1503,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
                 "dev-1",
                 affect_presence=dev_1_state != PresenceState.OFFLINE,
                 presence_state=dev_1_state,
-            ),
-            by=0.1,
+            )
         )
 
         # 2. Sync with the second device.
@@ -1518,8 +1513,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
                 "dev-2",
                 affect_presence=dev_2_state != PresenceState.OFFLINE,
                 presence_state=dev_2_state,
-            ),
-            by=0.1,
+            )
         )
 
         # 3. Assert the expected presence state.
@@ -1625,8 +1619,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase):
         self.get_success(
             worker_to_sync_against.get_presence_handler().user_syncing(
                 self.user_id, self.device_id, True, PresenceState.ONLINE
-            ),
-            by=0.1,
+            )
         )
 
         # Check against the main process that the user's presence did not change.

tests/handlers/test_profile.py

@@ -200,7 +200,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
             self.assertEqual(membership[state_tuple].content["displayname"], "Frank")
 
             # Let's be sure we are over the delay introduced by slow_update_membership
-            self.get_success(self.clock.sleep(Duration(milliseconds=20)), by=1)
+            self.reactor.advance(Duration(milliseconds=20).as_secs())
 
             membership = self.get_success(
                 self.storage_controllers.state.get_current_state(
@@ -278,7 +278,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
 
             # Let's be sure we are over the delay introduced by slow_update_membership
             # and that the task was not executed as expected
-            self.get_success(self.clock.sleep(Duration(milliseconds=20)), by=1)
+            self.reactor.advance(Duration(milliseconds=20).as_secs())
 
             membership = self.get_success(
                 self.storage_controllers.state.get_current_state(
@@ -299,8 +299,10 @@ class ProfileTestCase(unittest.HomeserverTestCase):
                 )
             )
 
+            # Wait for the `TaskScheduler.SCHEDULE_INTERVAL`
+            self.reactor.advance(Duration(minutes=1).as_secs())
             # Let's be sure we are over the delay introduced by slow_update_membership
-            self.get_success(self.clock.sleep(Duration(milliseconds=20)), by=1)
+            self.reactor.advance(Duration(milliseconds=20).as_secs())
 
             # Updates should have been resumed from room 2 after the restart
             # so room 1 should not have been updated this time

tests/handlers/test_room_member.py

@@ -71,7 +71,6 @@ class TestJoinsLimitedByPerRoomRateLimiter(FederatingHomeserverTestCase):
                 action=Membership.JOIN,
             ),
             LimitExceededError,
-            by=0.5,
         )
 
     @override_config({"rc_joins_per_room": {"per_second": 0.1, "burst_count": 2}})
@@ -213,7 +212,6 @@ class TestJoinsLimitedByPerRoomRateLimiter(FederatingHomeserverTestCase):
                     remote_room_hosts=[self.OTHER_SERVER_NAME],
                 ),
                 LimitExceededError,
-                by=0.5,
             )
 
     # TODO: test that remote joins to a room are rate limited.
@@ -281,7 +279,6 @@ class TestReplicatedJoinsLimitedByPerRoomRateLimiter(BaseMultiWorkerStreamTestCa
                 action=Membership.JOIN,
             ),
             LimitExceededError,
-            by=0.5,
         )
 
         # Try to join as Chris on the original worker. Should get denied because Alice
@@ -294,7 +291,6 @@ class TestReplicatedJoinsLimitedByPerRoomRateLimiter(BaseMultiWorkerStreamTestCa
                 action=Membership.JOIN,
             ),
             LimitExceededError,
-            by=0.5,
         )
 
 

tests/handlers/test_send_email.py

@@ -146,7 +146,7 @@ class SendEmailHandlerTestCaseIPv4(HomeserverTestCase):
         )
 
         # the message should now get delivered
-        self.get_success(d, by=0.1)
+        self.get_success(d)
 
         # check it arrived
         self.assertEqual(len(message_delivery.messages), 1)
@@ -213,7 +213,7 @@ class SendEmailHandlerTestCaseIPv4(HomeserverTestCase):
         )
 
         # the message should now get delivered
-        self.get_success(d, by=0.1)
+        self.get_success(d)
 
         # check it arrived
         self.assertEqual(len(message_delivery.messages), 1)

tests/handlers/test_typing.py

@@ -248,6 +248,14 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase):
             )
         )
 
+        # Wait for the EDU to get pushed out over federation
+        #
+        # `started_typing` is fire-and-forget and handles the remote federation part as
+        # part of a background process which isn't waited on.
+        #
+        # We're specifically waiting for the database queries in the background process
+        self.reactor.advance(0)
+
         self.mock_federation_client.put_json.assert_called_once_with(
             "farm",
             path="/_matrix/federation/v1/send/1000000",
@@ -367,6 +375,14 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase):
             [call(StreamKeyType.TYPING, 1, rooms=[ROOM_ID])]
         )
 
+        # Wait for the EDU to get pushed out over federation
+        #
+        # `stopped_typing` is fire-and-forget and handles the remote federation part as
+        # part of a background process which isn't waited on.
+        #
+        # We're specifically waiting for the database queries in the background process
+        self.reactor.advance(0)
+
         self.mock_federation_client.put_json.assert_called_once_with(
             "farm",
             path="/_matrix/federation/v1/send/1000000",

tests/handlers/test_user_directory.py

@@ -555,7 +555,15 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase):
         # Process the leave and join in one go.
         dir_handler.update_user_directory = True
         dir_handler.notify_new_event()
-        self.wait_for_background_updates()
+
+        # Wait for the user directory to update
+        #
+        # `notify_new_event` is fire-and-forget and the actual changes happen as part of
+        # a background process loop which isn't waited on.
+        #
+        # We're specifically waiting for the database queries in the `notify_new_event`
+        # background process.
+        self.reactor.advance(0)
 
         # The user sharing tables should have been updated.
         public3 = self.get_success(self.user_dir_helper.get_users_in_public_rooms())
@@ -1124,7 +1132,6 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase):
 
         # Alice leaves the other. She should still be in the directory.
         self.helper.leave(room2, alice, tok=alice_token)
-        self.wait_for_background_updates()
         users, in_public, in_private = self.get_success(
             self.user_dir_helper.get_tables()
         )

tests/media/test_media_storage.py

@@ -132,12 +132,7 @@ class MediaStorageTests(unittest.HomeserverTestCase):
 
         # This uses a real blocking threadpool so we have to wait for it to be
         # actually done :/
-        x = defer.ensureDeferred(test_ensure_media())
-
-        # Hotloop until the threadpool does its job...
-        self.wait_on_thread(x)
-
-        self.get_success(x)
+        self.get_success(test_ensure_media())
 
 
 @attr.s(auto_attribs=True, slots=True, frozen=True)

tests/replication/tcp/test_handler.py

@@ -147,6 +147,12 @@ class ChannelsTestCase(BaseMultiWorkerStreamTestCase):
         # ... but worker1 finishing (and so sending an update) should.
         self.get_success(ctx_worker1.__aexit__(None, None, None))
 
+        # Wait for the stream position to be replicated to the master process
+        #
+        # Replication travels over `FakeTransport` and we're specifically flushing the
+        # write
+        self.reactor.advance(0)
+
         self.assertTrue(d.called)
 
     def test_wait_for_stream_position_rdata(self) -> None:
@@ -206,6 +212,12 @@ class ChannelsTestCase(BaseMultiWorkerStreamTestCase):
         # Finish the context manager, triggering the data to be sent to master.
         self.get_success(ctx_worker1.__aexit__(None, None, None))
 
+        # Wait for the stream position to be replicated to the master process
+        #
+        # Replication travels over `FakeTransport` and we're specifically flushing the
+        # write
+        self.reactor.advance(0)
+
         # Master should get told about `next_token2`, so the deferred should
         # resolve.
         self.assertTrue(d.called)

tests/replication/test_federation_ack.py

@@ -81,6 +81,14 @@ class FederationAckTestCase(HomeserverTestCase):
             )
         )
 
+        # Wait for the FEDERATION_ACK to be sent
+        #
+        # `on_rdata` handles this as part of a fire-and-forget background process (see
+        # `FederationSenderHandler.update_token`)
+        #
+        # We're specifically waiting for the database queries in the background process
+        self.reactor.advance(0)
+
         # now check that the FEDERATION_ACK was sent
         mock_connection.send_command.assert_called_once()
         cmd = mock_connection.send_command.call_args[0][0]

tests/rest/admin/test_user.py

@@ -59,7 +59,7 @@ from synapse.rest.client import (
 from synapse.server import HomeServer
 from synapse.storage.databases.main.client_ips import LAST_SEEN_GRANULARITY
 from synapse.types import JsonDict, UserID, create_requester
-from synapse.util.clock import Clock
+from synapse.util.clock import CLOCK_SCHEDULE_EPSILON, Clock
 
 from tests import unittest
 from tests.replication._base import BaseMultiWorkerStreamTestCase
@@ -5850,10 +5850,16 @@ class UserRedactionBackgroundTaskTestCase(BaseMultiWorkerStreamTestCase):
         self.assertEqual(channel.code, 200)
         id = channel.json_body.get("redact_id")
 
-        # Need 1 tick as we send 1 replication request per original event
-        # and each wait must be >= `_EPSILON` from `http/client.py`
+        # `/redact` just schedules a background task that runs in the background
+        # (fire-and-forget) so we need to do the waiting here.
+        #
+        # Need 1 tick as we send 1 replication request for the redaction of each
+        # original event. The replication request body is streamed by a `Cooperator`
+        # that uses the clock to schedule each chunk at a tiny *non-zero* delay
+        # (`CLOCK_SCHEDULE_EPSILON`), so we need to actually advance the clock for it to
+        # fire.
         for _ in range(len(original_event_ids)):
-            self.reactor.advance(0.001)
+            self.reactor.advance(CLOCK_SCHEDULE_EPSILON.as_secs())
 
         # Verify the HTTP `redact_status` endpoint reports completion.
         channel2 = self.make_request(

tests/server.py

@@ -255,7 +255,7 @@ class FakeChannel:
         def _produce() -> None:
             if self._producer:
                 self._producer.resumeProducing()
-                self._reactor.callLater(0.1, _produce)
+                self._reactor.callLater(0.0, _produce)
 
         if not streaming:
             self._reactor.callLater(0.0, _produce)
@@ -940,7 +940,7 @@ class FakeTransport:
             # mypy ignored here because:
             #   - this is part of the test infrastructure (outside of Synapse) so tracking
             #     these calls for for homeserver shutdown doesn't make sense.
-            d.addCallback(lambda x: self._reactor.callLater(0.1, _produce))  # type: ignore[call-later-not-tracked,call-overload]
+            d.addCallback(lambda x: self._reactor.callLater(0.0, _produce))  # type: ignore[call-later-not-tracked,call-overload]
 
         if not streaming:
             # mypy ignored here because:

tests/storage/test_background_update.py

@@ -59,8 +59,8 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
         self.store = self.hs.get_datastores().main
 
     async def update(self, progress: JsonDict, count: int) -> int:
-        duration_ms = 10
-        await self.clock.sleep(Duration(milliseconds=count * duration_ms))
+        fake_work_duration = Duration(seconds=1)
+        await self.clock.sleep(fake_work_duration)
         progress = {"my_key": progress["my_key"] + 1}
         await self.store.db_pool.runInteraction(
             "update_progress",
@@ -86,10 +86,15 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
 
         self.update_handler.side_effect = self.update
         self.update_handler.reset_mock()
-        res = self.get_success(
-            self.updates.do_next_background_update(False),
-            by=0.02,
+        background_update_d = ensureDeferred(
+            self.updates.do_next_background_update(False)
         )
+        # Wait for database queries to run in `do_next_background_update(...)` so the
+        # background update actually gets scheduled
+        self.reactor.advance(0)
+        # Wait for the actual background update `fake_work_duration`
+        self.reactor.advance(Duration(seconds=1).as_secs())
+        res = self.get_success(background_update_d)
         self.assertFalse(res)
 
         # on the first call, we should get run with the default background update size
@@ -143,10 +148,15 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
 
         self.update_handler.side_effect = self.update
         self.update_handler.reset_mock()
-        res = self.get_success(
-            self.updates.do_next_background_update(False),
-            by=0.01,
+        background_update_d = ensureDeferred(
+            self.updates.do_next_background_update(False)
         )
+        # Wait for database queries to run in `do_next_background_update(...)` so the
+        # background update actually gets scheduled
+        self.reactor.advance(0)
+        # Wait for the actual background update `fake_work_duration`
+        self.reactor.advance(Duration(seconds=1).as_secs())
+        res = self.get_success(background_update_d)
         self.assertFalse(res)
 
         # on the first call, we should get run with the default background update size specified in the config
@@ -265,10 +275,15 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
 
         self.update_handler.side_effect = self.update
         self.update_handler.reset_mock()
-        res = self.get_success(
-            self.updates.do_next_background_update(False),
-            by=0.02,
+        background_update_d = ensureDeferred(
+            self.updates.do_next_background_update(False)
         )
+        # Wait for database queries to run in `do_next_background_update(...)` so the
+        # background update actually gets scheduled
+        self.reactor.advance(0)
+        # Wait for the actual background update `fake_work_duration`
+        self.reactor.advance(Duration(seconds=1).as_secs())
+        res = self.get_success(background_update_d)
         self.assertFalse(res)
 
         # the first update was run with the default batch size, this should be run with 500ms as the
@@ -298,9 +313,6 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
         """
         Test that the minimum batch size set in the config is used
         """
-        # a very long-running individual update
-        duration_ms = 50
-
         self.get_success(
             self.store.db_pool.simple_insert(
                 "background_updates",
@@ -310,7 +322,8 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
 
         # Run the update with the long-running update item
         async def update_long(progress: JsonDict, count: int) -> int:
-            await self.clock.sleep(Duration(milliseconds=count * duration_ms))
+            very_long_fake_work_duration = Duration(seconds=5)
+            await self.clock.sleep(very_long_fake_work_duration)
             progress = {"my_key": progress["my_key"] + 1}
             await self.store.db_pool.runInteraction(
                 "update_progress",
@@ -322,10 +335,15 @@ class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
 
         self.update_handler.side_effect = update_long
         self.update_handler.reset_mock()
-        res = self.get_success(
-            self.updates.do_next_background_update(False),
-            by=1,
+        background_update_d = ensureDeferred(
+            self.updates.do_next_background_update(False)
         )
+        # Wait for database queries to run in `do_next_background_update(...)` so the
+        # background update actually gets scheduled
+        self.reactor.advance(0)
+        # Wait for the actual background update `very_long_fake_work_duration`
+        self.reactor.advance(Duration(seconds=5).as_secs())
+        res = self.get_success(background_update_d)
         self.assertFalse(res)
 
         # the first update was run with the default batch size, this should be run with minimum batch size

tests/storage/test_event_chain.py

@@ -755,7 +755,7 @@ class EventChainBackgroundUpdateTestCase(HomeserverTestCase):
         ):
             iterations += 1
             self.get_success(
-                self.store.db_pool.updates.do_next_background_update(False), by=0.1
+                self.store.db_pool.updates.do_next_background_update(False)
             )
 
         # Ensure that we did actually take multiple iterations to process the
@@ -814,7 +814,7 @@ class EventChainBackgroundUpdateTestCase(HomeserverTestCase):
         ):
             iterations += 1
             self.get_success(
-                self.store.db_pool.updates.do_next_background_update(False), by=0.1
+                self.store.db_pool.updates.do_next_background_update(False)
             )
 
         # Ensure that we did actually take multiple iterations to process the

tests/synapse_rust/test_http_client.py

@@ -15,9 +15,8 @@ import logging
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Any, Coroutine, Generator, TypeVar, Union
+from typing import Any, TypeVar
 
-from twisted.internet.defer import Deferred, ensureDeferred
 from twisted.internet.testing import MemoryReactor
 
 from synapse.logging.context import (
@@ -118,31 +117,6 @@ class HttpClientTestCase(HomeserverTestCase):
             for callbable, args, kwargs in triggers:
                 callbable(*args, **kwargs)
 
-    def till_deferred_has_result(
-        self,
-        awaitable: Union[
-            "Coroutine[Deferred[Any], Any, T]",
-            "Generator[Deferred[Any], Any, T]",
-            "Deferred[T]",
-        ],
-    ) -> "Deferred[T]":
-        """Wait until a deferred has a result.
-
-        This is useful because the Rust HTTP client will resolve the deferred
-        using reactor.callFromThread, which are only run when we call
-        reactor.advance.
-        """
-        deferred = ensureDeferred(awaitable)
-        tries = 0
-        while not deferred.called:
-            time.sleep(0.1)
-            self.reactor.advance(0)
-            tries += 1
-            if tries > 100:
-                raise Exception("Timed out waiting for deferred to resolve")
-
-        return deferred
-
     def _check_current_logcontext(self, expected_logcontext_string: str) -> None:
         context = current_context()
         assert isinstance(context, LoggingContext) or isinstance(context, _Sentinel), (
@@ -168,7 +142,7 @@ class HttpClientTestCase(HomeserverTestCase):
             raw_response = json_decoder.decode(resp_body.decode("utf-8"))
             self.assertEqual(raw_response, {"ok": True})
 
-        self.get_success(self.till_deferred_has_result(do_request()))
+        self.get_success(do_request())
         self.assertEqual(self.server.calls, 1)
 
     def test_request_response_limit_exceeded(self) -> None:
@@ -183,8 +157,8 @@ class HttpClientTestCase(HomeserverTestCase):
                 response_limit=1,
             )
 
-        self.assertFailure(
-            self.till_deferred_has_result(do_request()),
+        self.get_failure(
+            do_request(),
             RuntimeError,
         )
         self.assertEqual(self.server.calls, 1)
@@ -227,8 +201,15 @@ class HttpClientTestCase(HomeserverTestCase):
             # Now wait for the function under test to have run
             with PreserveLoggingContext():
                 while not callback_finished:
-                    # await self.hs.get_clock().sleep(0)
-                    time.sleep(0.1)
+                    # Allow the async Rust to run
+                    #
+                    # Suspend execution of this thread to allow other the Tokio thread
+                    # pool to do work.
+                    time.sleep(0)
+                    # Advance the Twisted reactor and run any scheduled callbacks
+                    #
+                    # In terms of other threads, they may have scheduled something on the
+                    # reactor to run (like `reactor.callFromThread(...)`)
                     self.reactor.advance(0)
 
             # check that the logcontext is left in a sane state.

tests/unittest.py

@@ -48,6 +48,7 @@ import signedjson.key
 import unpaddedbase64
 from typing_extensions import Concatenate, ParamSpec
 
+from twisted.internet import defer
 from twisted.internet.defer import Deferred, ensureDeferred
 from twisted.internet.testing import MemoryReactor, MemoryReactorClock
 from twisted.python.failure import Failure
@@ -76,7 +77,7 @@ from synapse.rest import RegisterServletsFunc
 from synapse.server import HomeServer
 from synapse.storage.keys import FetchKeyResult
 from synapse.types import ISynapseReactor, JsonDict, Requester, UserID, create_requester
-from synapse.util.clock import Clock
+from synapse.util.clock import CLOCK_SCHEDULE_EPSILON, Clock
 from synapse.util.httpresourcetree import create_resource_tree
 
 from tests.server import (
@@ -474,27 +475,13 @@ class HomeserverTestCase(TestCase):
         # Reset to not use frozen dicts.
         events.USE_FROZEN_DICTS = False
 
-    def wait_on_thread(self, deferred: Deferred, timeout: int = 10) -> None:
-        """
-        Wait until a Deferred is done, where it's waiting on a real thread.
-        """
-        start_time = time.time()
-
-        while not deferred.called:
-            if start_time + timeout < time.time():
-                raise ValueError("Timed out waiting for threadpool")
-            self.reactor.advance(0.01)
-            time.sleep(0.01)
-
     def wait_for_background_updates(self) -> None:
         """Block until all background database updates have completed."""
         store = self.hs.get_datastores().main
         while not self.get_success(
             store.db_pool.updates.has_completed_background_updates()
         ):
-            self.get_success(
-                store.db_pool.updates.do_next_background_update(False), by=0.1
-            )
+            self.get_success(store.db_pool.updates.do_next_background_update(False))
 
     def make_homeserver(
         self, reactor: ThreadedMemoryReactorClock, clock: Clock
@@ -736,21 +723,165 @@ class HomeserverTestCase(TestCase):
         # whole chain to completion.
         self.reactor.pump([by] * 100)
 
-    def get_success(self, d: Awaitable[TV], by: float = 0.0) -> TV:
+    def _wait_for_deferred(
+        self,
+        d: "Deferred[Any]",
+    ) -> None:
+        """
+        Wait for the deferred to finish or raise.
+
+        Does not advance time in the Twisted reactor clock but will loop 100 times
+        waiting for a result. 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:
+            d: Twisted Deferred
+
+        Raises:
+            defer.TimeoutError: If the timeout expires before the deferred completes.
+        """
+        # Wait until the deferred has a result
+        #
+        # Checking `d.called` by itself is not sufficient by itself as this is possible:
+        #
+        # If you have a first `Deferred` `D1`, you can add a callback which returns
+        # another `Deferred` `D2`, and `D2` must then complete before any further
+        # callbacks on `D1` will execute (and later callbacks on `D1` get the *result*
+        # of `D2` rather than `D2` itself).
+        #
+        # So, `D1` might have `called=True` (as in, it has started running its
+        # callbacks), but any new callbacks added to `D1` won't get run until `D2`
+        # completes. Fortunately, we can detect this by checking `d.paused`.
+        loop_count = 0
+        while not d.called or d.paused:
+            # 100 loops is arbitrary but based on previous code which used to "pump" and
+            # advance the reactor 100 times. 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).
+            if loop_count > 100:
+                raise defer.TimeoutError("Timed out waiting for deferred to finish")
+
+            # 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 swtich 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
+            # posssible.
+            if loop_count < 10:
+                time.sleep(0)
+            else:
+                time.sleep(0.001)
+
+            # Advance the Twisted reactor and run any scheduled callbacks
+            #
+            # In terms of other threads, they may have scheduled something on the
+            # reactor to run (like `reactor.callFromThread(...)`)
+            #
+            # Ideally, we'd advance by `0` but the `Cooperator` used in our HTTP clients
+            # use `CLOCK_SCHEDULE_EPSILON` and we want to make usage in downstream tests
+            # as simple as possible. A common use case this helps with is anything that
+            # needs to make a HTTP request (like a replication requests)
+            self.reactor.advance(CLOCK_SCHEDULE_EPSILON.as_secs())
+
+            loop_count += 1
+
+    def get_success(
+        self,
+        d: Awaitable[TV],
+    ) -> TV:
+        """
+        Get the success result of an awaitable.
+
+        Does not advance time in the Twisted reactor clock but will loop 100 times
+        waiting for a result. 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).
+
+        If you need to advance the Twisted reactor by an actual time increment, you can
+        use the following pattern:
+        ```python
+        # We use `ensureDeferred(...)` as a `Deferred` can run in the background on its own (unlike a Python coroutine)
+        task_d = ensureDeferred(my_async_task())
+        # Please explain why/what scheduled call you're trying to trigger
+        self.reactor.advance(Duration(seconds=1).as_secs())
+        result = self.get_success(sync_d)
+        ```
+
+        Args:
+            d: awaitable
+
+        Raises:
+            defer.TimeoutError: If the timeout expires before the awaitable completes.
+            SynchronousTestCase.failureException: If the awaitable has a failure result or has no result
+                (although you would probably run into `defer.TimeoutError` in that case).
+        """
         deferred: Deferred[TV] = ensureDeferred(d)  # type: ignore[arg-type]
-        self.pump(by=by)
+        self._wait_for_deferred(deferred)
+
         return self.successResultOf(deferred)
 
     def get_failure(
-        self, d: Awaitable[Any], exc: type[_ExcType], by: float = 0.0
+        self,
+        d: Awaitable[Any],
+        exc: type[_ExcType],
     ) -> _TypedFailure[_ExcType]:
         """
-        Run a Deferred and get a Failure from it. The failure must be of the type `exc`.
+        Get the failure result of an awaitable. The failure must be of the type `exc`.
+
+        Does not advance time in the Twisted reactor clock but will loop 100 times
+        waiting for a result. 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).
+
+        If you need to advance the Twisted reactor by an actual time increment, you can
+        use the following pattern:
+        ```python
+        # We use `ensureDeferred(...)` as a `Deferred` can run in the background on its own (unlike a Python coroutine)
+        task_d = ensureDeferred(my_async_task())
+        # Please explain why/what scheduled call you're trying to trigger
+        self.reactor.advance(Duration(seconds=1).as_secs())
+        result = self.get_success(sync_d)
+        ```
+
+        Args:
+            d: awaitable
+            exc: Exception type to expect
+
+        Raises:
+            defer.TimeoutError: If the timeout expires before the awaitable completes.
+            SynchronousTestCase.failureException: If the awaitable has a success result,
+                or has an unexpected failure result, or has no result (although you would
+                probably run into `defer.TimeoutError` in that case).
         """
         deferred: Deferred[Any] = ensureDeferred(d)  # type: ignore[arg-type]
-        self.pump(by)
+        self._wait_for_deferred(deferred)
+
         return self.failureResultOf(deferred, exc)
 
+    # FIXME: Remove as this has the exact same semantics as `get_success()`. In
+    # https://github.com/matrix-org/synapse/pull/8402#discussion_r495992506 where it was
+    # introduced, it was claimed that "get_success fails the test if the deferred fails
+    # rather than raising, which I find a bit unintuitive." but `get_success()` actually
+    # does raise "@raise SynchronousTestCase.failureException : If the
+    # L{Deferred<twisted.internet.defer.Deferred>} has no result or has a failure
+    # result." at-least in today's world.
+    #
+    # As another alternative, we could also just update `get_success(...)` to have this
+    # behavior as the default, see
+    # https://github.com/element-hq/synapse/pull/19871#discussion_r3483616710
     def get_success_or_raise(self, d: Awaitable[TV], by: float = 0.0) -> TV:
         """Drive deferred to completion and return result or raise exception
         on failure.

tests/util/test_task_scheduler.py

@@ -260,7 +260,7 @@ class TestTaskScheduler(HomeserverTestCase):
             await self.task_scheduler.update_task(
                 task.id, result={"counter": current_counter}
             )
-            await self.hs.get_clock().sleep(Duration(microseconds=1))
+            await self.hs.get_clock().sleep(Duration(seconds=1))
 
         return TaskStatus.COMPLETE, None, None  # type: ignore[unreachable]