diff --git a/tests/federation/test_federation_sender.py b/tests/federation/test_federation_sender.py
index 6a8887fe74..cd906bbbc7 100644
--- a/tests/federation/test_federation_sender.py
+++ b/tests/federation/test_federation_sender.py
@@ -34,6 +34,7 @@ from synapse.handlers.device import DeviceHandler
from synapse.rest import admin
from synapse.rest.client import login
from synapse.server import HomeServer
+from synapse.storage.databases.main.events_worker import EventMetadata
from synapse.types import JsonDict, ReadReceipt
from synapse.util import Clock
@@ -55,12 +56,15 @@ class FederationSenderReceiptsTestCases(HomeserverTestCase):
federation_transport_client=self.federation_transport_client,
)
- hs.get_storage_controllers().state.get_current_hosts_in_room = AsyncMock( # type: ignore[method-assign]
+ self.main_store = hs.get_datastores().main
+ self.state_controller = hs.get_storage_controllers().state
+
+ self.state_controller.get_current_hosts_in_room = AsyncMock( # type: ignore[method-assign]
return_value={"test", "host2"}
)
- hs.get_storage_controllers().state.get_current_hosts_in_room_or_partial_state_approximation = ( # type: ignore[method-assign]
- hs.get_storage_controllers().state.get_current_hosts_in_room
+ self.state_controller.get_current_hosts_in_room_or_partial_state_approximation = ( # type: ignore[method-assign]
+ self.state_controller.get_current_hosts_in_room
)
return hs
@@ -185,12 +189,15 @@ class FederationSenderReceiptsTestCases(HomeserverTestCase):
],
)
- def test_send_receipts_with_backoff(self) -> None:
- """Send two receipts in quick succession; the second should be flushed, but
- only after 20ms"""
+ def test_send_receipts_with_backoff_small_room(self) -> None:
+ """Read receipt in small rooms should not be delayed"""
mock_send_transaction = self.federation_transport_client.send_transaction
mock_send_transaction.return_value = {}
+ self.state_controller.get_current_hosts_in_room_or_partial_state_approximation = AsyncMock( # type: ignore[method-assign]
+ return_value={"test", "host2"}
+ )
+
sender = self.hs.get_federation_sender()
receipt = ReadReceipt(
"room_id",
@@ -206,47 +213,104 @@ class FederationSenderReceiptsTestCases(HomeserverTestCase):
# expect a call to send_transaction
mock_send_transaction.assert_called_once()
- json_cb = mock_send_transaction.call_args[0][1]
- data = json_cb()
- self.assertEqual(
- data["edus"],
- [
- {
- "edu_type": EduTypes.RECEIPT,
- "content": {
- "room_id": {
- "m.read": {
- "user_id": {
- "event_ids": ["event_id"],
- "data": {"ts": 1234},
- }
- }
- }
- },
- }
- ],
+ self._assert_edu_in_call(mock_send_transaction.call_args[0][1])
+
+ def test_send_receipts_with_backoff_recent_event(self) -> None:
+ """Read receipt for a recent message should not be delayed"""
+ mock_send_transaction = self.federation_transport_client.send_transaction
+ mock_send_transaction.return_value = {}
+
+ # Pretend this is a big room
+ self.state_controller.get_current_hosts_in_room_or_partial_state_approximation = AsyncMock( # type: ignore[method-assign]
+ return_value={"test"} | {f"host{i}" for i in range(20)}
)
+
+ self.main_store.get_metadata_for_event = AsyncMock(
+ return_value=EventMetadata(
+ received_ts=self.clock.time_msec(),
+ sender="@test:test",
+ )
+ )
+
+ sender = self.hs.get_federation_sender()
+ receipt = ReadReceipt(
+ "room_id",
+ "m.read",
+ "user_id",
+ ["event_id"],
+ thread_id=None,
+ data={"ts": 1234},
+ )
+ self.get_success(sender.send_read_receipt(receipt))
+
+ self.pump()
+
+ # expect a call to send_transaction for each host
+ self.assertEqual(mock_send_transaction.call_count, 20)
+ self._assert_edu_in_call(mock_send_transaction.call_args.args[1])
+
mock_send_transaction.reset_mock()
- # send the second RR
+ def test_send_receipts_with_backoff_sender(self) -> None:
+ """Read receipt for a message should not be delayed to the sender, but
+ is delayed to everyone else"""
+ mock_send_transaction = self.federation_transport_client.send_transaction
+ mock_send_transaction.return_value = {}
+
+ # Pretend this is a big room
+ self.state_controller.get_current_hosts_in_room_or_partial_state_approximation = AsyncMock( # type: ignore[method-assign]
+ return_value={"test"} | {f"host{i}" for i in range(20)}
+ )
+
+ self.main_store.get_metadata_for_event = AsyncMock(
+ return_value=EventMetadata(
+ received_ts=self.clock.time_msec() - 5 * 60_000,
+ sender="@test:host1",
+ )
+ )
+
+ sender = self.hs.get_federation_sender()
receipt = ReadReceipt(
"room_id",
"m.read",
"user_id",
- ["other_id"],
+ ["event_id"],
thread_id=None,
data={"ts": 1234},
)
- self.successResultOf(defer.ensureDeferred(sender.send_read_receipt(receipt)))
+ self.get_success(sender.send_read_receipt(receipt))
+
self.pump()
- mock_send_transaction.assert_not_called()
- self.reactor.advance(19)
- mock_send_transaction.assert_not_called()
+ # First, expect a call to send_transaction for the sending host
+ mock_send_transaction.assert_called()
- self.reactor.advance(10)
- mock_send_transaction.assert_called_once()
- json_cb = mock_send_transaction.call_args[0][1]
+ transaction = mock_send_transaction.call_args_list[0].args[0]
+ self.assertEqual(transaction.destination, "host1")
+ self._assert_edu_in_call(mock_send_transaction.call_args_list[0].args[1])
+
+ # We also expect a call to one of the other hosts, as the first
+ # destination to wake up.
+ self.assertEqual(mock_send_transaction.call_count, 2)
+ self._assert_edu_in_call(mock_send_transaction.call_args.args[1])
+
+ mock_send_transaction.reset_mock()
+
+ # We now expect to see 18 more transactions to the remaining hosts
+ # periodically.
+ for _ in range(18):
+ self.reactor.advance(
+ 1.0
+ / self.hs.config.ratelimiting.federation_rr_transactions_per_room_per_second
+ )
+
+ mock_send_transaction.assert_called_once()
+ self._assert_edu_in_call(mock_send_transaction.call_args.args[1])
+ mock_send_transaction.reset_mock()
+
+ def _assert_edu_in_call(self, json_cb: Callable[[], JsonDict]) -> None:
+ """Assert that the given `json_cb` from a `send_transaction` has a
+ receipt in it."""
data = json_cb()
self.assertEqual(
data["edus"],
@@ -257,7 +321,7 @@ class FederationSenderReceiptsTestCases(HomeserverTestCase):
"room_id": {
"m.read": {
"user_id": {
- "event_ids": ["other_id"],
+ "event_ids": ["event_id"],
"data": {"ts": 1234},
}
}
|
