Skip to content

Revert "Limit outgoing to_device EDU size to 65536" #19614

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
erikjohnston merged 1 commit into from
Mar 27, 2026
Merged

Revert "Limit outgoing to_device EDU size to 65536" #19614

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion changelog.d/18416.bugfix
View file
Open in desktop

This file was deleted.

16 changes: 0 additions & 16 deletions synapse/api/constants.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,22 +30,6 @@

# 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

# 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
Expand Down
13 changes: 5 additions & 8 deletions synapse/federation/sender/per_destination_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,7 @@

from twisted.internet import defer

from synapse.api.constants import (
MAX_EDUS_PER_TRANSACTION,
NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION,
EduTypes,
)
from synapse.api.constants import EduTypes
from synapse.api.errors import (
FederationDeniedError,
HttpResponseException,
Expand All @@ -55,6 +51,9 @@
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__)


Expand Down Expand Up @@ -799,9 +798,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 - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION
)
) = await self.queue._get_to_device_message_edus(edu_limit - 10)

if to_device_edus:
self._device_stream_id = device_stream_id
Expand Down
158 changes: 23 additions & 135 deletions synapse/handlers/devicemessage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,8 @@
from http import HTTPStatus
from typing import TYPE_CHECKING, Any

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.constants import EduTypes, EventContentFields, ToDeviceEventTypes
from synapse.api.errors import Codes, SynapseError
from synapse.api.ratelimiting import Ratelimiter
from synapse.logging.context import run_in_background
from synapse.logging.opentracing import (
Expand All @@ -42,7 +35,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_insecure_fast
from synapse.util.stringutils import random_string

if TYPE_CHECKING:
from synapse.server import HomeServer
Expand Down Expand Up @@ -229,7 +222,6 @@ 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):
Expand Down Expand Up @@ -285,33 +277,28 @@ async def send_device_message(
destination = get_domain_from_id(user_id)
remote_messages.setdefault(destination, {})[user_id] = by_device

# Add local messages to the database.
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.
# Retrieve the stream id of the last-processed to-device message.
last_stream_id = (
await self.store.add_local_messages_from_client_to_device_inbox(
local_messages
)
last_stream_id = await self.store.add_messages_to_device_inbox(
local_messages, remote_edu_contents
)
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.
Expand Down Expand Up @@ -410,102 +397,3 @@ 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. `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

Raises:
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_by_user_then_device.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": EduTypes.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(
"Created EDU with %d recipients from %s (message_id=%s), (total EDUs so far: %d)",
target_count,
sender_user_id,
edu_content["message_id"],
len(split_edus_content),
)
break
else:
if target_count == 1:
# 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",
unpersistable=True,
)

# Halve the number of messages and try again
target_count = target_count // 2

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
47 changes: 8 additions & 39 deletions synapse/storage/databases/main/deviceinbox.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -739,15 +739,18 @@ def get_all_new_device_messages_txn(
)

@trace
async def add_local_messages_from_client_to_device_inbox(
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],
) -> int:
"""Queue local device messages that will be sent to devices of local users.
"""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.

Returns:
The new stream_id.
Expand All @@ -763,39 +766,6 @@ 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.
Expand Down Expand Up @@ -854,11 +824,10 @@ 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_remote_messages_from_client_to_device_inbox",
add_messages_txn,
now_ms,
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, stream_id)
for destination in remote_messages_by_destination.keys():
self._device_federation_outbox_stream_cache.entity_has_changed(
destination, stream_id
Expand Down
4 changes: 1 addition & 3 deletions tests/handlers/test_appservice.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -856,9 +856,7 @@ 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_local_messages_from_client_to_device_inbox(
messages
)
self.hs.get_datastores().main.add_messages_to_device_inbox(messages, {})
)

# Now have local_user send a final to-device message to exclusive_as_user. All unsent
Expand Down
Loading
Loading