Update FakeChannel.await_result(...) to drive async Rust (Tokio runtime/thread pool)

changelog.d/19879.misc

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

synapse/api/auth_blocking.py

@@ -117,7 +117,7 @@ async def check_auth_blocking(
             # If the user is already part of the MAU cohort or a trial user
             if user_id:
                 timestamp = await self.store.user_last_seen_monthly_active(user_id)
-                if timestamp:
+                if timestamp is not None:
                     return
 
                 is_trial = await self.store.is_trial_user(user_id)

tests/federation/test_federation_sender.py

@@ -37,6 +37,7 @@
 from synapse.storage.databases.main.events_worker import EventMetadata
 from synapse.types import JsonDict, ReadReceipt
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests.unittest import HomeserverTestCase
 
@@ -517,6 +518,24 @@ async def record_transaction(
         self.edus.extend(data["edus"])
         return {}
 
+    def wait_for_device_list_updates_to_be_sent(self) -> None:
+        """
+        Wait for the device list update EDU's to get pushed out over federation
+
+        For example, each login does a fire-and-forget (`LoginRestServlet` ->
+        `register_device` -> `notify_device_update` -> `handle_new_device_update` ->
+        `send_device_messages(hosts, immediate=False)`) which adds to the
+        `_DestinationWakeupQueue` which has a background process that sends depending on
+        how `federation_rr_transactions_per_room_per_second` is configured.
+
+        The default `federation_rr_transactions_per_room_per_second` is `50` (1s/50 ->
+        0.02s)
+        """
+        self.reactor.advance(
+            1.0
+            / self.hs.config.ratelimiting.federation_rr_transactions_per_room_per_second
+        )
+
     def test_send_device_updates(self) -> None:
         """Basic case: each device update should result in an EDU"""
         # create a device
@@ -527,9 +546,7 @@ def test_send_device_updates(self) -> None:
         self.assertEqual(len(self.edus), 1)
         stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D1", None)
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # a second call should produce no new device EDUs
         self.get_success(
@@ -568,7 +585,7 @@ def test_dont_send_device_updates_for_remote_users(self) -> None:
             )
         )
 
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # We shouldn't see an EDU for that update
         self.assertEqual(self.edus, [])
@@ -590,6 +607,8 @@ def test_upload_signatures(self) -> None:
         self.login(u1, "pass", device_id="D1")
         self.login(u1, "pass", device_id="D2")
 
+        self.wait_for_device_list_updates_to_be_sent()
+
         # expect two edus
         self.assertEqual(len(self.edus), 2)
         stream_id: int | None = None
@@ -600,9 +619,7 @@ def test_upload_signatures(self) -> None:
         device1_signing_key = self.generate_and_upload_device_signing_key(u1, "D1")
         device2_signing_key = self.generate_and_upload_device_signing_key(u1, "D2")
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect two more edus
         self.assertEqual(len(self.edus), 2)
@@ -637,9 +654,7 @@ def test_upload_signatures(self) -> None:
             e2e_handler.upload_signing_keys_for_user(u1, cross_signing_keys)
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect signing key update edu
         self.assertEqual(len(self.edus), 2)
@@ -662,9 +677,7 @@ def test_upload_signatures(self) -> None:
         )
         self.assertEqual(ret["failures"], {})
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect two edus, in one or two transactions. We don't know what order the
         # devices will be updated.
@@ -689,9 +702,7 @@ def test_delete_devices(self) -> None:
         self.login("user", "pass", device_id="D2")
         self.login("user", "pass", device_id="D3")
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect three edus
         self.assertEqual(len(self.edus), 3)
@@ -702,9 +713,7 @@ def test_delete_devices(self) -> None:
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # expect three edus, in an unknown order
         self.assertEqual(len(self.edus), 3)
@@ -730,15 +739,18 @@ def test_unreachable_server(self) -> None:
         # create devices
         u1 = self.register_user("user", "pass")
         self.login("user", "pass", device_id="D1")
+        # Wait some time in between each device list update as we want each of them to
+        # be attempted to be sent in their own transaction
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D2")
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D3")
+        self.reactor.advance(Duration(seconds=1).as_secs())
 
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         self.assertGreaterEqual(mock_send_txn.call_count, 4)
 
