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]