From 61fa1b9d223671ef3b181359351d4570538bd1bb Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Fri, 9 May 2025 00:54:21 +0200 Subject: [PATCH 01/50] Limit to_device EDU size to 65536 --- changelog.d/18416.bugfix | 1 + synapse/api/constants.py | 1 + synapse/handlers/devicemessage.py | 98 ++++++++++++++++--- synapse/storage/databases/main/deviceinbox.py | 79 ++++++++------- tests/rest/client/test_sendtodevice.py | 96 +++++++++++++++++- 5 files changed, 225 insertions(+), 50 deletions(-) create mode 100644 changelog.d/18416.bugfix diff --git a/changelog.d/18416.bugfix b/changelog.d/18416.bugfix new file mode 100644 index 00000000000..71f181fed40 --- /dev/null +++ b/changelog.d/18416.bugfix @@ -0,0 +1 @@ +A long queue of to-device messages could prevent outgoing federation because of the size of the transaction, let's limit the to-device EDU size to a reasonable value. diff --git a/synapse/api/constants.py b/synapse/api/constants.py index c564a8635a0..6b74eac367a 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -28,6 +28,7 @@ # the max size of a (canonical-json-encoded) event MAX_PDU_SIZE = 65536 +MAX_EDU_SIZE = 65536 # Max/min size of ints in canonical JSON CANONICALJSON_MAX_INT = (2**53) - 1 diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index e56bdb40720..6f9e25f7751 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -20,11 +20,19 @@ # import logging +from copy import deepcopy from http import HTTPStatus -from typing import TYPE_CHECKING, Any, Dict, Optional +from typing import TYPE_CHECKING, Any, Dict, List, Optional -from synapse.api.constants import EduTypes, EventContentFields, ToDeviceEventTypes -from synapse.api.errors import Codes, SynapseError +from canonicaljson import encode_canonical_json + +from synapse.api.constants import ( + MAX_EDU_SIZE, + EduTypes, + EventContentFields, + ToDeviceEventTypes, +) +from synapse.api.errors import Codes, EventSizeError, SynapseError from synapse.api.ratelimiting import Ratelimiter from synapse.logging.context import run_in_background from synapse.logging.opentracing import ( @@ -293,18 +301,18 @@ async def send_device_message( remote_edu_contents = {} for destination, messages in remote_messages.items(): - # The EDU contains a "message_id" property which is used for - # idempotence. Make up a random one. - message_id = random_string(16) - log_kv({"destination": destination, "message_id": message_id}) - - remote_edu_contents[destination] = { - "messages": messages, - "sender": sender_user_id, - "type": message_type, - "message_id": message_id, - "org.matrix.opentracing_context": json_encoder.encode(context), - } + edu_contents = get_device_message_edu_contents( + sender_user_id, message_type, messages, context + ) + remote_edu_contents[destination] = edu_contents + log_kv( + { + "destination": destination, + "message_ids": [ + edu_content["message_id"] for edu_content in edu_contents + ], + } + ) # Add messages to the database. # Retrieve the stream id of the last-processed to-device message. @@ -409,3 +417,63 @@ async def get_events_for_dehydrated_device( "events": messages, "next_batch": f"d{stream_id}", } + + +def get_device_message_edu_contents( + sender_user_id: str, + message_type: str, + messages: Dict[str, Dict[str, JsonDict]], + context: Dict[str, Any], +) -> List[JsonDict]: + """ + This function takes a dictionary of messages and splits them into several EDUs if needed. + + It will raise an EventSizeError if a single message is too large to fit into an EDU. + """ + + base_edu_content = { + "messages": {}, + "sender": sender_user_id, + "type": message_type, + "message_id": random_string(16), + } + # This is the size of the full EDU without any messages and without the opentracing context + base_edu_size = len( + encode_canonical_json( + { + "edu_type": "m.direct_to_device", + "content": base_edu_content, + } + ) + ) + base_edu_content["org.matrix.opentracing_context"] = json_encoder.encode(context) + + edu_contents = [] + + current_edu_content: JsonDict = deepcopy(base_edu_content) + current_edu_size = base_edu_size + + for recipient, message in messages.items(): + # We remove 2 for the curly braces and add 2 for the colon and comma + # We may overshoot by 1 for single message EDUs because of the comma, but that's fine + message_entry_size = len(encode_canonical_json({recipient: message})) + + if current_edu_size + message_entry_size > MAX_EDU_SIZE: + if len(current_edu_content["messages"]) == 0: + raise EventSizeError("device message too large", unpersistable=True) + + edu_contents.append(current_edu_content) + + current_edu_content = deepcopy(base_edu_content) + current_edu_content["message_id"] = random_string(16) + + current_edu_size = base_edu_size + else: + current_edu_size += message_entry_size + + current_edu_content["messages"][recipient] = message + + if len(current_edu_content["messages"]) > 0: + edu_contents.append(current_edu_content) + + return edu_contents diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 0612b82b9b7..1e7177b38f3 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -718,15 +718,15 @@ def get_all_new_device_messages_txn( async def add_messages_to_device_inbox( self, local_messages_by_user_then_device: Dict[str, Dict[str, JsonDict]], - remote_messages_by_destination: Dict[str, JsonDict], + remote_edu_contents: Dict[str, List[JsonDict]], ) -> int: """Used to send messages from this server. Args: local_messages_by_user_then_device: Dictionary of recipient user_id to recipient device_id to message. - remote_messages_by_destination: - Dictionary of destination server_name to the EDU JSON to send. + remote_edu_contents: + Dictionary of destination server_name to the list of EDU contents to send. Returns: The new stream_id. @@ -760,42 +760,53 @@ def add_messages_txn( destination, stream_id, now_ms, - json_encoder.encode(edu), + json_encoder.encode(edu_content), self._instance_name, ) - for destination, edu in remote_messages_by_destination.items() + for destination, edu_contents in remote_edu_contents.items() + for edu_content in edu_contents ], ) - for destination, edu in remote_messages_by_destination.items(): - if issue9533_logger.isEnabledFor(logging.DEBUG): - issue9533_logger.debug( - "Queued outgoing to-device messages with " - "stream_id %i, EDU message_id %s, type %s for %s: %s", - stream_id, - edu["message_id"], - edu["type"], - destination, - [ - f"{user_id}/{device_id} (msgid " - f"{msg.get(EventContentFields.TO_DEVICE_MSGID)})" - for (user_id, messages_by_device) in edu["messages"].items() - for (device_id, msg) in messages_by_device.items() - ], - ) + for destination, edu_contents in remote_edu_contents.items(): + for edu_content in edu_contents: + if issue9533_logger.isEnabledFor(logging.DEBUG): + issue9533_logger.debug( + "Queued outgoing to-device messages with " + "stream_id %i, EDU message_id %s, type %s for %s: %s", + stream_id, + edu_content["message_id"], + edu_content["type"], + destination, + [ + f"{user_id}/{device_id} (msgid " + f"{msg.get(EventContentFields.TO_DEVICE_MSGID)})" + for (user_id, messages_by_device) in edu_content[ + "messages" + ].items() + for (device_id, msg) in messages_by_device.items() + ], + ) - for user_id, messages_by_device in edu["messages"].items(): - for device_id, msg in messages_by_device.items(): - with start_active_span("store_outgoing_to_device_message"): - set_tag(SynapseTags.TO_DEVICE_EDU_ID, edu["sender"]) - set_tag(SynapseTags.TO_DEVICE_EDU_ID, edu["message_id"]) - set_tag(SynapseTags.TO_DEVICE_TYPE, edu["type"]) - set_tag(SynapseTags.TO_DEVICE_RECIPIENT, user_id) - set_tag(SynapseTags.TO_DEVICE_RECIPIENT_DEVICE, device_id) - set_tag( - SynapseTags.TO_DEVICE_MSGID, - msg.get(EventContentFields.TO_DEVICE_MSGID), - ) + for user_id, messages_by_device in edu_content["messages"].items(): + for device_id, msg in messages_by_device.items(): + with start_active_span("store_outgoing_to_device_message"): + set_tag( + SynapseTags.TO_DEVICE_EDU_ID, edu_content["sender"] + ) + set_tag( + SynapseTags.TO_DEVICE_EDU_ID, + edu_content["message_id"], + ) + set_tag(SynapseTags.TO_DEVICE_TYPE, edu_content["type"]) + set_tag(SynapseTags.TO_DEVICE_RECIPIENT, user_id) + set_tag( + SynapseTags.TO_DEVICE_RECIPIENT_DEVICE, device_id + ) + set_tag( + SynapseTags.TO_DEVICE_MSGID, + msg.get(EventContentFields.TO_DEVICE_MSGID), + ) async with self._to_device_msg_id_gen.get_next() as stream_id: now_ms = self._clock.time_msec() @@ -804,7 +815,7 @@ def add_messages_txn( ) for user_id in local_messages_by_user_then_device.keys(): self._device_inbox_stream_cache.entity_has_changed(user_id, stream_id) - for destination in remote_messages_by_destination.keys(): + for destination in remote_edu_contents.keys(): self._device_federation_outbox_stream_cache.entity_has_changed( destination, stream_id ) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 5ef501c6d51..2f2c92a1919 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -18,12 +18,20 @@ # [This file includes modifications made by New Vector Limited] # # +from unittest.mock import AsyncMock, Mock + from parameterized import parameterized_class -from synapse.api.constants import EduTypes +from twisted.test.proto_helpers import MemoryReactor + +from synapse.api.constants import MAX_EDU_SIZE, EduTypes +from synapse.api.errors import Codes from synapse.rest import admin from synapse.rest.client import login, sendtodevice, sync +from synapse.server import HomeServer from synapse.types import JsonDict +from synapse.util import Clock +from synapse.util.stringutils import random_string from tests.unittest import HomeserverTestCase, override_config @@ -61,8 +69,18 @@ class SendToDeviceTestCase(HomeserverTestCase): def default_config(self) -> JsonDict: config = super().default_config() config["experimental_features"] = self.experimental_features + config["federation_sender_instances"] = None return config + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + self.federation_transport_client = Mock(spec=["send_transaction"]) + self.federation_transport_client.send_transaction = AsyncMock() + hs = self.setup_test_homeserver( + federation_transport_client=self.federation_transport_client, + ) + + return hs + def test_user_to_user(self) -> None: """A to-device message from one user to another should get delivered""" @@ -113,6 +131,82 @@ def test_user_to_user(self) -> None: self.assertEqual(channel.code, 200, channel.result) self.assertEqual(channel.json_body.get("to_device", {}).get("events", []), []) + def test_large_remote_todevice(self) -> None: + """A to-device message needs to fit in the EDU size limit""" + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + + # send the message + test_msg = {"foo": random_string(MAX_EDU_SIZE)} + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/12345", + content={"messages": {"@remote_user:secondserver": {"device": test_msg}}}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 413, channel.result) + self.assertEqual(Codes.TOO_LARGE, channel.json_body["errcode"]) + + def test_edu_splitting(self) -> None: + """Test that a bunch of to-device messages are split into multiple EDUs if they are too large""" + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} + + sender = self.hs.get_federation_sender() + + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} + + # 2 small messages that should fit in a single EDU + for i in range(2): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(100)} + } + + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/123456", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + + self.get_success(sender.send_device_messages([destination])) + + self.pump() + + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(len(data["edus"]), 1) + + mock_send_transaction.reset_mock() + + # 2 messages, each just big enough to fit in an EDU + for i in range(2): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + } + + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/1234567", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + + self.get_success(sender.send_device_messages([destination])) + + self.pump() + + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(len(data["edus"]), 2) + @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: """m.room_key_request has special-casing; test from local user""" From c80f24d7162f2d6e88f9e44bb952e1a7fc996978 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 01:43:00 +0200 Subject: [PATCH 02/50] Increment to_device stream for each EDU otherwise we loose some --- synapse/handlers/devicemessage.py | 3 +- synapse/storage/databases/main/deviceinbox.py | 31 +++++++++----- tests/rest/client/test_sendtodevice.py | 40 +++++++++++++++++++ 3 files changed, 62 insertions(+), 12 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 6f9e25f7751..8e5de9669ea 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -468,9 +468,8 @@ def get_device_message_edu_contents( current_edu_content["message_id"] = random_string(16) current_edu_size = base_edu_size - else: - current_edu_size += message_entry_size + current_edu_size += message_entry_size current_edu_content["messages"][recipient] = message if len(current_edu_content["messages"]) > 0: diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 1e7177b38f3..278adf97156 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -735,11 +735,14 @@ async def add_messages_to_device_inbox( assert self._can_write_to_device def add_messages_txn( - txn: LoggingTransaction, now_ms: int, stream_id: int + txn: LoggingTransaction, + now_ms: int, + stream_ids: List[int], + last_stream_id: int, ) -> None: # Add the local messages directly to the local inbox. self._add_messages_to_local_device_inbox_txn( - txn, stream_id, local_messages_by_user_then_device + txn, last_stream_id, local_messages_by_user_then_device ) # Add the remote messages to the federation outbox. @@ -758,23 +761,23 @@ def add_messages_txn( values=[ ( destination, - stream_id, + stream_ids[i], now_ms, json_encoder.encode(edu_content), self._instance_name, ) for destination, edu_contents in remote_edu_contents.items() - for edu_content in edu_contents + for i, edu_content in enumerate(edu_contents) ], ) for destination, edu_contents in remote_edu_contents.items(): - for edu_content in edu_contents: + for i, edu_content in enumerate(edu_contents): if issue9533_logger.isEnabledFor(logging.DEBUG): issue9533_logger.debug( "Queued outgoing to-device messages with " "stream_id %i, EDU message_id %s, type %s for %s: %s", - stream_id, + stream_ids[i], edu_content["message_id"], edu_content["type"], destination, @@ -808,16 +811,24 @@ def add_messages_txn( msg.get(EventContentFields.TO_DEVICE_MSGID), ) - async with self._to_device_msg_id_gen.get_next() as stream_id: + nb_edus = sum(len(edus) for edus in remote_edu_contents.values()) + async with self._to_device_msg_id_gen.get_next_mult(nb_edus) as stream_ids: + last_stream_id = stream_ids[len(stream_ids) - 1] now_ms = self._clock.time_msec() await self.db_pool.runInteraction( - "add_messages_to_device_inbox", add_messages_txn, now_ms, stream_id + "add_messages_to_device_inbox", + add_messages_txn, + now_ms, + stream_ids, + last_stream_id, ) for user_id in local_messages_by_user_then_device.keys(): - self._device_inbox_stream_cache.entity_has_changed(user_id, stream_id) + self._device_inbox_stream_cache.entity_has_changed( + user_id, last_stream_id + ) for destination in remote_edu_contents.keys(): self._device_federation_outbox_stream_cache.entity_has_changed( - destination, stream_id + destination, last_stream_id ) return self._to_device_msg_id_gen.get_current_token() diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 2f2c92a1919..ad4b9a15207 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -207,6 +207,46 @@ def test_edu_splitting(self) -> None: data = json_cb() self.assertEqual(len(data["edus"]), 2) + def test_transaction_splitting(self) -> None: + """Test that a bunch of to-device messages are split into multiple transactions if they are too many EDUs""" + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} + + sender = self.hs.get_federation_sender() + + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} + + for i in range(101): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + } + + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/12345678", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + + self.get_success(sender.send_device_messages([destination])) + + self.pump() + + self.assertEqual(mock_send_transaction.call_count, 2) + + # A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes + first_call = mock_send_transaction.call_args_list[0][0][1]() + self.assertEqual(len(first_call["edus"]), 90) + + second_call = mock_send_transaction.call_args_list[1][0][1]() + self.assertEqual(len(second_call["edus"]), 11) + @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: """m.room_key_request has special-casing; test from local user""" From eda00e1d54b43149a89d7ceaa76ca561febed9c9 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 02:31:26 +0200 Subject: [PATCH 03/50] Simplify --- synapse/handlers/devicemessage.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 8e5de9669ea..df51f9b78f7 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -454,14 +454,17 @@ def get_device_message_edu_contents( current_edu_size = base_edu_size for recipient, message in messages.items(): - # We remove 2 for the curly braces and add 2 for the colon and comma - # We may overshoot by 1 for single message EDUs because of the comma, but that's fine - message_entry_size = len(encode_canonical_json({recipient: message})) + # We remove 2 for the curly braces and add 1 for the colon + message_entry_size = len(encode_canonical_json({recipient: message})) - 2 + 1 - if current_edu_size + message_entry_size > MAX_EDU_SIZE: - if len(current_edu_content["messages"]) == 0: - raise EventSizeError("device message too large", unpersistable=True) + if base_edu_size + message_entry_size > MAX_EDU_SIZE: + raise EventSizeError("device message too large", unpersistable=True) + + first_message = len(current_edu_content["messages"]) == 0 + if not first_message: + message_entry_size += 1 # Add 1 for the comma + if current_edu_size + message_entry_size > MAX_EDU_SIZE: edu_contents.append(current_edu_content) current_edu_content = deepcopy(base_edu_content) From 6627bed2b5656e55a70ef6ba37e263008ed57c41 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 11:02:35 +0200 Subject: [PATCH 04/50] Add comment --- synapse/handlers/devicemessage.py | 3 +-- synapse/storage/databases/main/deviceinbox.py | 2 ++ 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index df51f9b78f7..4db3b568a67 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -460,8 +460,7 @@ def get_device_message_edu_contents( if base_edu_size + message_entry_size > MAX_EDU_SIZE: raise EventSizeError("device message too large", unpersistable=True) - first_message = len(current_edu_content["messages"]) == 0 - if not first_message: + if len(current_edu_content["messages"]) > 0: message_entry_size += 1 # Add 1 for the comma if current_edu_size + message_entry_size > MAX_EDU_SIZE: diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 278adf97156..e2bd548ef5d 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -811,6 +811,8 @@ def add_messages_txn( msg.get(EventContentFields.TO_DEVICE_MSGID), ) + # We allocate one stream id per EDU so we can track if they were + # successfully sent or not. nb_edus = sum(len(edus) for edus in remote_edu_contents.values()) async with self._to_device_msg_id_gen.get_next_mult(nb_edus) as stream_ids: last_stream_id = stream_ids[len(stream_ids) - 1] From c6bc6911da36e338413d0accd050a6f4a2d97278 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 11:53:21 +0200 Subject: [PATCH 05/50] Cosmetic --- synapse/handlers/devicemessage.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 4db3b568a67..777b65fcdd9 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -458,7 +458,10 @@ def get_device_message_edu_contents( message_entry_size = len(encode_canonical_json({recipient: message})) - 2 + 1 if base_edu_size + message_entry_size > MAX_EDU_SIZE: - raise EventSizeError("device message too large", unpersistable=True) + raise EventSizeError( + f"device message to {recipient} too large to fit in a single EDU", + unpersistable=True, + ) if len(current_edu_content["messages"]) > 0: message_entry_size += 1 # Add 1 for the comma @@ -471,8 +474,8 @@ def get_device_message_edu_contents( current_edu_size = base_edu_size - current_edu_size += message_entry_size current_edu_content["messages"][recipient] = message + current_edu_size += message_entry_size if len(current_edu_content["messages"]) > 0: edu_contents.append(current_edu_content) From 57ab5418f9d1960b6c92828b06096224ad6b28af Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 12:00:15 +0200 Subject: [PATCH 06/50] Cosmetic --- synapse/handlers/devicemessage.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 777b65fcdd9..8f5e1c715fa 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -431,33 +431,33 @@ def get_device_message_edu_contents( It will raise an EventSizeError if a single message is too large to fit into an EDU. """ - base_edu_content = { + BASE_EDU_CONTENT = { "messages": {}, "sender": sender_user_id, "type": message_type, "message_id": random_string(16), } # This is the size of the full EDU without any messages and without the opentracing context - base_edu_size = len( + BASE_EDU_SIZE = len( encode_canonical_json( { "edu_type": "m.direct_to_device", - "content": base_edu_content, + "content": BASE_EDU_CONTENT, } ) ) - base_edu_content["org.matrix.opentracing_context"] = json_encoder.encode(context) + BASE_EDU_CONTENT["org.matrix.opentracing_context"] = json_encoder.encode(context) edu_contents = [] - current_edu_content: JsonDict = deepcopy(base_edu_content) - current_edu_size = base_edu_size + current_edu_content: JsonDict = deepcopy(BASE_EDU_CONTENT) + current_edu_size = BASE_EDU_SIZE for recipient, message in messages.items(): # We remove 2 for the curly braces and add 1 for the colon message_entry_size = len(encode_canonical_json({recipient: message})) - 2 + 1 - if base_edu_size + message_entry_size > MAX_EDU_SIZE: + if current_edu_size + message_entry_size > MAX_EDU_SIZE: raise EventSizeError( f"device message to {recipient} too large to fit in a single EDU", unpersistable=True, @@ -469,10 +469,10 @@ def get_device_message_edu_contents( if current_edu_size + message_entry_size > MAX_EDU_SIZE: edu_contents.append(current_edu_content) - current_edu_content = deepcopy(base_edu_content) + current_edu_content = deepcopy(BASE_EDU_CONTENT) current_edu_content["message_id"] = random_string(16) - current_edu_size = base_edu_size + current_edu_size = BASE_EDU_SIZE current_edu_content["messages"][recipient] = message current_edu_size += message_entry_size From a0e6dc3bf84d6711ae05f8b5f1fdbc9695ddfcce Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 12:09:38 +0200 Subject: [PATCH 07/50] Add logs --- synapse/handlers/devicemessage.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 8f5e1c715fa..fb3f00e625f 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -468,6 +468,12 @@ def get_device_message_edu_contents( if current_edu_size + message_entry_size > MAX_EDU_SIZE: edu_contents.append(current_edu_content) + logger.debug( + "Splitting %d device messages from %s into a separate EDU, %d EDUs queued", + len(current_edu_content["messages"]), + sender_user_id, + len(edu_contents), + ) current_edu_content = deepcopy(BASE_EDU_CONTENT) current_edu_content["message_id"] = random_string(16) @@ -479,5 +485,11 @@ def get_device_message_edu_contents( if len(current_edu_content["messages"]) > 0: edu_contents.append(current_edu_content) + logger.debug( + "Queuing last %d device messages from %s, %d EDUs queued", + len(current_edu_content["messages"]), + sender_user_id, + len(edu_contents), + ) return edu_contents From 9ce14880fde8561215790feb57571b30fb65880f Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 12:16:32 +0200 Subject: [PATCH 08/50] Improve logs --- synapse/handlers/devicemessage.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index fb3f00e625f..40d1f07975c 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -469,9 +469,10 @@ def get_device_message_edu_contents( if current_edu_size + message_entry_size > MAX_EDU_SIZE: edu_contents.append(current_edu_content) logger.debug( - "Splitting %d device messages from %s into a separate EDU, %d EDUs queued", + "Splitting %d device messages from %s into a separate EDU msgid %s, %d EDUs queued", len(current_edu_content["messages"]), sender_user_id, + current_edu_content["message_id"], len(edu_contents), ) @@ -486,9 +487,10 @@ def get_device_message_edu_contents( if len(current_edu_content["messages"]) > 0: edu_contents.append(current_edu_content) logger.debug( - "Queuing last %d device messages from %s, %d EDUs queued", + "Queuing last %d device messages from %s into EDU msgid %s, %d EDUs queued", len(current_edu_content["messages"]), sender_user_id, + current_edu_content["message_id"], len(edu_contents), ) From 5e86c59812fe92e3d3edae5412d21788a3313a50 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 12:27:20 +0200 Subject: [PATCH 09/50] fix bug --- synapse/handlers/devicemessage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 40d1f07975c..8f6fb849f37 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -457,7 +457,7 @@ def get_device_message_edu_contents( # We remove 2 for the curly braces and add 1 for the colon message_entry_size = len(encode_canonical_json({recipient: message})) - 2 + 1 - if current_edu_size + message_entry_size > MAX_EDU_SIZE: + if BASE_EDU_SIZE + message_entry_size > MAX_EDU_SIZE: raise EventSizeError( f"device message to {recipient} too large to fit in a single EDU", unpersistable=True, From 35d98b66c70241409356c732b8d7b4fd7ebabe97 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 12 May 2025 21:18:23 +0200 Subject: [PATCH 10/50] Improve logs --- synapse/handlers/devicemessage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 8f6fb849f37..7df5ac8b128 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -469,7 +469,7 @@ def get_device_message_edu_contents( if current_edu_size + message_entry_size > MAX_EDU_SIZE: edu_contents.append(current_edu_content) logger.debug( - "Splitting %d device messages from %s into a separate EDU msgid %s, %d EDUs queued", + "Splitting %d device messages from %s into EDU msgid %s, %d EDUs queued", len(current_edu_content["messages"]), sender_user_id, current_edu_content["message_id"], From a15a585984a649e831f9123725cde37b2dfdc9d9 Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 15:53:26 +0200 Subject: [PATCH 11/50] Apply suggestion from @MadLittleMods Co-authored-by: Eric Eastwood --- tests/rest/client/test_sendtodevice.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index ad4b9a15207..266c828c142 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -185,7 +185,7 @@ def test_edu_splitting(self) -> None: mock_send_transaction.reset_mock() - # 2 messages, each just big enough to fit in an EDU + # 2 messages, each just big enough to fit into their own EDU for i in range(2): messages[f"@remote_user{i}:" + destination] = { "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} From e81e737a652f88baeccce94e07b3d136d108e256 Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 15:53:37 +0200 Subject: [PATCH 12/50] Apply suggestion from @MadLittleMods Co-authored-by: Eric Eastwood --- synapse/api/constants.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 6b74eac367a..3e63fcfff53 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -28,6 +28,15 @@ # the max size of a (canonical-json-encoded) event MAX_PDU_SIZE = 65536 +# This isn't spec'ed but is our own reasonable default to play nice with Synapse's +# `max_request_size`/`max_request_body_size`. We chose the same as `MAX_PDU_SIZE` as our +# `max_request_body_size` math is currently limited by 200 `MAX_PDU_SIZE` things. The +# spec for a `/federation/v1/send` request sets the limit at 100 EDU's and 50 PDU's +# which is below that 200 `MAX_PDU_SIZE` limit (`max_request_body_size`). +# +# Allowing oversized EDU's results in failed `/federation/v1/send` transactions (because +# the request overall can overrun the `max_request_body_size`) which are retried over +# and over and prevent other outbound federation traffic from happening. MAX_EDU_SIZE = 65536 # Max/min size of ints in canonical JSON From 1f9d243ae77bd0fe0fb08b47e9e608a53f48eb9f Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 15:56:20 +0200 Subject: [PATCH 13/50] Apply suggestion from @MadLittleMods Co-authored-by: Eric Eastwood --- synapse/handlers/devicemessage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 7df5ac8b128..aa6422fcd06 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -469,7 +469,7 @@ def get_device_message_edu_contents( if current_edu_size + message_entry_size > MAX_EDU_SIZE: edu_contents.append(current_edu_content) logger.debug( - "Splitting %d device messages from %s into EDU msgid %s, %d EDUs queued", + "Splitting %d to-device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", len(current_edu_content["messages"]), sender_user_id, current_edu_content["message_id"], From b503f9782a065b42bafa00e57ef2faf940a63bdf Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 15:57:01 +0200 Subject: [PATCH 14/50] Apply suggestion from @MadLittleMods Co-authored-by: Eric Eastwood --- synapse/handlers/devicemessage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index aa6422fcd06..d9103eb16c9 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -487,7 +487,7 @@ def get_device_message_edu_contents( if len(current_edu_content["messages"]) > 0: edu_contents.append(current_edu_content) logger.debug( - "Queuing last %d device messages from %s into EDU msgid %s, %d EDUs queued", + "Splitting remaining %d device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", len(current_edu_content["messages"]), sender_user_id, current_edu_content["message_id"], From eef81324e6b70f412ea927eab0eeb057ee7e4892 Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 18:33:58 +0200 Subject: [PATCH 15/50] add comment and add helper create_new_to_device_edu_content() --- synapse/handlers/devicemessage.py | 58 ++++++++++++++++++++++++++----- 1 file changed, 49 insertions(+), 9 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index d9103eb16c9..3c294438e9b 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -20,7 +20,6 @@ # import logging -from copy import deepcopy from http import HTTPStatus from typing import TYPE_CHECKING, Any, Dict, List, Optional @@ -426,9 +425,21 @@ def get_device_message_edu_contents( context: Dict[str, Any], ) -> List[JsonDict]: """ - This function takes a dictionary of messages and splits them into several EDUs if needed. - - It will raise an EventSizeError if a single message is too large to fit into an EDU. + This function takes a dictionary of to-device messages and splits them into several + EDUs by recipient if necessary as the overall request can overrun the + `max_request_body_size` and prevent outbound federation traffic because of the size + of the transaction (cf. MAX_EDU_SIZE). + + Args: + sender_user_id: The user that is sending the to-device messages. + message_type: The type of to-device messages that are being sent. + messages: A dictionary containing recipients mapped to messages intended for them. + context: The span context for opentracing + + Returns: + A list of dictionary containing the EDUs of to-device messages + Raises an EventSizeError if a single to-device message is too large + to fit into an EDU. """ BASE_EDU_CONTENT = { @@ -437,7 +448,8 @@ def get_device_message_edu_contents( "type": message_type, "message_id": random_string(16), } - # This is the size of the full EDU without any messages and without the opentracing context + # This is the size of the full EDU without any messages and without the + # opentracing context as this is not sent as part of the transaction BASE_EDU_SIZE = len( encode_canonical_json( { @@ -450,11 +462,17 @@ def get_device_message_edu_contents( edu_contents = [] - current_edu_content: JsonDict = deepcopy(BASE_EDU_CONTENT) + current_edu_content: JsonDict = create_new_to_device_edu_content( + BASE_EDU_CONTENT['sender'], + BASE_EDU_CONTENT['type'], + BASE_EDU_CONTENT['org.matrix.opentracing_context'], + BASE_EDU_CONTENT['message_id'], + ) current_edu_size = BASE_EDU_SIZE for recipient, message in messages.items(): - # We remove 2 for the curly braces and add 1 for the colon + # As curly braces is already taken into account in BASE_EDU_CONTENT["messages"] + # we remove 2 for those and add 1 for the comma per message message_entry_size = len(encode_canonical_json({recipient: message})) - 2 + 1 if BASE_EDU_SIZE + message_entry_size > MAX_EDU_SIZE: @@ -476,8 +494,11 @@ def get_device_message_edu_contents( len(edu_contents), ) - current_edu_content = deepcopy(BASE_EDU_CONTENT) - current_edu_content["message_id"] = random_string(16) + current_edu_content = create_new_to_device_edu_content( + BASE_EDU_CONTENT['sender'], + BASE_EDU_CONTENT['type'], + BASE_EDU_CONTENT['org.matrix.opentracing_context'], + ) current_edu_size = BASE_EDU_SIZE @@ -495,3 +516,22 @@ def get_device_message_edu_contents( ) return edu_contents + + +def create_new_to_device_edu_content( + sender_user_id: str, + message_type: str, + context: Dict[str, Any], + message_id: str = random_string(16), +) -> JsonDict: + """ + Create a new `m.direct_to_device` EDU `content` object with a unique message ID. + """ + content = { + "messages": {}, + "sender": sender_user_id, + "type": message_type, + "message_id": message_id, + "org.matrix.opentracing_context": json_encoder.encode(context) + } + return content From cf7441aeaaa7b04990f9475e3877c73cadb96256 Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 19:55:19 +0200 Subject: [PATCH 16/50] add helper method and refactor EDUs constant --- synapse/api/constants.py | 5 +++++ .../federation/sender/per_destination_queue.py | 13 ++++++++----- synapse/handlers/devicemessage.py | 17 ++++++++--------- tests/rest/client/test_sendtodevice.py | 12 ++++++++++-- 4 files changed, 31 insertions(+), 16 deletions(-) diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 97a6103ba44..49640b99670 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -39,6 +39,11 @@ # and over and prevent other outbound federation traffic from happening. MAX_EDU_SIZE = 65536 +# This is defined in the Matrix spec and enforced by the receiver. +MAX_EDUS_PER_TRANSACTION = 100 +# A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes +SYNAPSE_EDUS_PER_TRANSACTION = 10 + # Max/min size of ints in canonical JSON CANONICALJSON_MAX_INT = (2**53) - 1 CANONICALJSON_MIN_INT = -CANONICALJSON_MAX_INT diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index 4c844d403a2..b4f193ba92f 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -28,7 +28,11 @@ import attr from prometheus_client import Counter -from synapse.api.constants import EduTypes +from synapse.api.constants import ( + MAX_EDUS_PER_TRANSACTION, + SYNAPSE_EDUS_PER_TRANSACTION, + EduTypes, +) from synapse.api.errors import ( FederationDeniedError, HttpResponseException, @@ -49,9 +53,6 @@ if TYPE_CHECKING: import synapse.server -# This is defined in the Matrix spec and enforced by the receiver. -MAX_EDUS_PER_TRANSACTION = 100 - logger = logging.getLogger(__name__) @@ -773,7 +774,9 @@ async def __aenter__(self) -> Tuple[List[EventBase], List[Edu]]: ( to_device_edus, device_stream_id, - ) = await self.queue._get_to_device_message_edus(edu_limit - 10) + ) = await self.queue._get_to_device_message_edus( + edu_limit - SYNAPSE_EDUS_PER_TRANSACTION + ) if to_device_edus: self._device_stream_id = device_stream_id diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index aa524cca1b9..1336644b067 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -430,11 +430,12 @@ def get_device_message_edu_contents( to fit into an EDU. """ + first_message_id = random_string(16) BASE_EDU_CONTENT = { "messages": {}, "sender": sender_user_id, "type": message_type, - "message_id": random_string(16), + "message_id": first_message_id, } # This is the size of the full EDU without any messages and without the # opentracing context as this is not sent as part of the transaction @@ -451,10 +452,10 @@ def get_device_message_edu_contents( edu_contents = [] current_edu_content: JsonDict = create_new_to_device_edu_content( - BASE_EDU_CONTENT['sender'], - BASE_EDU_CONTENT['type'], - BASE_EDU_CONTENT['org.matrix.opentracing_context'], - BASE_EDU_CONTENT['message_id'], + sender_user_id, + message_type, + context, + first_message_id, ) current_edu_size = BASE_EDU_SIZE @@ -483,9 +484,7 @@ def get_device_message_edu_contents( ) current_edu_content = create_new_to_device_edu_content( - BASE_EDU_CONTENT['sender'], - BASE_EDU_CONTENT['type'], - BASE_EDU_CONTENT['org.matrix.opentracing_context'], + sender_user_id, message_type, context ) current_edu_size = BASE_EDU_SIZE @@ -520,6 +519,6 @@ def create_new_to_device_edu_content( "sender": sender_user_id, "type": message_type, "message_id": message_id, - "org.matrix.opentracing_context": json_encoder.encode(context) + "org.matrix.opentracing_context": json_encoder.encode(context), } return content diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 266c828c142..ce1c9e9482e 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -24,7 +24,12 @@ from twisted.test.proto_helpers import MemoryReactor -from synapse.api.constants import MAX_EDU_SIZE, EduTypes +from synapse.api.constants import ( + MAX_EDU_SIZE, + MAX_EDUS_PER_TRANSACTION, + SYNAPSE_EDUS_PER_TRANSACTION, + EduTypes, +) from synapse.api.errors import Codes from synapse.rest import admin from synapse.rest.client import login, sendtodevice, sync @@ -242,7 +247,10 @@ def test_transaction_splitting(self) -> None: # A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes first_call = mock_send_transaction.call_args_list[0][0][1]() - self.assertEqual(len(first_call["edus"]), 90) + self.assertEqual( + len(first_call["edus"]), + MAX_EDUS_PER_TRANSACTION - SYNAPSE_EDUS_PER_TRANSACTION, + ) second_call = mock_send_transaction.call_args_list[1][0][1]() self.assertEqual(len(second_call["edus"]), 11) From 71a26a4393affc4555a284281de30d270714cbd4 Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Wed, 20 Aug 2025 22:38:36 +0200 Subject: [PATCH 17/50] fix trial tests due to twisted lib issue --- tests/rest/client/test_sendtodevice.py | 173 ++++++++++++++----------- 1 file changed, 98 insertions(+), 75 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index ce1c9e9482e..d0c9db5f63d 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -18,6 +18,7 @@ # [This file includes modifications made by New Vector Limited] # # +import logging from unittest.mock import AsyncMock, Mock from parameterized import parameterized_class @@ -154,106 +155,128 @@ def test_large_remote_todevice(self) -> None: def test_edu_splitting(self) -> None: """Test that a bunch of to-device messages are split into multiple EDUs if they are too large""" - mock_send_transaction: AsyncMock = ( - self.federation_transport_client.send_transaction - ) - mock_send_transaction.return_value = {} + # FIXME: Because huge log line is triggered in this test, + # trial breaks, sometimes (flakily) failing the test run. + # ref: https://github.com/twisted/twisted/issues/12482 + # To remove this, we would need to fix the above issue and + # update, including in olddeps (so several years' wait). + server_logger = logging.getLogger("tests.server") + server_logger_was_disabled = server_logger.disabled + server_logger.disabled = True + try: + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} - sender = self.hs.get_federation_sender() + sender = self.hs.get_federation_sender() - _ = self.register_user("u1", "pass") - user1_tok = self.login("u1", "pass", "d1") - destination = "secondserver" - messages = {} + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} - # 2 small messages that should fit in a single EDU - for i in range(2): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(100)} - } + # 2 small messages that should fit in a single EDU + for i in range(2): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(100)} + } - channel = self.make_request( - "PUT", - "/_matrix/client/r0/sendToDevice/m.test/123456", - content={"messages": messages}, - access_token=user1_tok, - ) - self.assertEqual(channel.code, 200, channel.result) + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/123456", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) - self.get_success(sender.send_device_messages([destination])) + self.get_success(sender.send_device_messages([destination])) - self.pump() + self.pump() - json_cb = mock_send_transaction.call_args[0][1] - data = json_cb() - self.assertEqual(len(data["edus"]), 1) + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(len(data["edus"]), 1) - mock_send_transaction.reset_mock() + mock_send_transaction.reset_mock() - # 2 messages, each just big enough to fit into their own EDU - for i in range(2): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} - } + # 2 messages, each just big enough to fit into their own EDU + for i in range(2): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + } - channel = self.make_request( - "PUT", - "/_matrix/client/r0/sendToDevice/m.test/1234567", - content={"messages": messages}, - access_token=user1_tok, - ) - self.assertEqual(channel.code, 200, channel.result) + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/1234567", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) - self.get_success(sender.send_device_messages([destination])) + self.get_success(sender.send_device_messages([destination])) - self.pump() + self.pump() - json_cb = mock_send_transaction.call_args[0][1] - data = json_cb() - self.assertEqual(len(data["edus"]), 2) + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(len(data["edus"]), 2) + finally: + server_logger.disabled = server_logger_was_disabled def test_transaction_splitting(self) -> None: """Test that a bunch of to-device messages are split into multiple transactions if they are too many EDUs""" - mock_send_transaction: AsyncMock = ( - self.federation_transport_client.send_transaction - ) - mock_send_transaction.return_value = {} + # FIXME: Because huge log line is triggered in this test, + # trial breaks, sometimes (flakily) failing the test run. + # ref: https://github.com/twisted/twisted/issues/12482 + # To remove this, we would need to fix the above issue and + # update, including in olddeps (so several years' wait). + server_logger = logging.getLogger("tests.server") + server_logger_was_disabled = server_logger.disabled + server_logger.disabled = True + try: + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} - sender = self.hs.get_federation_sender() + sender = self.hs.get_federation_sender() - _ = self.register_user("u1", "pass") - user1_tok = self.login("u1", "pass", "d1") - destination = "secondserver" - messages = {} + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} - for i in range(101): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} - } + for i in range(101): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + } - channel = self.make_request( - "PUT", - "/_matrix/client/r0/sendToDevice/m.test/12345678", - content={"messages": messages}, - access_token=user1_tok, - ) - self.assertEqual(channel.code, 200, channel.result) + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/12345678", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) - self.get_success(sender.send_device_messages([destination])) + self.get_success(sender.send_device_messages([destination])) - self.pump() + self.pump() - self.assertEqual(mock_send_transaction.call_count, 2) + self.assertEqual(mock_send_transaction.call_count, 2) - # A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes - first_call = mock_send_transaction.call_args_list[0][0][1]() - self.assertEqual( - len(first_call["edus"]), - MAX_EDUS_PER_TRANSACTION - SYNAPSE_EDUS_PER_TRANSACTION, - ) + # A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes + first_call = mock_send_transaction.call_args_list[0][0][1]() + self.assertEqual( + len(first_call["edus"]), + MAX_EDUS_PER_TRANSACTION - SYNAPSE_EDUS_PER_TRANSACTION, + ) - second_call = mock_send_transaction.call_args_list[1][0][1]() - self.assertEqual(len(second_call["edus"]), 11) + second_call = mock_send_transaction.call_args_list[1][0][1]() + self.assertEqual(len(second_call["edus"]), 11) + finally: + server_logger.disabled = server_logger_was_disabled @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: From 55afc88c33aa15993961966a7522bf11f0c9b73b Mon Sep 17 00:00:00 2001 From: mcalinghee Date: Thu, 21 Aug 2025 11:54:27 +0200 Subject: [PATCH 18/50] try to fix test flaky issue --- tests/rest/client/test_sendtodevice.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index d0c9db5f63d..9beff743712 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -160,7 +160,7 @@ def test_edu_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - server_logger = logging.getLogger("tests.server") + server_logger = logging.getLogger("synapse.storage.SQL") server_logger_was_disabled = server_logger.disabled server_logger.disabled = True try: @@ -231,7 +231,7 @@ def test_transaction_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - server_logger = logging.getLogger("tests.server") + server_logger = logging.getLogger("synapse.storage.SQL") server_logger_was_disabled = server_logger.disabled server_logger.disabled = True try: From 7536ac0fba3a847f273d473e26a6d7ee8aea4dc8 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 23 Dec 2025 16:51:24 +0100 Subject: [PATCH 19/50] Adress comments --- synapse/api/constants.py | 3 ++- synapse/federation/sender/per_destination_queue.py | 4 ++-- tests/rest/client/test_sendtodevice.py | 4 ++-- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 0afb26d0f23..a3f9efe5e16 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -42,7 +42,8 @@ # This is defined in the Matrix spec and enforced by the receiver. MAX_EDUS_PER_TRANSACTION = 100 # A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes -SYNAPSE_EDUS_PER_TRANSACTION = 10 +# like trickling out some device list updates. +NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION = 10 # The maximum allowed size of an HTTP request. # Other than media uploads, the biggest request we expect to see is a fully-loaded diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index 85cfeadb0c1..32f8630c9d1 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -32,7 +32,7 @@ from synapse.api.constants import ( MAX_EDUS_PER_TRANSACTION, - SYNAPSE_EDUS_PER_TRANSACTION, + NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION, EduTypes, ) from synapse.api.errors import ( @@ -800,7 +800,7 @@ async def __aenter__(self) -> tuple[list[EventBase], list[Edu]]: to_device_edus, device_stream_id, ) = await self.queue._get_to_device_message_edus( - edu_limit - SYNAPSE_EDUS_PER_TRANSACTION + edu_limit - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION ) if to_device_edus: diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 43c523f2868..c2001302420 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -26,7 +26,7 @@ from synapse.api.constants import ( MAX_EDU_SIZE, MAX_EDUS_PER_TRANSACTION, - SYNAPSE_EDUS_PER_TRANSACTION, + NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION, EduTypes, ) from synapse.api.errors import Codes @@ -253,7 +253,7 @@ def test_transaction_splitting(self) -> None: first_call = mock_send_transaction.call_args_list[0][0][1]() self.assertEqual( len(first_call["edus"]), - MAX_EDUS_PER_TRANSACTION - SYNAPSE_EDUS_PER_TRANSACTION, + MAX_EDUS_PER_TRANSACTION - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION, ) second_call = mock_send_transaction.call_args_list[1][0][1]() From 206b0b0cff0c715ebafc1d5843bdb6fd05142989 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 23 Dec 2025 17:14:16 +0100 Subject: [PATCH 20/50] Fix mypy --- synapse/storage/databases/main/deviceinbox.py | 108 +++++++----------- tests/rest/client/test_sendtodevice.py | 2 +- 2 files changed, 43 insertions(+), 67 deletions(-) diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index e06e205ebd1..fc61f46c1c5 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -742,15 +742,15 @@ def get_all_new_device_messages_txn( async def add_messages_to_device_inbox( self, local_messages_by_user_then_device: dict[str, dict[str, JsonDict]], - remote_edu_contents: dict[str, list[JsonDict]], + remote_messages_by_destination: dict[str, JsonDict], ) -> int: """Used to send messages from this server. Args: local_messages_by_user_then_device: Dictionary of recipient user_id to recipient device_id to message. - remote_edu_contents: - Dictionary of destination server_name to the list of EDU contents to send. + remote_messages_by_destination: + Dictionary of destination server_name to the EDU JSON to send. Returns: The new stream_id. @@ -759,14 +759,11 @@ async def add_messages_to_device_inbox( assert self._can_write_to_device def add_messages_txn( - txn: LoggingTransaction, - now_ms: int, - stream_ids: list[int], - last_stream_id: int, + txn: LoggingTransaction, now_ms: int, stream_id: int ) -> None: # Add the local messages directly to the local inbox. self._add_messages_to_local_device_inbox_txn( - txn, last_stream_id, local_messages_by_user_then_device + txn, stream_id, local_messages_by_user_then_device ) # Add the remote messages to the federation outbox. @@ -785,76 +782,55 @@ def add_messages_txn( values=[ ( destination, - stream_ids[i], + stream_id, now_ms, - json_encoder.encode(edu_content), + json_encoder.encode(edu), self._instance_name, ) - for destination, edu_contents in remote_edu_contents.items() - for i, edu_content in enumerate(edu_contents) + for destination, edu in remote_messages_by_destination.items() ], ) - for destination, edu_contents in remote_edu_contents.items(): - for i, edu_content in enumerate(edu_contents): - if issue9533_logger.isEnabledFor(logging.DEBUG): - issue9533_logger.debug( - "Queued outgoing to-device messages with " - "stream_id %i, EDU message_id %s, type %s for %s: %s", - stream_ids[i], - edu_content["message_id"], - edu_content["type"], - destination, - [ - f"{user_id}/{device_id} (msgid " - f"{msg.get(EventContentFields.TO_DEVICE_MSGID)})" - for (user_id, messages_by_device) in edu_content[ - "messages" - ].items() - for (device_id, msg) in messages_by_device.items() - ], - ) + for destination, edu in remote_messages_by_destination.items(): + if issue9533_logger.isEnabledFor(logging.DEBUG): + issue9533_logger.debug( + "Queued outgoing to-device messages with " + "stream_id %i, EDU message_id %s, type %s for %s: %s", + stream_id, + edu["message_id"], + edu["type"], + destination, + [ + f"{user_id}/{device_id} (msgid " + f"{msg.get(EventContentFields.TO_DEVICE_MSGID)})" + for (user_id, messages_by_device) in edu["messages"].items() + for (device_id, msg) in messages_by_device.items() + ], + ) - for user_id, messages_by_device in edu_content["messages"].items(): - for device_id, msg in messages_by_device.items(): - with start_active_span("store_outgoing_to_device_message"): - set_tag( - SynapseTags.TO_DEVICE_EDU_ID, edu_content["sender"] - ) - set_tag( - SynapseTags.TO_DEVICE_EDU_ID, - edu_content["message_id"], - ) - set_tag(SynapseTags.TO_DEVICE_TYPE, edu_content["type"]) - set_tag(SynapseTags.TO_DEVICE_RECIPIENT, user_id) - set_tag( - SynapseTags.TO_DEVICE_RECIPIENT_DEVICE, device_id - ) - set_tag( - SynapseTags.TO_DEVICE_MSGID, - msg.get(EventContentFields.TO_DEVICE_MSGID), - ) - - # We allocate one stream id per EDU so we can track if they were - # successfully sent or not. - nb_edus = sum(len(edus) for edus in remote_edu_contents.values()) - async with self._to_device_msg_id_gen.get_next_mult(nb_edus) as stream_ids: - last_stream_id = stream_ids[len(stream_ids) - 1] + for user_id, messages_by_device in edu["messages"].items(): + for device_id, msg in messages_by_device.items(): + with start_active_span("store_outgoing_to_device_message"): + set_tag(SynapseTags.TO_DEVICE_EDU_ID, edu["sender"]) + set_tag(SynapseTags.TO_DEVICE_EDU_ID, edu["message_id"]) + set_tag(SynapseTags.TO_DEVICE_TYPE, edu["type"]) + set_tag(SynapseTags.TO_DEVICE_RECIPIENT, user_id) + set_tag(SynapseTags.TO_DEVICE_RECIPIENT_DEVICE, device_id) + set_tag( + SynapseTags.TO_DEVICE_MSGID, + msg.get(EventContentFields.TO_DEVICE_MSGID), + ) + + async with self._to_device_msg_id_gen.get_next() as stream_id: now_ms = self.clock.time_msec() await self.db_pool.runInteraction( - "add_messages_to_device_inbox", - add_messages_txn, - now_ms, - stream_ids, - last_stream_id, + "add_messages_to_device_inbox", add_messages_txn, now_ms, stream_id ) for user_id in local_messages_by_user_then_device.keys(): - self._device_inbox_stream_cache.entity_has_changed( - user_id, last_stream_id - ) - for destination in remote_edu_contents.keys(): + self._device_inbox_stream_cache.entity_has_changed(user_id, stream_id) + for destination in remote_messages_by_destination.keys(): self._device_federation_outbox_stream_cache.entity_has_changed( - destination, last_stream_id + destination, stream_id ) return self._to_device_msg_id_gen.get_current_token() diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index c2001302420..c2f8edc450a 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -34,7 +34,7 @@ from synapse.rest.client import login, sendtodevice, sync from synapse.server import HomeServer from synapse.types import JsonDict -from synapse.util import Clock +from synapse.util.clock import Clock from synapse.util.stringutils import random_string from tests.unittest import HomeserverTestCase, override_config From d1700d17f6dada6bad14df1aa35a60c5770e5eb7 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 23 Dec 2025 17:16:23 +0100 Subject: [PATCH 21/50] Adress comment --- synapse/handlers/devicemessage.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 5b2590875ed..0bd4032ac36 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -509,11 +509,14 @@ def create_new_to_device_edu_content( sender_user_id: str, message_type: str, context: dict[str, Any], - message_id: str = random_string(16), + message_id: str | None, ) -> JsonDict: """ Create a new `m.direct_to_device` EDU `content` object with a unique message ID. """ + if message_id is None: + message_id = random_string(16) + content = { "messages": {}, "sender": sender_user_id, From 4bb5f261f933f99415af479369b9b030a7876c5e Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 24 Dec 2025 11:40:46 +0100 Subject: [PATCH 22/50] Adress comment --- synapse/handlers/devicemessage.py | 30 ++++++++++++++++-------------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 0bd4032ac36..99a9b3f267f 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -286,12 +286,19 @@ async def send_device_message( context = get_active_span_text_map() - remote_edu_contents = {} + last_stream_id = None for destination, messages in remote_messages.items(): edu_contents = get_device_message_edu_contents( sender_user_id, message_type, messages, context ) - remote_edu_contents[destination] = edu_contents + for edu_content in edu_contents: + remote_edu_content = {destination: edu_content} + # Add messages to the database. + # Retrieve the stream id of the last-processed to-device message. + last_stream_id = await self.store.add_messages_to_device_inbox( + local_messages, remote_edu_content + ) + log_kv( { "destination": destination, @@ -301,17 +308,12 @@ async def send_device_message( } ) - # Add messages to the database. - # Retrieve the stream id of the last-processed to-device message. - last_stream_id = await self.store.add_messages_to_device_inbox( - local_messages, remote_edu_contents - ) - - # Notify listeners that there are new to-device messages to process, - # handing them the latest stream id. - self.notifier.on_new_event( - StreamKeyType.TO_DEVICE, last_stream_id, users=local_messages.keys() - ) + if last_stream_id is not None: + # Notify listeners that there are new to-device messages to process, + # handing them the latest stream id. + self.notifier.on_new_event( + StreamKeyType.TO_DEVICE, last_stream_id, users=local_messages.keys() + ) if self.federation_sender: # Enqueue a new federation transaction to send the new @@ -509,7 +511,7 @@ def create_new_to_device_edu_content( sender_user_id: str, message_type: str, context: dict[str, Any], - message_id: str | None, + message_id: str | None = None, ) -> JsonDict: """ Create a new `m.direct_to_device` EDU `content` object with a unique message ID. From 9a1e64673abc823d901077f1304ed01b519a7687 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 24 Dec 2025 11:49:31 +0100 Subject: [PATCH 23/50] Adress comments --- synapse/handlers/devicemessage.py | 22 +++++++++------------- 1 file changed, 9 insertions(+), 13 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 99a9b3f267f..9d95b94d85f 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -288,25 +288,21 @@ async def send_device_message( last_stream_id = None for destination, messages in remote_messages.items(): - edu_contents = get_device_message_edu_contents( + split_edu_contents = get_device_message_edu_contents( sender_user_id, message_type, messages, context ) - for edu_content in edu_contents: - remote_edu_content = {destination: edu_content} + for edu_contents in split_edu_contents: # Add messages to the database. # Retrieve the stream id of the last-processed to-device message. last_stream_id = await self.store.add_messages_to_device_inbox( - local_messages, remote_edu_content + local_messages, {destination: edu_contents} + ) + log_kv( + { + "destination": destination, + "message_id": edu_contents["message_id"], + } ) - - log_kv( - { - "destination": destination, - "message_ids": [ - edu_content["message_id"] for edu_content in edu_contents - ], - } - ) if last_stream_id is not None: # Notify listeners that there are new to-device messages to process, From fd615d3624d1d9336ad9f6119a8ab57d4ad2e94a Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 24 Dec 2025 11:53:42 +0100 Subject: [PATCH 24/50] Adress comments --- synapse/handlers/devicemessage.py | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 9d95b94d85f..e04db736e1a 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -284,12 +284,10 @@ async def send_device_message( destination = get_domain_from_id(user_id) remote_messages.setdefault(destination, {})[user_id] = by_device - context = get_active_span_text_map() - last_stream_id = None for destination, messages in remote_messages.items(): split_edu_contents = get_device_message_edu_contents( - sender_user_id, message_type, messages, context + sender_user_id, message_type, messages ) for edu_contents in split_edu_contents: # Add messages to the database. @@ -408,7 +406,6 @@ def get_device_message_edu_contents( sender_user_id: str, message_type: str, messages: dict[str, dict[str, JsonDict]], - context: dict[str, Any], ) -> list[JsonDict]: """ This function takes a dictionary of to-device messages and splits them into several @@ -427,7 +424,7 @@ def get_device_message_edu_contents( Raises an EventSizeError if a single to-device message is too large to fit into an EDU. """ - + tracing_context = get_active_span_text_map() first_message_id = random_string(16) BASE_EDU_CONTENT = { "messages": {}, @@ -445,14 +442,16 @@ def get_device_message_edu_contents( } ) ) - BASE_EDU_CONTENT["org.matrix.opentracing_context"] = json_encoder.encode(context) + BASE_EDU_CONTENT["org.matrix.opentracing_context"] = json_encoder.encode( + tracing_context + ) edu_contents = [] current_edu_content: JsonDict = create_new_to_device_edu_content( sender_user_id, message_type, - context, + tracing_context, first_message_id, ) current_edu_size = BASE_EDU_SIZE @@ -482,7 +481,7 @@ def get_device_message_edu_contents( ) current_edu_content = create_new_to_device_edu_content( - sender_user_id, message_type, context + sender_user_id, message_type, tracing_context ) current_edu_size = BASE_EDU_SIZE @@ -506,7 +505,7 @@ def get_device_message_edu_contents( def create_new_to_device_edu_content( sender_user_id: str, message_type: str, - context: dict[str, Any], + tracing_context: dict[str, Any], message_id: str | None = None, ) -> JsonDict: """ @@ -520,6 +519,6 @@ def create_new_to_device_edu_content( "sender": sender_user_id, "type": message_type, "message_id": message_id, - "org.matrix.opentracing_context": json_encoder.encode(context), + "org.matrix.opentracing_context": json_encoder.encode(tracing_context), } return content From 5c115b9000928c3fc639ae7c2ebb45fa65067a13 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 24 Dec 2025 12:01:53 +0100 Subject: [PATCH 25/50] Adress comments --- synapse/handlers/devicemessage.py | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index e04db736e1a..917a07cec57 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -284,16 +284,19 @@ async def send_device_message( destination = get_domain_from_id(user_id) remote_messages.setdefault(destination, {})[user_id] = by_device - last_stream_id = None + # Add local messages to the database. + # Retrieve the stream id of the last-processed to-device message. + last_stream_id = await self.store.add_messages_to_device_inbox( + local_messages, {} + ) for destination, messages in remote_messages.items(): split_edu_contents = get_device_message_edu_contents( sender_user_id, message_type, messages ) for edu_contents in split_edu_contents: - # Add messages to the database. - # Retrieve the stream id of the last-processed to-device message. + # Add remote messages to the database. last_stream_id = await self.store.add_messages_to_device_inbox( - local_messages, {destination: edu_contents} + {}, {destination: edu_contents} ) log_kv( { @@ -302,12 +305,11 @@ async def send_device_message( } ) - if last_stream_id is not None: - # Notify listeners that there are new to-device messages to process, - # handing them the latest stream id. - self.notifier.on_new_event( - StreamKeyType.TO_DEVICE, last_stream_id, users=local_messages.keys() - ) + # Notify listeners that there are new to-device messages to process, + # handing them the latest stream id. + self.notifier.on_new_event( + StreamKeyType.TO_DEVICE, last_stream_id, users=local_messages.keys() + ) if self.federation_sender: # Enqueue a new federation transaction to send the new From ca9af7ebc0b74aded3c977f5c87db2c695785c6c Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 5 Jan 2026 16:50:35 +0100 Subject: [PATCH 26/50] Address comments --- synapse/handlers/devicemessage.py | 122 +++++++----------- synapse/storage/databases/main/deviceinbox.py | 2 +- 2 files changed, 50 insertions(+), 74 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 917a07cec57..8cb2312add5 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -290,18 +290,21 @@ async def send_device_message( local_messages, {} ) for destination, messages in remote_messages.items(): - split_edu_contents = get_device_message_edu_contents( + split_edus = get_split_device_message_edus( sender_user_id, message_type, messages ) - for edu_contents in split_edu_contents: + for edu in split_edus: + edu["org.matrix.opentracing_context"] = json_encoder.encode( + get_active_span_text_map() + ) # Add remote messages to the database. last_stream_id = await self.store.add_messages_to_device_inbox( - {}, {destination: edu_contents} + {}, {destination: edu} ) log_kv( { "destination": destination, - "message_id": edu_contents["message_id"], + "message_id": edu["message_id"], } ) @@ -404,7 +407,7 @@ async def get_events_for_dehydrated_device( } -def get_device_message_edu_contents( +def get_split_device_message_edus( sender_user_id: str, message_type: str, messages: dict[str, dict[str, JsonDict]], @@ -419,108 +422,81 @@ def get_device_message_edu_contents( sender_user_id: The user that is sending the to-device messages. message_type: The type of to-device messages that are being sent. messages: A dictionary containing recipients mapped to messages intended for them. - context: The span context for opentracing Returns: A list of dictionary containing the EDUs of to-device messages Raises an EventSizeError if a single to-device message is too large to fit into an EDU. """ - tracing_context = get_active_span_text_map() - first_message_id = random_string(16) - BASE_EDU_CONTENT = { - "messages": {}, - "sender": sender_user_id, - "type": message_type, - "message_id": first_message_id, - } - # This is the size of the full EDU without any messages and without the - # opentracing context as this is not sent as part of the transaction - BASE_EDU_SIZE = len( - encode_canonical_json( - { - "edu_type": "m.direct_to_device", - "content": BASE_EDU_CONTENT, - } - ) - ) - BASE_EDU_CONTENT["org.matrix.opentracing_context"] = json_encoder.encode( - tracing_context - ) - - edu_contents = [] + split_edus: list[JsonDict] = [] - current_edu_content: JsonDict = create_new_to_device_edu_content( - sender_user_id, - message_type, - tracing_context, - first_message_id, + current_edu_content: dict[str, JsonDict] = create_new_to_device_edu_content( + sender_user_id, message_type, messages ) - current_edu_size = BASE_EDU_SIZE - for recipient, message in messages.items(): - # As curly braces is already taken into account in BASE_EDU_CONTENT["messages"] - # we remove 2 for those and add 1 for the comma per message - message_entry_size = len(encode_canonical_json({recipient: message})) - 2 + 1 + if len(encode_canonical_json(current_edu_content)) < MAX_EDU_SIZE: + # let's optimize the more common case where everything fits in an EDU + split_edus.append(current_edu_content) + else: + # Otherwise we fall back to something quite slower where we add messages one + # recipient at a time, and check the full size everytime + current_edu_content = create_new_to_device_edu_content( + sender_user_id, message_type + ) + for recipient, messages_by_device in messages.items(): + current_edu_content["messages"][recipient] = messages_by_device + if len(encode_canonical_json(current_edu_content)) > MAX_EDU_SIZE: + current_edu_content["messages"].pop(recipient) + split_edus.append(current_edu_content) + if len(current_edu_content["messages"]) == 0: + raise EventSizeError( + f"device message to {recipient} too large to fit in a single EDU", + unpersistable=True, + ) - if BASE_EDU_SIZE + message_entry_size > MAX_EDU_SIZE: - raise EventSizeError( - f"device message to {recipient} too large to fit in a single EDU", - unpersistable=True, - ) + logger.debug( + "Splitting %d to-device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", + len(current_edu_content["messages"]), + sender_user_id, + current_edu_content["message_id"], + len(split_edus), + ) - if len(current_edu_content["messages"]) > 0: - message_entry_size += 1 # Add 1 for the comma + current_edu_content = create_new_to_device_edu_content( + sender_user_id, message_type, {recipient: messages_by_device} + ) - if current_edu_size + message_entry_size > MAX_EDU_SIZE: - edu_contents.append(current_edu_content) + if len(current_edu_content["messages"]) > 0: + split_edus.append(current_edu_content) logger.debug( - "Splitting %d to-device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", + "Splitting remaining %d device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", len(current_edu_content["messages"]), sender_user_id, current_edu_content["message_id"], - len(edu_contents), - ) - - current_edu_content = create_new_to_device_edu_content( - sender_user_id, message_type, tracing_context + len(split_edus), ) - current_edu_size = BASE_EDU_SIZE - - current_edu_content["messages"][recipient] = message - current_edu_size += message_entry_size - - if len(current_edu_content["messages"]) > 0: - edu_contents.append(current_edu_content) - logger.debug( - "Splitting remaining %d device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", - len(current_edu_content["messages"]), - sender_user_id, - current_edu_content["message_id"], - len(edu_contents), - ) - - return edu_contents + return split_edus def create_new_to_device_edu_content( sender_user_id: str, message_type: str, - tracing_context: dict[str, Any], + messages: dict[str, Any] | None = None, message_id: str | None = None, ) -> JsonDict: """ Create a new `m.direct_to_device` EDU `content` object with a unique message ID. """ + if messages is None: + messages = {} if message_id is None: message_id = random_string(16) content = { - "messages": {}, "sender": sender_user_id, "type": message_type, "message_id": message_id, - "org.matrix.opentracing_context": json_encoder.encode(tracing_context), + "messages": messages, } return content diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index fc61f46c1c5..538e1174ba1 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -792,6 +792,7 @@ def add_messages_txn( ) for destination, edu in remote_messages_by_destination.items(): + # print(edu) if issue9533_logger.isEnabledFor(logging.DEBUG): issue9533_logger.debug( "Queued outgoing to-device messages with " @@ -807,7 +808,6 @@ def add_messages_txn( for (device_id, msg) in messages_by_device.items() ], ) - for user_id, messages_by_device in edu["messages"].items(): for device_id, msg in messages_by_device.items(): with start_active_span("store_outgoing_to_device_message"): From e59be78c5e5d47fe17060c32a176a249cde15fbe Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 5 Jan 2026 16:53:45 +0100 Subject: [PATCH 27/50] Remove useless whitespace change --- synapse/storage/databases/main/deviceinbox.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 538e1174ba1..721764b57c5 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -808,6 +808,7 @@ def add_messages_txn( for (device_id, msg) in messages_by_device.items() ], ) + for user_id, messages_by_device in edu["messages"].items(): for device_id, msg in messages_by_device.items(): with start_active_span("store_outgoing_to_device_message"): From 6b57cd355f32992cd629e01c0e41f4b4f80df1d6 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 6 Jan 2026 10:56:17 +0100 Subject: [PATCH 28/50] Update tests/rest/client/test_sendtodevice.py Co-authored-by: Eric Eastwood --- tests/rest/client/test_sendtodevice.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index c2f8edc450a..1e05845a46a 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -143,9 +143,9 @@ def test_edu_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - server_logger = logging.getLogger("synapse.storage.SQL") - server_logger_was_disabled = server_logger.disabled - server_logger.disabled = True + sql_logger = logging.getLogger("synapse.storage.SQL") + sql_logger_was_disabled = sql_logger.disabled + sql_logger.disabled = True try: mock_send_transaction: AsyncMock = ( self.federation_transport_client.send_transaction From 55afe9a0beebda269ee4e18b3883a55da96260b7 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 6 Jan 2026 10:56:42 +0100 Subject: [PATCH 29/50] Update tests/rest/client/test_sendtodevice.py Co-authored-by: Eric Eastwood --- tests/rest/client/test_sendtodevice.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 1e05845a46a..8718b536349 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -208,7 +208,7 @@ def test_edu_splitting(self) -> None: server_logger.disabled = server_logger_was_disabled def test_transaction_splitting(self) -> None: - """Test that a bunch of to-device messages are split into multiple transactions if they are too many EDUs""" + """Test that a bunch of to-device messages are split into multiple transactions if there are too many EDUs""" # FIXME: Because huge log line is triggered in this test, # trial breaks, sometimes (flakily) failing the test run. # ref: https://github.com/twisted/twisted/issues/12482 From 67e7b50979fb8584b3eefce52d773b259c86cd59 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:07:35 +0100 Subject: [PATCH 30/50] Apply suggestion from @MadLittleMods Co-authored-by: Eric Eastwood --- synapse/handlers/devicemessage.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 8cb2312add5..29cd84f50b2 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -425,8 +425,10 @@ def get_split_device_message_edus( Returns: A list of dictionary containing the EDUs of to-device messages - Raises an EventSizeError if a single to-device message is too large - to fit into an EDU. + + Raises: + EventSizeError: If a single to-device message is too large to fit into an + EDU. """ split_edus: list[JsonDict] = [] From 384ce4d238646037be59b50cb32b5407b9462306 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:08:36 +0100 Subject: [PATCH 31/50] Apply suggestion from @MadLittleMods Co-authored-by: Eric Eastwood --- synapse/handlers/devicemessage.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 29cd84f50b2..64a75d44274 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -413,10 +413,10 @@ def get_split_device_message_edus( messages: dict[str, dict[str, JsonDict]], ) -> list[JsonDict]: """ - This function takes a dictionary of to-device messages and splits them into several - EDUs by recipient if necessary as the overall request can overrun the + This function takes many to-device messages and fits/splits them into several EDUs + as necessary. We split the messages up as the overall request can overrun the `max_request_body_size` and prevent outbound federation traffic because of the size - of the transaction (cf. MAX_EDU_SIZE). + of the transaction (cf. `MAX_EDU_SIZE`). Args: sender_user_id: The user that is sending the to-device messages. From f38f8539f93356fa3f2e30aed2b3b6d3bac777af Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:18:56 +0100 Subject: [PATCH 32/50] Apply suggestions from code review Co-authored-by: Eric Eastwood --- tests/rest/client/test_sendtodevice.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 8718b536349..91c1f02dc22 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -137,7 +137,10 @@ def test_large_remote_todevice(self) -> None: self.assertEqual(Codes.TOO_LARGE, channel.json_body["errcode"]) def test_edu_splitting(self) -> None: - """Test that a bunch of to-device messages are split into multiple EDUs if they are too large""" + """ + Test that a bunch of to-device messages are split over multiple EDUs if they are + collectively too large to fit into a single EDU + """ # FIXME: Because huge log line is triggered in this test, # trial breaks, sometimes (flakily) failing the test run. # ref: https://github.com/twisted/twisted/issues/12482 From 71a2c23ec3e7f8b49de7758912defd39bec9da54 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:28:48 +0100 Subject: [PATCH 33/50] Update tests/rest/client/test_sendtodevice.py Co-authored-by: Eric Eastwood --- tests/rest/client/test_sendtodevice.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 91c1f02dc22..45436093faf 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -126,7 +126,7 @@ def test_large_remote_todevice(self) -> None: user1_tok = self.login("u1", "pass", "d1") # send the message - test_msg = {"foo": random_string(MAX_EDU_SIZE)} + test_msg = {"foo": "a" * MAX_EDU_SIZE} channel = self.make_request( "PUT", "/_matrix/client/r0/sendToDevice/m.test/12345", From cf5ca168937668bfac13fb34bf7d93a279ad9fb8 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:29:47 +0100 Subject: [PATCH 34/50] Update tests/rest/client/test_sendtodevice.py Co-authored-by: Eric Eastwood --- tests/rest/client/test_sendtodevice.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 45436093faf..4966a129b60 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -125,7 +125,7 @@ def test_large_remote_todevice(self) -> None: _ = self.register_user("u1", "pass") user1_tok = self.login("u1", "pass", "d1") - # send the message + # Create a message that is over the `MAX_EDU_SIZE` test_msg = {"foo": "a" * MAX_EDU_SIZE} channel = self.make_request( "PUT", From c34037378155b6b7d1c95e382324b50b8d4f1adc Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:33:16 +0100 Subject: [PATCH 35/50] typos --- tests/rest/client/test_sendtodevice.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 4966a129b60..e3d171629eb 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -208,7 +208,7 @@ def test_edu_splitting(self) -> None: data = json_cb() self.assertEqual(len(data["edus"]), 2) finally: - server_logger.disabled = server_logger_was_disabled + sql_logger.disabled = sql_logger_was_disabled def test_transaction_splitting(self) -> None: """Test that a bunch of to-device messages are split into multiple transactions if there are too many EDUs""" @@ -217,9 +217,9 @@ def test_transaction_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - server_logger = logging.getLogger("synapse.storage.SQL") - server_logger_was_disabled = server_logger.disabled - server_logger.disabled = True + sql_logger = logging.getLogger("synapse.storage.SQL") + sql_logger_was_disabled = sql_logger.disabled + sql_logger.disabled = True try: mock_send_transaction: AsyncMock = ( self.federation_transport_client.send_transaction @@ -262,7 +262,7 @@ def test_transaction_splitting(self) -> None: second_call = mock_send_transaction.call_args_list[1][0][1]() self.assertEqual(len(second_call["edus"]), 11) finally: - server_logger.disabled = server_logger_was_disabled + sql_logger.disabled = sql_logger_was_disabled @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: From 741abb22dbb2aa78b515a4a63aecbd8a6cbb0bd2 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Wed, 7 Jan 2026 18:46:55 +0100 Subject: [PATCH 36/50] Use split in 2 alg --- synapse/handlers/devicemessage.py | 91 +++++++++++++++---------------- 1 file changed, 45 insertions(+), 46 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 64a75d44274..6b7da301da6 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -426,59 +426,58 @@ def get_split_device_message_edus( Returns: A list of dictionary containing the EDUs of to-device messages - Raises: - EventSizeError: If a single to-device message is too large to fit into an - EDU. + Raises: + EventSizeError: If a single to-device message is too large to fit into an + EDU. """ - split_edus: list[JsonDict] = [] - - current_edu_content: dict[str, JsonDict] = create_new_to_device_edu_content( - sender_user_id, message_type, messages - ) - - if len(encode_canonical_json(current_edu_content)) < MAX_EDU_SIZE: - # let's optimize the more common case where everything fits in an EDU - split_edus.append(current_edu_content) - else: - # Otherwise we fall back to something quite slower where we add messages one - # recipient at a time, and check the full size everytime - current_edu_content = create_new_to_device_edu_content( - sender_user_id, message_type - ) - for recipient, messages_by_device in messages.items(): - current_edu_content["messages"][recipient] = messages_by_device - if len(encode_canonical_json(current_edu_content)) > MAX_EDU_SIZE: - current_edu_content["messages"].pop(recipient) - split_edus.append(current_edu_content) - if len(current_edu_content["messages"]) == 0: - raise EventSizeError( - f"device message to {recipient} too large to fit in a single EDU", - unpersistable=True, - ) + split_edus_content: list[JsonDict] = [] + + # Convert messages dict to a list of (recipient, messages_by_device) pairs + message_items = list(messages.items()) + + while message_items: + edu_messages = {} + # Start by trying to fit all remaining messages + target_count = len(message_items) + + while target_count > 0: + # Take the first target_count messages + edu_messages = dict(message_items[:target_count]) + edu_content = create_new_to_device_edu_content( + sender_user_id, message_type, edu_messages + ) + # Let's add the whole EDU structure before testing the size + edu = { + "content": edu_content, + "edu_type": "m.direct_to_device", + } + + if len(encode_canonical_json(edu)) <= MAX_EDU_SIZE: + # It fits! Add this EDU and remove these messages from the list + split_edus_content.append(edu_content) + message_items = message_items[target_count:] logger.debug( - "Splitting %d to-device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", - len(current_edu_content["messages"]), + "Created EDU with %d recipients from %s (message_id=%s), (total EDUs so far: %d)", + target_count, sender_user_id, - current_edu_content["message_id"], - len(split_edus), - ) - - current_edu_content = create_new_to_device_edu_content( - sender_user_id, message_type, {recipient: messages_by_device} + edu_content["message_id"], + len(split_edus_content), ) + break + else: + if target_count == 1: + # Single recipient's messages are too large + recipient = message_items[0][0] + raise EventSizeError( + f"device message to {recipient} too large to fit in a single EDU", + unpersistable=True, + ) - if len(current_edu_content["messages"]) > 0: - split_edus.append(current_edu_content) - logger.debug( - "Splitting remaining %d device messages from %s into EDU (message_id=%s), (total EDUs so far: %d)", - len(current_edu_content["messages"]), - sender_user_id, - current_edu_content["message_id"], - len(split_edus), - ) + # Halve the number of messages and try again + target_count = target_count // 2 - return split_edus + return split_edus_content def create_new_to_device_edu_content( From 6c1606c55adcd85c83f43abfdcea250f218eedbe Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 17 Mar 2026 15:33:36 +0100 Subject: [PATCH 37/50] improve tests --- tests/rest/client/test_sendtodevice.py | 51 ++++++++++++++++++-------- 1 file changed, 36 insertions(+), 15 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index e3d171629eb..d9bb7a46594 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -136,7 +136,7 @@ def test_large_remote_todevice(self) -> None: self.assertEqual(channel.code, 413, channel.result) self.assertEqual(Codes.TOO_LARGE, channel.json_body["errcode"]) - def test_edu_splitting(self) -> None: + def test_edu_large_messages_splitting(self) -> None: """ Test that a bunch of to-device messages are split over multiple EDUs if they are collectively too large to fit into a single EDU @@ -162,15 +162,15 @@ def test_edu_splitting(self) -> None: destination = "secondserver" messages = {} - # 2 small messages that should fit in a single EDU + # 2 messages, each just big enough to fit into their own EDU for i in range(2): messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(100)} + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} } channel = self.make_request( "PUT", - "/_matrix/client/r0/sendToDevice/m.test/123456", + "/_matrix/client/r0/sendToDevice/m.test/1234567", content={"messages": messages}, access_token=user1_tok, ) @@ -178,23 +178,46 @@ def test_edu_splitting(self) -> None: self.get_success(sender.send_device_messages([destination])) - self.pump() - json_cb = mock_send_transaction.call_args[0][1] data = json_cb() - self.assertEqual(len(data["edus"]), 1) + self.assertEqual(len(data["edus"]), 2) + finally: + sql_logger.disabled = sql_logger_was_disabled + + def test_edu_small_messages_not_splitting(self) -> None: + """ + Test that a couple of small messages do not get split into multiple EDUs + """ + # FIXME: Because huge log line is triggered in this test, + # trial breaks, sometimes (flakily) failing the test run. + # ref: https://github.com/twisted/twisted/issues/12482 + # To remove this, we would need to fix the above issue and + # update, including in olddeps (so several years' wait). + sql_logger = logging.getLogger("synapse.storage.SQL") + sql_logger_was_disabled = sql_logger.disabled + sql_logger.disabled = True + try: + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} - mock_send_transaction.reset_mock() + sender = self.hs.get_federation_sender() - # 2 messages, each just big enough to fit into their own EDU + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} + + # 2 small messages that should fit in a single EDU for i in range(2): messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + "device": {"foo": random_string(100)} } channel = self.make_request( "PUT", - "/_matrix/client/r0/sendToDevice/m.test/1234567", + "/_matrix/client/r0/sendToDevice/m.test/123456", content={"messages": messages}, access_token=user1_tok, ) @@ -202,11 +225,9 @@ def test_edu_splitting(self) -> None: self.get_success(sender.send_device_messages([destination])) - self.pump() - json_cb = mock_send_transaction.call_args[0][1] data = json_cb() - self.assertEqual(len(data["edus"]), 2) + self.assertEqual(len(data["edus"]), 1) finally: sql_logger.disabled = sql_logger_was_disabled @@ -233,7 +254,7 @@ def test_transaction_splitting(self) -> None: destination = "secondserver" messages = {} - for i in range(101): + for i in range(MAX_EDUS_PER_TRANSACTION + 1): messages[f"@remote_user{i}:" + destination] = { "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} } From eb050422dbf7d017cc7b492d8ef234b4d274e42b Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 17 Mar 2026 15:37:41 +0100 Subject: [PATCH 38/50] Rename split method --- synapse/handlers/devicemessage.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 6b7da301da6..64db242d41b 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -290,7 +290,7 @@ async def send_device_message( local_messages, {} ) for destination, messages in remote_messages.items(): - split_edus = get_split_device_message_edus( + split_edus = split_device_messages_into_edus( sender_user_id, message_type, messages ) for edu in split_edus: @@ -407,7 +407,7 @@ async def get_events_for_dehydrated_device( } -def get_split_device_message_edus( +def split_device_messages_into_edus( sender_user_id: str, message_type: str, messages: dict[str, dict[str, JsonDict]], From 045c0bf50cec47e36ad5fff22f98f2df6b1b7ca1 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 17 Mar 2026 16:32:51 +0100 Subject: [PATCH 39/50] split db method in 2 --- synapse/handlers/devicemessage.py | 12 +++-- synapse/storage/databases/main/deviceinbox.py | 48 +++++++++++++++---- tests/handlers/test_appservice.py | 4 +- 3 files changed, 50 insertions(+), 14 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 64db242d41b..caa052fdad7 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -286,8 +286,10 @@ async def send_device_message( # Add local messages to the database. # Retrieve the stream id of the last-processed to-device message. - last_stream_id = await self.store.add_messages_to_device_inbox( - local_messages, {} + last_stream_id = ( + await self.store.add_local_messages_from_client_to_device_inbox( + local_messages + ) ) for destination, messages in remote_messages.items(): split_edus = split_device_messages_into_edus( @@ -298,8 +300,10 @@ async def send_device_message( get_active_span_text_map() ) # Add remote messages to the database. - last_stream_id = await self.store.add_messages_to_device_inbox( - {}, {destination: edu} + last_stream_id = ( + await self.store.add_remote_messages_from_client_to_device_inbox( + {destination: edu} + ) ) log_kv( { diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 721764b57c5..9fed22c1b3f 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -739,18 +739,15 @@ def get_all_new_device_messages_txn( ) @trace - async def add_messages_to_device_inbox( + async def add_local_messages_from_client_to_device_inbox( self, local_messages_by_user_then_device: dict[str, dict[str, JsonDict]], - remote_messages_by_destination: dict[str, JsonDict], ) -> int: - """Used to send messages from this server. + """Queue local device messages that will be sent to devices of local users. Args: local_messages_by_user_then_device: Dictionary of recipient user_id to recipient device_id to message. - remote_messages_by_destination: - Dictionary of destination server_name to the EDU JSON to send. Returns: The new stream_id. @@ -766,6 +763,39 @@ def add_messages_txn( txn, stream_id, local_messages_by_user_then_device ) + async with self._to_device_msg_id_gen.get_next() as stream_id: + now_ms = self.clock.time_msec() + await self.db_pool.runInteraction( + "add_local_messages_from_client_to_device_inbox", + add_messages_txn, + now_ms, + stream_id, + ) + for user_id in local_messages_by_user_then_device.keys(): + self._device_inbox_stream_cache.entity_has_changed(user_id, stream_id) + + return self._to_device_msg_id_gen.get_current_token() + + @trace + async def add_remote_messages_from_client_to_device_inbox( + self, + remote_messages_by_destination: dict[str, JsonDict], + ) -> int: + """Queue device messages that will be sent to remote servers. + + Args: + remote_messages_by_destination: + Dictionary of destination server_name to the EDU JSON to send. + + Returns: + The new stream_id. + """ + + assert self._can_write_to_device + + def add_messages_txn( + txn: LoggingTransaction, now_ms: int, stream_id: int + ) -> None: # Add the remote messages to the federation outbox. # We'll send them to a remote server when we next send a # federation transaction to that destination. @@ -792,7 +822,6 @@ def add_messages_txn( ) for destination, edu in remote_messages_by_destination.items(): - # print(edu) if issue9533_logger.isEnabledFor(logging.DEBUG): issue9533_logger.debug( "Queued outgoing to-device messages with " @@ -825,10 +854,11 @@ def add_messages_txn( async with self._to_device_msg_id_gen.get_next() as stream_id: now_ms = self.clock.time_msec() await self.db_pool.runInteraction( - "add_messages_to_device_inbox", add_messages_txn, now_ms, stream_id + "add_remote_messages_from_client_to_device_inbox", + add_messages_txn, + now_ms, + stream_id, ) - for user_id in local_messages_by_user_then_device.keys(): - self._device_inbox_stream_cache.entity_has_changed(user_id, stream_id) for destination in remote_messages_by_destination.keys(): self._device_federation_outbox_stream_cache.entity_has_changed( destination, stream_id diff --git a/tests/handlers/test_appservice.py b/tests/handlers/test_appservice.py index 6336edb108a..ac7411291bc 100644 --- a/tests/handlers/test_appservice.py +++ b/tests/handlers/test_appservice.py @@ -856,7 +856,9 @@ def test_application_services_receive_bursts_of_to_device(self) -> None: # Seed the device_inbox table with our fake messages self.get_success( - self.hs.get_datastores().main.add_messages_to_device_inbox(messages, {}) + self.hs.get_datastores().main.add_local_messages_from_client_to_device_inbox( + messages + ) ) # Now have local_user send a final to-device message to exclusive_as_user. All unsent From 64064157e0fcacc8d3b567d296fd3fb386ceb2c6 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 17 Mar 2026 16:52:57 +0100 Subject: [PATCH 40/50] Naming and comments --- synapse/handlers/devicemessage.py | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index caa052fdad7..db8098674c9 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -42,7 +42,7 @@ ) from synapse.types import JsonDict, Requester, StreamKeyType, UserID, get_domain_from_id from synapse.util.json import json_encoder -from synapse.util.stringutils import random_string +from synapse.util.stringutils import random_string_insecure_fast if TYPE_CHECKING: from synapse.server import HomeServer @@ -414,7 +414,7 @@ async def get_events_for_dehydrated_device( def split_device_messages_into_edus( sender_user_id: str, message_type: str, - messages: dict[str, dict[str, JsonDict]], + messages_by_user_then_device: dict[str, dict[str, JsonDict]], ) -> list[JsonDict]: """ This function takes many to-device messages and fits/splits them into several EDUs @@ -425,19 +425,18 @@ def split_device_messages_into_edus( Args: sender_user_id: The user that is sending the to-device messages. message_type: The type of to-device messages that are being sent. - messages: A dictionary containing recipients mapped to messages intended for them. + messages_by_user_then_device: Dictionary of recipient user_id to recipient device_id to message. Returns: - A list of dictionary containing the EDUs of to-device messages + Bin-packed list of EDU JSON content for the given to_device messages Raises: - EventSizeError: If a single to-device message is too large to fit into an - EDU. + EventSizeError: If a single to-device message is too large to fit into an EDU. """ split_edus_content: list[JsonDict] = [] # Convert messages dict to a list of (recipient, messages_by_device) pairs - message_items = list(messages.items()) + message_items = list(messages_by_user_then_device.items()) while message_items: edu_messages = {} @@ -487,21 +486,16 @@ def split_device_messages_into_edus( def create_new_to_device_edu_content( sender_user_id: str, message_type: str, - messages: dict[str, Any] | None = None, - message_id: str | None = None, + messages_by_user_then_device: dict[str, dict[str, JsonDict]], ) -> JsonDict: """ Create a new `m.direct_to_device` EDU `content` object with a unique message ID. """ - if messages is None: - messages = {} - if message_id is None: - message_id = random_string(16) - + message_id = random_string_insecure_fast(16) content = { "sender": sender_user_id, "type": message_type, "message_id": message_id, - "messages": messages, + "messages": messages_by_user_then_device, } return content From 5347b789a61308016c94df21dbb682eca0485e78 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 17 Mar 2026 16:56:24 +0100 Subject: [PATCH 41/50] Add comment --- synapse/handlers/devicemessage.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index db8098674c9..5d423e5a9e4 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -229,6 +229,7 @@ async def send_device_message( set_tag(SynapseTags.TO_DEVICE_TYPE, message_type) set_tag(SynapseTags.TO_DEVICE_SENDER, sender_user_id) local_messages = {} + # Map from destination (server) -> recipient (user ID) -> device_id -> JSON message content remote_messages: dict[str, dict[str, dict[str, JsonDict]]] = {} for user_id, by_device in messages.items(): if not UserID.is_valid(user_id): From 4aa1bab5ca99ff7ad421ec837a510d27089b50ea Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 17 Mar 2026 17:01:50 +0100 Subject: [PATCH 42/50] Add comment --- synapse/handlers/devicemessage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 5d423e5a9e4..2972b331927 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -471,7 +471,7 @@ def split_device_messages_into_edus( break else: if target_count == 1: - # Single recipient's messages are too large + # Single recipient's messages are too large, let's reject the client call with `M_TOO_LARGE` recipient = message_items[0][0] raise EventSizeError( f"device message to {recipient} too large to fit in a single EDU", From ab2951bfdeba9621c0f263eb13b3eccaea5ac398 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Fri, 20 Mar 2026 17:15:18 +0100 Subject: [PATCH 43/50] Address comments --- synapse/handlers/devicemessage.py | 7 +++++-- tests/rest/client/test_sendtodevice.py | 16 +++++++--------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 2972b331927..ddee4d74963 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -453,7 +453,7 @@ def split_device_messages_into_edus( # Let's add the whole EDU structure before testing the size edu = { "content": edu_content, - "edu_type": "m.direct_to_device", + "edu_type": EduTypes.DIRECT_TO_DEVICE, } if len(encode_canonical_json(edu)) <= MAX_EDU_SIZE: @@ -471,7 +471,8 @@ def split_device_messages_into_edus( break else: if target_count == 1: - # Single recipient's messages are too large, let's reject the client call with `M_TOO_LARGE` + # Single recipient's messages are too large, let's reject the client call with `M_TOO_LARGE`, + # we expect this error to reach the client in the case of the /sendToDevice endpoint. recipient = message_items[0][0] raise EventSizeError( f"device message to {recipient} too large to fit in a single EDU", @@ -492,6 +493,8 @@ def create_new_to_device_edu_content( """ Create a new `m.direct_to_device` EDU `content` object with a unique message ID. """ + # The EDU contains a "message_id" property which is used for + # idempotence. Make up a random one. message_id = random_string_insecure_fast(16) content = { "sender": sender_user_id, diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index d9bb7a46594..b26455e25fa 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -26,7 +26,6 @@ from synapse.api.constants import ( MAX_EDU_SIZE, MAX_EDUS_PER_TRANSACTION, - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION, EduTypes, ) from synapse.api.errors import Codes @@ -254,7 +253,9 @@ def test_transaction_splitting(self) -> None: destination = "secondserver" messages = {} - for i in range(MAX_EDUS_PER_TRANSACTION + 1): + nb_of_edus_to_send = MAX_EDUS_PER_TRANSACTION + 1 + + for i in range(nb_of_edus_to_send): messages[f"@remote_user{i}:" + destination] = { "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} } @@ -271,17 +272,14 @@ def test_transaction_splitting(self) -> None: self.pump() - self.assertEqual(mock_send_transaction.call_count, 2) + # At least 2 transactions should be sent since we are over the EDU limit per transaction + self.assertGreaterEqual(mock_send_transaction.call_count, 2) - # A transaction can contain up to 100 EDUs but synapse reserves 10 EDUs for other purposes first_call = mock_send_transaction.call_args_list[0][0][1]() + second_call = mock_send_transaction.call_args_list[1][0][1]() self.assertEqual( - len(first_call["edus"]), - MAX_EDUS_PER_TRANSACTION - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION, + len(first_call["edus"]) + len(second_call["edus"]), nb_of_edus_to_send ) - - second_call = mock_send_transaction.call_args_list[1][0][1]() - self.assertEqual(len(second_call["edus"]), 11) finally: sql_logger.disabled = sql_logger_was_disabled From 31f002998390846bad81fde45193dff16b0d4962 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Fri, 20 Mar 2026 17:37:35 +0100 Subject: [PATCH 44/50] remove useless pump --- tests/rest/client/test_sendtodevice.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index b26455e25fa..86580fae0e1 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -270,8 +270,6 @@ def test_transaction_splitting(self) -> None: self.get_success(sender.send_device_messages([destination])) - self.pump() - # At least 2 transactions should be sent since we are over the EDU limit per transaction self.assertGreaterEqual(mock_send_transaction.call_count, 2) From 33ff986a5ed4d952abb38c9554c0e83b0ec20596 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Fri, 20 Mar 2026 17:39:34 +0100 Subject: [PATCH 45/50] Remove try except --- tests/rest/client/test_sendtodevice.py | 36 +++++++++++++------------- 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 86580fae0e1..074eaa9cae0 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -145,10 +145,10 @@ def test_edu_large_messages_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - sql_logger = logging.getLogger("synapse.storage.SQL") - sql_logger_was_disabled = sql_logger.disabled - sql_logger.disabled = True - try: + # sql_logger = logging.getLogger("synapse.storage.SQL") + # sql_logger_was_disabled = sql_logger.disabled + # sql_logger.disabled = True + # try: mock_send_transaction: AsyncMock = ( self.federation_transport_client.send_transaction ) @@ -180,8 +180,8 @@ def test_edu_large_messages_splitting(self) -> None: json_cb = mock_send_transaction.call_args[0][1] data = json_cb() self.assertEqual(len(data["edus"]), 2) - finally: - sql_logger.disabled = sql_logger_was_disabled + # finally: + # sql_logger.disabled = sql_logger_was_disabled def test_edu_small_messages_not_splitting(self) -> None: """ @@ -192,10 +192,10 @@ def test_edu_small_messages_not_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - sql_logger = logging.getLogger("synapse.storage.SQL") - sql_logger_was_disabled = sql_logger.disabled - sql_logger.disabled = True - try: + # sql_logger = logging.getLogger("synapse.storage.SQL") + # sql_logger_was_disabled = sql_logger.disabled + # sql_logger.disabled = True + # try: mock_send_transaction: AsyncMock = ( self.federation_transport_client.send_transaction ) @@ -227,8 +227,8 @@ def test_edu_small_messages_not_splitting(self) -> None: json_cb = mock_send_transaction.call_args[0][1] data = json_cb() self.assertEqual(len(data["edus"]), 1) - finally: - sql_logger.disabled = sql_logger_was_disabled + # finally: + # sql_logger.disabled = sql_logger_was_disabled def test_transaction_splitting(self) -> None: """Test that a bunch of to-device messages are split into multiple transactions if there are too many EDUs""" @@ -237,10 +237,10 @@ def test_transaction_splitting(self) -> None: # ref: https://github.com/twisted/twisted/issues/12482 # To remove this, we would need to fix the above issue and # update, including in olddeps (so several years' wait). - sql_logger = logging.getLogger("synapse.storage.SQL") - sql_logger_was_disabled = sql_logger.disabled - sql_logger.disabled = True - try: + # sql_logger = logging.getLogger("synapse.storage.SQL") + # sql_logger_was_disabled = sql_logger.disabled + # sql_logger.disabled = True + # try: mock_send_transaction: AsyncMock = ( self.federation_transport_client.send_transaction ) @@ -278,8 +278,8 @@ def test_transaction_splitting(self) -> None: self.assertEqual( len(first_call["edus"]) + len(second_call["edus"]), nb_of_edus_to_send ) - finally: - sql_logger.disabled = sql_logger_was_disabled + # finally: + # sql_logger.disabled = sql_logger_was_disabled @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: From ff046f79378ae3370ecbb8557be6254c0c897cd1 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Fri, 20 Mar 2026 17:41:36 +0100 Subject: [PATCH 46/50] remove workaround for test failures --- tests/rest/client/test_sendtodevice.py | 192 ++++++++++--------------- 1 file changed, 79 insertions(+), 113 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 074eaa9cae0..2cf79342c70 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -18,7 +18,6 @@ # [This file includes modifications made by New Vector Limited] # # -import logging from unittest.mock import AsyncMock, Mock from twisted.test.proto_helpers import MemoryReactor @@ -140,146 +139,113 @@ def test_edu_large_messages_splitting(self) -> None: Test that a bunch of to-device messages are split over multiple EDUs if they are collectively too large to fit into a single EDU """ - # FIXME: Because huge log line is triggered in this test, - # trial breaks, sometimes (flakily) failing the test run. - # ref: https://github.com/twisted/twisted/issues/12482 - # To remove this, we would need to fix the above issue and - # update, including in olddeps (so several years' wait). - # sql_logger = logging.getLogger("synapse.storage.SQL") - # sql_logger_was_disabled = sql_logger.disabled - # sql_logger.disabled = True - # try: - mock_send_transaction: AsyncMock = ( - self.federation_transport_client.send_transaction - ) - mock_send_transaction.return_value = {} + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} - sender = self.hs.get_federation_sender() + sender = self.hs.get_federation_sender() - _ = self.register_user("u1", "pass") - user1_tok = self.login("u1", "pass", "d1") - destination = "secondserver" - messages = {} + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} - # 2 messages, each just big enough to fit into their own EDU - for i in range(2): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} - } + # 2 messages, each just big enough to fit into their own EDU + for i in range(2): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + } - channel = self.make_request( - "PUT", - "/_matrix/client/r0/sendToDevice/m.test/1234567", - content={"messages": messages}, - access_token=user1_tok, - ) - self.assertEqual(channel.code, 200, channel.result) + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/1234567", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) - self.get_success(sender.send_device_messages([destination])) + self.get_success(sender.send_device_messages([destination])) - json_cb = mock_send_transaction.call_args[0][1] - data = json_cb() - self.assertEqual(len(data["edus"]), 2) - # finally: - # sql_logger.disabled = sql_logger_was_disabled + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(len(data["edus"]), 2) def test_edu_small_messages_not_splitting(self) -> None: """ Test that a couple of small messages do not get split into multiple EDUs """ - # FIXME: Because huge log line is triggered in this test, - # trial breaks, sometimes (flakily) failing the test run. - # ref: https://github.com/twisted/twisted/issues/12482 - # To remove this, we would need to fix the above issue and - # update, including in olddeps (so several years' wait). - # sql_logger = logging.getLogger("synapse.storage.SQL") - # sql_logger_was_disabled = sql_logger.disabled - # sql_logger.disabled = True - # try: - mock_send_transaction: AsyncMock = ( - self.federation_transport_client.send_transaction - ) - mock_send_transaction.return_value = {} + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} - sender = self.hs.get_federation_sender() + sender = self.hs.get_federation_sender() - _ = self.register_user("u1", "pass") - user1_tok = self.login("u1", "pass", "d1") - destination = "secondserver" - messages = {} + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} - # 2 small messages that should fit in a single EDU - for i in range(2): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(100)} - } + # 2 small messages that should fit in a single EDU + for i in range(2): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(100)} + } - channel = self.make_request( - "PUT", - "/_matrix/client/r0/sendToDevice/m.test/123456", - content={"messages": messages}, - access_token=user1_tok, - ) - self.assertEqual(channel.code, 200, channel.result) + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/123456", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) - self.get_success(sender.send_device_messages([destination])) + self.get_success(sender.send_device_messages([destination])) - json_cb = mock_send_transaction.call_args[0][1] - data = json_cb() - self.assertEqual(len(data["edus"]), 1) - # finally: - # sql_logger.disabled = sql_logger_was_disabled + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(len(data["edus"]), 1) def test_transaction_splitting(self) -> None: """Test that a bunch of to-device messages are split into multiple transactions if there are too many EDUs""" - # FIXME: Because huge log line is triggered in this test, - # trial breaks, sometimes (flakily) failing the test run. - # ref: https://github.com/twisted/twisted/issues/12482 - # To remove this, we would need to fix the above issue and - # update, including in olddeps (so several years' wait). - # sql_logger = logging.getLogger("synapse.storage.SQL") - # sql_logger_was_disabled = sql_logger.disabled - # sql_logger.disabled = True - # try: - mock_send_transaction: AsyncMock = ( - self.federation_transport_client.send_transaction - ) - mock_send_transaction.return_value = {} + mock_send_transaction: AsyncMock = ( + self.federation_transport_client.send_transaction + ) + mock_send_transaction.return_value = {} - sender = self.hs.get_federation_sender() + sender = self.hs.get_federation_sender() - _ = self.register_user("u1", "pass") - user1_tok = self.login("u1", "pass", "d1") - destination = "secondserver" - messages = {} + _ = self.register_user("u1", "pass") + user1_tok = self.login("u1", "pass", "d1") + destination = "secondserver" + messages = {} - nb_of_edus_to_send = MAX_EDUS_PER_TRANSACTION + 1 + nb_of_edus_to_send = MAX_EDUS_PER_TRANSACTION + 1 - for i in range(nb_of_edus_to_send): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} - } + for i in range(nb_of_edus_to_send): + messages[f"@remote_user{i}:" + destination] = { + "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + } - channel = self.make_request( - "PUT", - "/_matrix/client/r0/sendToDevice/m.test/12345678", - content={"messages": messages}, - access_token=user1_tok, - ) - self.assertEqual(channel.code, 200, channel.result) + channel = self.make_request( + "PUT", + "/_matrix/client/r0/sendToDevice/m.test/12345678", + content={"messages": messages}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) - self.get_success(sender.send_device_messages([destination])) + self.get_success(sender.send_device_messages([destination])) - # At least 2 transactions should be sent since we are over the EDU limit per transaction - self.assertGreaterEqual(mock_send_transaction.call_count, 2) + # At least 2 transactions should be sent since we are over the EDU limit per transaction + self.assertGreaterEqual(mock_send_transaction.call_count, 2) - first_call = mock_send_transaction.call_args_list[0][0][1]() - second_call = mock_send_transaction.call_args_list[1][0][1]() - self.assertEqual( - len(first_call["edus"]) + len(second_call["edus"]), nb_of_edus_to_send - ) - # finally: - # sql_logger.disabled = sql_logger_was_disabled + first_call = mock_send_transaction.call_args_list[0][0][1]() + second_call = mock_send_transaction.call_args_list[1][0][1]() + self.assertEqual( + len(first_call["edus"]) + len(second_call["edus"]), nb_of_edus_to_send + ) @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: From 14a7515b5e411afebc80170b56c5cda9b83e93cd Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Fri, 20 Mar 2026 18:07:33 +0100 Subject: [PATCH 47/50] remove random from tests --- tests/rest/client/test_sendtodevice.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index 2cf79342c70..f9c6d853926 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -33,7 +33,6 @@ from synapse.server import HomeServer from synapse.types import JsonDict from synapse.util.clock import Clock -from synapse.util.stringutils import random_string from tests.unittest import HomeserverTestCase, override_config @@ -154,7 +153,7 @@ def test_edu_large_messages_splitting(self) -> None: # 2 messages, each just big enough to fit into their own EDU for i in range(2): messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + "device": {"foo": "a" * (MAX_EDU_SIZE - 1000)} } channel = self.make_request( @@ -189,9 +188,7 @@ def test_edu_small_messages_not_splitting(self) -> None: # 2 small messages that should fit in a single EDU for i in range(2): - messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(100)} - } + messages[f"@remote_user{i}:" + destination] = {"device": {"foo": "a" * 100}} channel = self.make_request( "PUT", @@ -225,7 +222,7 @@ def test_transaction_splitting(self) -> None: for i in range(nb_of_edus_to_send): messages[f"@remote_user{i}:" + destination] = { - "device": {"foo": random_string(MAX_EDU_SIZE - 1000)} + "device": {"foo": "a" * (MAX_EDU_SIZE - 1000)} } channel = self.make_request( From a55fc076aa5b6d799e238c3b183a750c6f9fcf3f Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 23 Mar 2026 09:38:48 +0100 Subject: [PATCH 48/50] Adress comments --- tests/rest/client/test_sendtodevice.py | 30 +++++++++++++++----------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/tests/rest/client/test_sendtodevice.py b/tests/rest/client/test_sendtodevice.py index f9c6d853926..526d9f0b3c4 100644 --- a/tests/rest/client/test_sendtodevice.py +++ b/tests/rest/client/test_sendtodevice.py @@ -166,9 +166,11 @@ def test_edu_large_messages_splitting(self) -> None: self.get_success(sender.send_device_messages([destination])) - json_cb = mock_send_transaction.call_args[0][1] - data = json_cb() - self.assertEqual(len(data["edus"]), 2) + number_of_edus_sent = 0 + for call in mock_send_transaction.call_args_list: + number_of_edus_sent += len(call[0][1]()["edus"]) + + self.assertEqual(number_of_edus_sent, 2) def test_edu_small_messages_not_splitting(self) -> None: """ @@ -200,9 +202,11 @@ def test_edu_small_messages_not_splitting(self) -> None: self.get_success(sender.send_device_messages([destination])) - json_cb = mock_send_transaction.call_args[0][1] - data = json_cb() - self.assertEqual(len(data["edus"]), 1) + number_of_edus_sent = 0 + for call in mock_send_transaction.call_args_list: + number_of_edus_sent += len(call[0][1]()["edus"]) + + self.assertEqual(number_of_edus_sent, 1) def test_transaction_splitting(self) -> None: """Test that a bunch of to-device messages are split into multiple transactions if there are too many EDUs""" @@ -218,9 +222,9 @@ def test_transaction_splitting(self) -> None: destination = "secondserver" messages = {} - nb_of_edus_to_send = MAX_EDUS_PER_TRANSACTION + 1 + number_of_edus_to_send = MAX_EDUS_PER_TRANSACTION + 1 - for i in range(nb_of_edus_to_send): + for i in range(number_of_edus_to_send): messages[f"@remote_user{i}:" + destination] = { "device": {"foo": "a" * (MAX_EDU_SIZE - 1000)} } @@ -238,11 +242,11 @@ def test_transaction_splitting(self) -> None: # At least 2 transactions should be sent since we are over the EDU limit per transaction self.assertGreaterEqual(mock_send_transaction.call_count, 2) - first_call = mock_send_transaction.call_args_list[0][0][1]() - second_call = mock_send_transaction.call_args_list[1][0][1]() - self.assertEqual( - len(first_call["edus"]) + len(second_call["edus"]), nb_of_edus_to_send - ) + number_of_edus_sent = 0 + for call in mock_send_transaction.call_args_list: + number_of_edus_sent += len(call[0][1]()["edus"]) + + self.assertEqual(number_of_edus_sent, number_of_edus_to_send) @override_config({"rc_key_requests": {"per_second": 10, "burst_count": 2}}) def test_local_room_key_request(self) -> None: From 6c93d88a00ee8922d37ed8d0f4acc8132ad69b75 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Mon, 23 Mar 2026 22:33:39 +0100 Subject: [PATCH 49/50] Add spec link --- synapse/handlers/devicemessage.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index ddee4d74963..06778e89ba8 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -473,6 +473,7 @@ def split_device_messages_into_edus( if target_count == 1: # Single recipient's messages are too large, let's reject the client call with `M_TOO_LARGE`, # we expect this error to reach the client in the case of the /sendToDevice endpoint. + # cf https://github.com/matrix-org/matrix-spec/pull/2340 for the matching spec change proposal. recipient = message_items[0][0] raise EventSizeError( f"device message to {recipient} too large to fit in a single EDU", From 4b7da00420945815e19f400c350b3f4e58f24cb4 Mon Sep 17 00:00:00 2001 From: Mathieu Velten Date: Tue, 24 Mar 2026 09:59:06 +0100 Subject: [PATCH 50/50] Update synapse/handlers/devicemessage.py Co-authored-by: Eric Eastwood --- synapse/handlers/devicemessage.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 06778e89ba8..2096b44f6ca 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -471,9 +471,14 @@ def split_device_messages_into_edus( break else: if target_count == 1: - # Single recipient's messages are too large, let's reject the client call with `M_TOO_LARGE`, - # we expect this error to reach the client in the case of the /sendToDevice endpoint. - # cf https://github.com/matrix-org/matrix-spec/pull/2340 for the matching spec change proposal. + # Single recipient's messages are too large, let's reject the client + # call with 413/`M_TOO_LARGE`, we expect this error to reach the + # client in the case of the /sendToDevice endpoint. + # + # 413 is currently an unspecced response for `/sendToDevice` but is + # probably the best thing we can do. + # https://github.com/matrix-org/matrix-spec/pull/2340 tracks adding + # this to the spec recipient = message_items[0][0] raise EventSizeError( f"device message to {recipient} too large to fit in a single EDU",