@@ -748,9 +760,7 @@ def test_unreachable_server(self) -> None:
             self.hs.get_federation_sender().send_device_messages(["host2"])
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # for each device, there should be a single update
         self.assertEqual(len(self.edus), 3)
@@ -777,16 +787,20 @@ def test_prune_outbound_device_pokes1(self) -> None:
         # create devices
         u1 = self.register_user("user", "pass")
         self.login("user", "pass", device_id="D1")
+        # Wait some time in between each device list update as we want each of them to
+        # be attempted to be sent in their own transaction
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D2")
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D3")
+        self.reactor.advance(Duration(seconds=1).as_secs())
 
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
+        # Ensure that we tried sending the device list update EDU's out
         self.assertGreaterEqual(mock_send_txn.call_count, 4)
 
         # run the prune job
@@ -801,9 +815,7 @@ def test_prune_outbound_device_pokes1(self) -> None:
             self.hs.get_federation_sender().send_device_messages(["host2"])
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # there should be a single update for this user.
         self.assertEqual(len(self.edus), 1)
@@ -835,15 +847,17 @@ def test_prune_outbound_device_pokes2(self) -> None:
         mock_send_txn.side_effect = AssertionError("fail")
 
         self.login("user", "pass", device_id="D2")
+        # Wait some time in between each device list update as we want each of them to
+        # be attempted to be sent in their own transaction
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self.login("user", "pass", device_id="D3")
-
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.reactor.advance(Duration(seconds=1).as_secs())
 
         # delete them again
         self.get_success(self.device_handler.delete_devices(u1, ["D1", "D2", "D3"]))
 
+        self.wait_for_device_list_updates_to_be_sent()
+
         self.assertGreaterEqual(mock_send_txn.call_count, 3)
 
         # run the prune job
@@ -858,9 +872,7 @@ def test_prune_outbound_device_pokes2(self) -> None:
             self.hs.get_federation_sender().send_device_messages(["host2"])
         )
 
-        # We queue up device list updates to be sent over federation, so we
-        # advance to clear the queue.
-        self.reactor.advance(1)
+        self.wait_for_device_list_updates_to_be_sent()
 
         # ... and we should get a single update for this user.
         self.assertEqual(len(self.edus), 1)

tests/rest/client/sliding_sync/test_sliding_sync.py

@@ -564,9 +564,16 @@ def test_wait_for_sync_token(self) -> None:
         )
         # Block for 10 seconds to make `notifier.wait_for_stream_token(from_token)`
         # timeout
+        #
+        # First, block for *almost* 10 seconds to make sure we are
+        # `notifier.wait_for_stream_token(from_token)`
         with self.assertRaises(TimedOutException):
             channel.await_result(timeout_ms=9900)
-        channel.await_result(timeout_ms=200)
+        # Then wait for the rest of the 10 second timeout, 9900 + 500 > 10000
+        #
+        # `notifier.wait_for_stream_token(from_token)` only checks every 500ms so we
+        # need to match that in order to make sure we hit the wake-up for sure.
+        channel.await_result(timeout_ms=500)
         self.assertEqual(channel.code, 200, channel.json_body)
 
         # We expect the next `pos` in the result to be the same as what we requested

tests/rest/client/test_delayed_events.py

@@ -26,6 +26,7 @@
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests import unittest
 from tests.server import FakeChannel
@@ -90,6 +91,10 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
             room=self.room_id, user=self.user2_user_id, tok=self.user2_access_token
         )
 
+        # Advance enough time where any requests we made during `prepare(...)` doesn't
+        # affect the rate-limits in the test itself
+        self.reactor.advance(Duration(days=1).as_secs())
+
     def test_delayed_events_empty_on_startup(self) -> None:
         self.assertListEqual([], self._get_delayed_events())
 

tests/rest/client/test_msc4388_rendezvous.py

@@ -175,13 +175,11 @@ def test_rendezvous_public(self) -> None:
         self.assertEqual(channel.code, 200)
         rendezvous_id = channel.json_body["id"]
         sequence_token = channel.json_body["sequence_token"]
-        expires_in_ms = channel.json_body["expires_in_ms"]
-        self.assertGreater(expires_in_ms, 0)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         session_endpoint = rz_endpoint + f"/{rendezvous_id}"
 
         # We can get the data back
