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

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.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 @@
 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 @@ async def shutdown() -> None:
 
         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 @@ def test_clean_homeserver_shutdown_mid_background_updates(self) -> None:
         # 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 @@ async def shutdown() -> None:
 
         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 @@ async def get_json(destination: str, path: str, **kwargs: Any) -> JsonDict:
         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 @@ def test_get_keys_from_perspectives(self) -> None:
         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 @@ def test_get_perspectives_own_key(self) -> None:
         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 @@ def create_invite() -> EventBase:
                 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 @@ def test_simple_introspection(self) -> None:
             "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 @@ def test_unexpiring_token(self) -> None:
             "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 @@ def test_inexistent_device(self) -> None:
         }
 
         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 @@ def test_inexistent_user(self) -> None:
         }
 
         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 @@ def test_missing_scope(self) -> None:
         }
 
         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 @@ def test_invalid_response(self) -> None:
         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 @@ def test_device_id_in_body(self) -> None:
             "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 @@ def test_admin_scope(self) -> None:
             "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 @@ def test_cached_expired_introspection(self) -> None:
         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 @@ def test_exchange_code_jwt_key(self) -> None:
         # 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 @@ def test_exchange_code_jwt_key(self) -> None:
         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 @@ def test_external_process_timeout(self) -> None:
         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 @@ def test_set_presence_from_syncing_multi_device(
                 "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 @@ def test_set_presence_from_syncing_multi_device(
                 "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 @@ def test_set_presence_from_syncing_multi_device(
                     "dev-3",
                     affect_presence=True,
                     presence_state=PresenceState.ONLINE,
-                ),
-                by=0.01,
+                )
             ):
                 pass
 
@@ -1507,8 +1503,7 @@ def test_set_presence_from_non_syncing_multi_device(
                 "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 @@ def test_set_presence_from_non_syncing_multi_device(
                 "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 @@ def test_set_presence_from_syncing_keeps_busy(
         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.