Limit to-device EDU sizes

changelog.d/19617.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.

synapse/api/constants.py

@@ -30,6 +30,33 @@
 
 # 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.
+#
+# We may send EDU's that are larger than this, but we aim to avoid doing so.
+SOFT_MAX_EDU_SIZE = 65536
+
+# This is the maximum size of the content of a to-device message. This is not
+# (yet) spec'ed but is our own reasonable default. We need to set a limit on the
+# size of to-device message contents, as they get sent over federation and
+# therefore need to fit inside transactions.
+#
+# https://github.com/matrix-org/matrix-spec/pull/2340 tracks adding this to the
+# spec.
+MAX_TO_DEVICE_CONTENT_SIZE = SOFT_MAX_EDU_SIZE
+
+# 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
+# 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

synapse/federation/sender/per_destination_queue.py

@@ -30,7 +30,11 @@
 
 from twisted.internet import defer
 
-from synapse.api.constants import EduTypes
+from synapse.api.constants import (
+    MAX_EDUS_PER_TRANSACTION,
+    NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION,
+    EduTypes,
+)
 from synapse.api.errors import (
     FederationDeniedError,
     HttpResponseException,
@@ -51,9 +55,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__)
 
 
@@ -798,7 +799,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 - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION
+        )
 
         if to_device_edus:
             self._device_stream_id = device_stream_id

synapse/handlers/devicemessage.py

@@ -23,8 +23,16 @@
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any
 
-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_TO_DEVICE_CONTENT_SIZE,
+    SOFT_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 (
@@ -34,8 +42,9 @@
     set_tag,
 )
 from synapse.types import JsonDict, Requester, StreamKeyType, UserID, get_domain_from_id
+from synapse.util import split_dict_to_fit_to_size
 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
@@ -222,6 +231,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):
@@ -233,6 +243,24 @@ async def send_device_message(
 
             # add an opentracing log entry for each message
             for device_id, message_content in by_device.items():
+                # Check the size of each message, as if these are too large we
+                # can't send them over federation.
+                #
+                # We do this for all to-device messages, even those that aren't
+                # over federation, so as to more easily catch clients that are
+                # sending excessively large messages.
+                if (
+                    message_len := len(encode_canonical_json(message_content))
+                ) > MAX_TO_DEVICE_CONTENT_SIZE:
+                    # 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
+                    raise EventSizeError(
+                        f"To-device message for {user_id}:{device_id} is too large to send ({message_len} > {MAX_TO_DEVICE_CONTENT_SIZE})",
+                        unpersistable=True,
+                    )
+
                 log_kv(
                     {
                         "event": "send_to_device_message",
@@ -277,28 +305,33 @@ 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()
-
-        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),
-            }
-
-        # Add messages to the database.
+        # 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, remote_edu_contents
+        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(
+                sender_user_id, message_type, messages
+            )
+            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_remote_messages_from_client_to_device_inbox(
+                        {destination: edu}
+                    )
+                )
+                log_kv(
+                    {
+                        "destination": destination,
+                        "message_id": edu["message_id"],
+                    }
+                )
 
         # Notify listeners that there are new to-device messages to process,
         # handing them the latest stream id.
@@ -397,3 +430,128 @@ async def get_events_for_dehydrated_device(
             "events": messages,
             "next_batch": f"d{stream_id}",
         }
+
+
+def split_device_messages_into_edus(
+    sender_user_id: str,
+    message_type: str,
+    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
+    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. `SOFT_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_by_user_then_device: Dictionary of recipient user_id to recipient device_id to message.
+
+    Returns:
+        Bin-packed list of EDU JSON content for the given to_device messages
+    """
+    split_edus_content: list[JsonDict] = []
+
+    # The header size of the full EDU.
+    base_edu_size = _EMPTY_EDU_SIZE + len(sender_user_id) + len(message_type)
+
+    # First split up the top-level dict of user_id to device messages.
+    for subset_messages, estimated_size in split_dict_to_fit_to_size(
+        messages_by_user_then_device,
+        soft_max_size=SOFT_MAX_EDU_SIZE,
+        wrapping_object_size=base_edu_size,
+    ):
+        # The returned subset might be larger than the soft max size if it
+        # contains a single entry that is larger than the soft max size.
+        if estimated_size <= SOFT_MAX_EDU_SIZE:
+            # This message fits in a single EDU, add it as is.
+            content = create_new_to_device_edu_content(
+                sender_user_id, message_type, subset_messages
+            )
+            split_edus_content.append(content)
+            logger.debug(
+                "Created EDU with %d recipients from %s (message_id=%s), (total EDUs so far: %d)",
+                len(subset_messages),
+                sender_user_id,
+                content["message_id"],
+                len(split_edus_content),
+            )
+        else:
+            # This message doesn't fit in a single EDU. We split the message up
+            # further by device.
+            #
+            # Note: `subset` should only have a single entry in it.
+            for recipient, messages_by_device in subset_messages.items():
+                # The header size of the EDU for this recipient, which includes
+                # the size of the recipient user ID and the wrapping structure
+                # for the device messages.
+                subset_base_size = len(
+                    encode_canonical_json(
+                        _create_new_to_device_edu(
+                            sender_user_id, message_type, {recipient: {}}
+                        )
+                    )
+                )
+
+                for subset_messages, _estimated_size in split_dict_to_fit_to_size(
+                    messages_by_device,
+                    soft_max_size=SOFT_MAX_EDU_SIZE,
+                    wrapping_object_size=subset_base_size,
+                ):
+                    # Again, the returned subset might be larger than the soft
+                    # max size, but we can't split it any further so we have to
+                    # add it as is.
+                    content = create_new_to_device_edu_content(
+                        sender_user_id, message_type, {recipient: subset_messages}
+                    )
+                    split_edus_content.append(content)
+                    logger.debug(
+                        "Created EDU with %d recipients from %s (message_id=%s), (total EDUs so far: %d)",
+                        len(subset_messages),
+                        sender_user_id,
+                        content["message_id"],
+                        len(split_edus_content),
+                    )
+
+    return split_edus_content
+
+
+def create_new_to_device_edu_content(
+    sender_user_id: str,
+    message_type: str,
+    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.
+    """
+    # 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,
+        "type": message_type,
+        "message_id": message_id,
+        "messages": messages_by_user_then_device,
+    }
+    return content
+
+
+def _create_new_to_device_edu(
+    sender_user_id: str,
+    message_type: str,
+    messages_by_user_then_device: dict[str, dict[str, JsonDict]],
+) -> dict[str, Any]:
+    """Create a new `m.direct_to_device` EDU, returning the full EDU dict."""
+    return {
+        "edu_type": EduTypes.DIRECT_TO_DEVICE,
+        "content": create_new_to_device_edu_content(
+            sender_user_id, message_type, messages_by_user_then_device
+        ),
+    }
+
+
+# The size of an empty EDU with no messages, which we use as the base size when
+# packing messages into EDUs. The size of the sender and message type must be
+# added to this when calculating the size of an EDU.
+_EMPTY_EDU_SIZE = len(encode_canonical_json(_create_new_to_device_edu("", "", {})))

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.
@@ -824,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