-        # Advances clock by 100ms
         channel = self.make_request(
             "GET",
             session_endpoint,
@@ -191,10 +189,9 @@ def test_rendezvous_public(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=bar")
         self.assertEqual(channel.json_body["sequence_token"], sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 100)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can update the data
-        # Advances clock by 100ms
         channel = self.make_request(
             "PUT",
             session_endpoint,
@@ -207,7 +204,6 @@ def test_rendezvous_public(self) -> None:
         new_sequence_token = channel.json_body["sequence_token"]
 
         # If we try to update it again with the old etag, it should fail
-        # Advances clock by 100ms
         channel = self.make_request(
             "PUT",
             session_endpoint,
@@ -221,7 +217,6 @@ def test_rendezvous_public(self) -> None:
         )
 
         # We should get the updated data
-        # Advances clock by 100ms
         channel = self.make_request(
             "GET",
             session_endpoint,
@@ -231,7 +226,7 @@ def test_rendezvous_public(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=baz")
         self.assertEqual(channel.json_body["sequence_token"], new_sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 400)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can delete the data
         channel = self.make_request(
@@ -376,8 +371,7 @@ def test_rendezvous_requires_authentication(self) -> None:
         self.assertEqual(channel.code, 200)
         rendezvous_id = channel.json_body["id"]
         sequence_token = channel.json_body["sequence_token"]
-        expires_in_ms = channel.json_body["expires_in_ms"]
-        self.assertEqual(expires_in_ms, 120000)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         session_endpoint = rz_endpoint + f"/{rendezvous_id}"
 
@@ -391,7 +385,7 @@ def test_rendezvous_requires_authentication(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=bar")
         self.assertEqual(channel.json_body["sequence_token"], sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 100)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can update the data without authentication
         channel = self.make_request(
@@ -414,7 +408,7 @@ def test_rendezvous_requires_authentication(self) -> None:
         self.assertEqual(channel.code, 200)
         self.assertEqual(channel.json_body["data"], "foo=baz")
         self.assertEqual(channel.json_body["sequence_token"], new_sequence_token)
-        self.assertEqual(channel.json_body["expires_in_ms"], expires_in_ms - 300)
+        self.assertGreater(channel.json_body["expires_in_ms"], 0)
 
         # We can delete the data without authentication
         channel = self.make_request(

tests/rest/client/test_sticky_events.py

@@ -58,7 +58,7 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         # Create a room
         self.room_id = self.helper.create_room_as(self.user_id, tok=self.token)
 
-    def _assert_event_sticky_for(self, event_id: str, sticky_ttl: int) -> None:
+    def _assert_event_sticky_for(self, event_id: str) -> None:
         channel = self.make_request(
             "GET",
             f"/rooms/{self.room_id}/event/{event_id}",
@@ -75,10 +75,10 @@ def _assert_event_sticky_for(self, event_id: str, sticky_ttl: int) -> None:
             event["unsigned"],
             f"No {EventUnsignedContentFields.STICKY_TTL} field in {event_id}; event not sticky: {event}",
         )
-        self.assertEqual(
+        self.assertGreater(
             event["unsigned"][EventUnsignedContentFields.STICKY_TTL],
-            sticky_ttl,
-            f"{event_id} had an unexpected sticky TTL: {event}",
+            0,
+            f"{event_id} had an unexpected sticky TTL (expected some value greater than 0): {event}",
         )
 
     def _assert_event_not_sticky(self, event_id: str) -> None:
@@ -112,17 +112,15 @@ def test_sticky_event_via_event_endpoint(self) -> None:
 
         # If we request the event immediately, it will still have
         # 1 minute of stickiness
-        # The other 100 ms is advanced in FakeChannel.await_result.
-        self._assert_event_sticky_for(event_id, 59_900)
+        self._assert_event_sticky_for(event_id)
 
-        # But if we advance time by 59.799 seconds...
-        # we will get the event on its last millisecond of stickiness
-        # The other 100 ms is advanced in FakeChannel.await_result.
-        self.reactor.advance(59.799)
-        self._assert_event_sticky_for(event_id, 1)
+        # But if we advance time by 59 seconds...
+        # we will get the event on its last second of stickiness
+        self.reactor.advance(Duration(seconds=59).as_secs())
+        self._assert_event_sticky_for(event_id)
 
         # Advancing time any more, the event is no longer sticky
-        self.reactor.advance(0.001)
+        self.reactor.advance(Duration(seconds=1).as_secs())
         self._assert_event_not_sticky(event_id)