From 7da21a715d956e2906f79aae24df703ebafab71e Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 15:20:38 -0500 Subject: [PATCH] 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 | 1 + synapse/http/client.py | 10 +- synapse/util/clock.py | 13 ++ tests/app/test_homeserver_shutdown.py | 16 ++- tests/crypto/test_keyring.py | 6 +- tests/handlers/test_federation.py | 1 - tests/handlers/test_oauth_delegation.py | 77 ++-------- tests/handlers/test_oidc.py | 6 +- tests/handlers/test_presence.py | 21 +-- tests/handlers/test_profile.py | 8 +- tests/handlers/test_room_member.py | 4 - tests/handlers/test_send_email.py | 4 +- tests/handlers/test_typing.py | 16 +++ tests/handlers/test_user_directory.py | 11 +- tests/media/test_media_storage.py | 7 +- tests/replication/tcp/test_handler.py | 12 ++ tests/replication/test_federation_ack.py | 8 ++ tests/rest/admin/test_user.py | 14 +- tests/server.py | 4 +- tests/storage/test_background_update.py | 54 ++++--- tests/storage/test_event_chain.py | 4 +- tests/synapse_rust/test_http_client.py | 45 ++---- tests/unittest.py | 173 ++++++++++++++++++++--- tests/util/test_task_scheduler.py | 2 +- 24 files changed, 325 insertions(+), 192 deletions(-) create mode 100644 changelog.d/19871.misc diff --git a/changelog.d/19871.misc b/changelog.d/19871.misc new file mode 100644 index 0000000000..be10ee0540 --- /dev/null +++ b/changelog.d/19871.misc @@ -0,0 +1 @@ +Update `HomeserverTestCase.get_success(...)` and friends to drive async Rust (Tokio runtime/thread pool). diff --git a/synapse/http/client.py b/synapse/http/client.py index 05c5f13a87..78f03ae58a 100644 --- a/synapse/http/client.py +++ b/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, ) diff --git a/synapse/util/clock.py b/synapse/util/clock.py index 7232a1331c..8c05675732 100644 --- a/synapse/util/clock.py +++ b/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 diff --git a/tests/app/test_homeserver_shutdown.py b/tests/app/test_homeserver_shutdown.py index 0f5d1c7338..20d314cb68 100644 --- a/tests/app/test_homeserver_shutdown.py +++ b/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 diff --git a/tests/crypto/test_keyring.py b/tests/crypto/test_keyring.py index 6bc935f272..01561b0d41 100644 --- a/tests/crypto/test_keyring.py +++ b/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)) diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index 794c0a3185..0c7edbaa2d 100644 --- a/tests/handlers/test_federation.py +++ b/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( diff --git a/tests/handlers/test_oauth_delegation.py b/tests/handlers/test_oauth_delegation.py index 6af2c068a4..995a1134b2 100644 --- a/tests/handlers/test_oauth_delegation.py +++ b/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 diff --git a/tests/handlers/test_oidc.py b/tests/handlers/test_oidc.py index 62b84c77a4..b81c2954dc 100644 --- a/tests/handlers/test_oidc.py +++ b/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"]) diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 44f1e6432d..5d02c70161 100644 --- a/tests/handlers/test_presence.py +++ b/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. diff --git a/tests/handlers/test_profile.py b/tests/handlers/test_profile.py index 5152e8fc53..561b45827f 100644 --- a/tests/handlers/test_profile.py +++ b/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 diff --git a/tests/handlers/test_room_member.py b/tests/handlers/test_room_member.py index d5b95e4ef6..0a7475856a 100644 --- a/tests/handlers/test_room_member.py +++ b/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, ) diff --git a/tests/handlers/test_send_email.py b/tests/handlers/test_send_email.py index eea88cd136..acb88343f2 100644 --- a/tests/handlers/test_send_email.py +++ b/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) diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py index 623eef0ecb..0bbe084547 100644 --- a/tests/handlers/test_typing.py +++ b/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", diff --git a/tests/handlers/test_user_directory.py b/tests/handlers/test_user_directory.py index f50fa1f4a0..dc6738ca28 100644 --- a/tests/handlers/test_user_directory.py +++ b/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() ) diff --git a/tests/media/test_media_storage.py b/tests/media/test_media_storage.py index f25b507aac..855a623ec0 100644 --- a/tests/media/test_media_storage.py +++ b/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) diff --git a/tests/replication/tcp/test_handler.py b/tests/replication/tcp/test_handler.py index a8eb7fc523..d35191e654 100644 --- a/tests/replication/tcp/test_handler.py +++ b/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) diff --git a/tests/replication/test_federation_ack.py b/tests/replication/test_federation_ack.py index e6b9ea5383..c8de7b1fad 100644 --- a/tests/replication/test_federation_ack.py +++ b/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] diff --git a/tests/rest/admin/test_user.py b/tests/rest/admin/test_user.py index add00453b6..bce199c564 100644 --- a/tests/rest/admin/test_user.py +++ b/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( diff --git a/tests/server.py b/tests/server.py index ce5eaad63d..eea11b301f 100644 --- a/tests/server.py +++ b/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: diff --git a/tests/storage/test_background_update.py b/tests/storage/test_background_update.py index e3f79d7670..139906e97c 100644 --- a/tests/storage/test_background_update.py +++ b/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 diff --git a/tests/storage/test_event_chain.py b/tests/storage/test_event_chain.py index 175a5ffc78..d09437c080 100644 --- a/tests/storage/test_event_chain.py +++ b/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 diff --git a/tests/synapse_rust/test_http_client.py b/tests/synapse_rust/test_http_client.py index 56fab3a0e1..845fe2b503 100644 --- a/tests/synapse_rust/test_http_client.py +++ b/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. diff --git a/tests/unittest.py b/tests/unittest.py index 93131521d0..5f7d0b3abf 100644 --- a/tests/unittest.py +++ b/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} 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. diff --git a/tests/util/test_task_scheduler.py b/tests/util/test_task_scheduler.py index 94c1d778e6..cab9695d33 100644 --- a/tests/util/test_task_scheduler.py +++ b/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]