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).

tests/app/test_homeserver_shutdown.py

@@ -24,6 +24,8 @@
 from typing import Any
 from unittest.mock import patch
 
+from twisted.internet.defer import ensureDeferred
+
 from synapse.app.homeserver import SynapseHomeServer
 from synapse.logging.context import LoggingContext
 from synapse.storage.background_updates import UpdaterStatus
@@ -76,6 +78,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 +115,10 @@ 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)
+        background_update_d = ensureDeferred(
+            store.db_pool.updates.do_next_background_update(False)
+        )
+        self.get_success(background_update_d)
 
         hs_ref = weakref.ref(self.hs)
 
@@ -127,6 +139,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/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_presence.py

@@ -29,6 +29,7 @@
     get_verify_key,
 )
 
+from twisted.internet.defer import ensureDeferred
 from twisted.internet.testing import MemoryReactor
 
 from synapse.api.constants import EventTypes, Membership, PresenceState
@@ -58,6 +59,7 @@
 from synapse.storage.keys import FetchKeyResult
 from synapse.types import JsonDict, UserID, get_domain_from_id
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests import unittest
 from tests.replication._base import BaseMultiWorkerStreamTestCase
@@ -948,12 +950,17 @@ def test_external_process_timeout(self) -> None:
         )
         worker_presence_handler = worker_to_sync_against.get_presence_handler()
 
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 self.user_id, self.device_id, True, PresenceState.ONLINE
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # Check that if we wait a while without telling the handler the user has
         # stopped syncing that their presence state doesn't get timed out.
@@ -1264,30 +1271,40 @@ def test_set_presence_from_syncing_multi_device(
             worker_presence_handler = worker_to_sync_against.get_presence_handler()
 
         # 1. Sync with the first device.
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-1",
                 affect_presence=dev_1_state != PresenceState.OFFLINE,
                 presence_state=dev_1_state,
-            ),
-            by=0.01,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # 2. Wait half the idle timer.
         self.reactor.advance(IDLE_TIMER / 1000 / 2)
         self.reactor.pump([0.1])
 
         # 3. Sync with the second device.
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-2",
                 affect_presence=dev_2_state != PresenceState.OFFLINE,
                 presence_state=dev_2_state,
-            ),
-            by=0.01,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # 4. Assert the expected presence state.
         state = self.get_success(
@@ -1305,15 +1322,21 @@ def test_set_presence_from_syncing_multi_device(
         #
         # This is due to EXTERNAL_PROCESS_EXPIRY being equivalent to IDLE_TIMER.
         if test_with_workers:
-            with self.get_success(
+            sync_d = ensureDeferred(
                 worker_presence_handler.user_syncing(
                     f"@other-user:{self.hs.config.server.server_name}",
                     "dev-3",
                     affect_presence=True,
                     presence_state=PresenceState.ONLINE,
-                ),
-                by=0.01,
-            ):
+                )
+            )
+            # `user_syncing` proxies the presence write to the main process over an HTTP
+            # replication request. The request body is streamed by a `Cooperator` that uses
+            # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+            # we need to actually advance the clock for it to fire.
+            self.reactor.advance(Duration(microseconds=1).as_secs())
+
+            with self.get_success(sync_d):
                 pass
 
         # 5. Advance such that the first device should be discarded (the idle timer),
@@ -1501,26 +1524,36 @@ def test_set_presence_from_non_syncing_multi_device(
             worker_presence_handler = worker_to_sync_against.get_presence_handler()
 
         # 1. Sync with the first device.
-        sync_1 = self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-1",
                 affect_presence=dev_1_state != PresenceState.OFFLINE,
                 presence_state=dev_1_state,
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        sync_1 = self.get_success(sync_d)
 
         # 2. Sync with the second device.
-        sync_2 = self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-2",
                 affect_presence=dev_2_state != PresenceState.OFFLINE,
                 presence_state=dev_2_state,
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        sync_2 = self.get_success(sync_d)
 
         # 3. Assert the expected presence state.
         state = self.get_success(
@@ -1622,12 +1655,17 @@ def test_set_presence_from_syncing_keeps_busy(
         # Perform a sync with a presence state other than busy. This should NOT change
         # our presence status; we only change from busy if we explicitly set it via
         # /presence/*.
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_to_sync_against.get_presence_handler().user_syncing(
                 self.user_id, self.device_id, True, PresenceState.ONLINE
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # Check against the main process that the user's presence did not change.
         state = self.get_success(self.presence_handler.get_state(self.user_id_obj))

tests/handlers/test_profile.py

@@ -200,7 +200,7 @@ async def slow_update_membership(*args: Any, **kwargs: Any) -> tuple[str, int]:
             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 @@ async def potentially_slow_update_membership(
 
             # 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(
@@ -300,7 +300,7 @@ async def potentially_slow_update_membership(
             )
 
             # Let's be sure we are over the delay introduced by slow_update_membership
-            self.get_success(self.clock.sleep(Duration(milliseconds=20)), by=1)
+            self.reactor.advance(Duration(milliseconds=20).as_secs())
 
             # Updates should have been resumed from room 2 after the restart
             # so room 1 should not have been updated this time

tests/handlers/test_room_member.py

@@ -71,7 +71,6 @@ def test_local_user_local_joins_contribute_to_limit_and_are_limited(self) -> Non
                 action=Membership.JOIN,
             ),
             LimitExceededError,
-            by=0.5,
         )
 
     @override_config({"rc_joins_per_room": {"per_second": 0.1, "burst_count": 2}})
@@ -213,7 +212,6 @@ def test_remote_joins_contribute_to_rate_limit(self) -> None:
                     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 @@ def test_local_users_joining_on_another_worker_contribute_to_rate_limit(
                 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 @@ def test_local_users_joining_on_another_worker_contribute_to_rate_limit(
                 action=Membership.JOIN,
             ),
             LimitExceededError,
-            by=0.5,
         )
 
 

tests/handlers/test_send_email.py

@@ -145,8 +145,12 @@ def test_send_email(self) -> None:
             )
         )
 
+        # This matches the two `callLater` delays in `FakeTransport.registerProducer`
+        self.reactor.advance(0)
+        self.reactor.advance(0.1)
+
         # 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)
@@ -212,8 +216,12 @@ def test_send_email_force_tls(self) -> None:
             )
         )
 
+        # This matches the two `callLater` delays in `FakeTransport.registerProducer`
+        self.reactor.advance(0)
+        self.reactor.advance(0.1)
+
         # 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)