diff --git a/changelog.d/19539.bugfix b/changelog.d/19539.bugfix new file mode 100644 index 00000000000..41beb3a179b --- /dev/null +++ b/changelog.d/19539.bugfix @@ -0,0 +1 @@ +[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): Update error responses to match their format in the current draft of the MSC. diff --git a/changelog.d/19539.feature b/changelog.d/19539.feature new file mode 100644 index 00000000000..93eb4cc1b97 --- /dev/null +++ b/changelog.d/19539.feature @@ -0,0 +1 @@ +[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): Limit how many delayed events a user may have scheduled at once. diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs index ef9ecb067ce..b5f641613b3 100644 --- a/rust/src/config/mod.rs +++ b/rust/src/config/mod.rs @@ -47,7 +47,7 @@ pub struct AuthConfig { } #[derive(FromPyObject, Clone)] pub struct ServerConfig { - pub max_event_delay_ms: Option, + pub msc4140_enabled: bool, } #[derive(FromPyObject, Clone)] diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index db8bca82d68..bfe064eb5ed 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -304,10 +304,7 @@ pub fn synapse_config_to_global_unstable_feature_map( msc4028: config.experimental.msc4028_push_encrypted_events, msc4108: config.experimental.msc4108_enabled || (config.experimental.msc4108_delegation_endpoint.is_some()), - msc4140: config - .server - .max_event_delay_ms - .is_some_and(|max_event_delay_ms| max_event_delay_ms > 0), + msc4140: config.server.msc4140_enabled, msc3575: config.experimental.msc3575_enabled, msc4133: config.experimental.msc4133_enabled, msc4133_stable: true, diff --git a/synapse/config/server.py b/synapse/config/server.py index 282679aabe4..695a6cb9d44 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -37,6 +37,7 @@ from synapse.api.room_versions import KNOWN_ROOM_VERSIONS from synapse.types import JsonDict, StrSequence +from synapse.util.duration import Duration from synapse.util.module_loader import load_module from synapse.util.stringutils import parse_and_validate_server_name @@ -949,13 +950,33 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: # The maximum allowed delay duration for delayed events (MSC4140). max_event_delay_duration = config.get("max_event_delay_duration") if max_event_delay_duration is not None: - self.max_event_delay_ms: int | None = self.parse_duration( - max_event_delay_duration - ) - if self.max_event_delay_ms <= 0: - raise ConfigError("max_event_delay_duration must be a positive value") + max_event_delay_ms = self.parse_duration(max_event_delay_duration) + if max_event_delay_ms <= 0: + raise ConfigError( + "'max_event_delay_duration' must be a positive value if set", + ("max_event_delay_duration",), + ) + self.max_event_delay_duration = Duration(milliseconds=max_event_delay_ms) else: - self.max_event_delay_ms = None + self.max_event_delay_duration = Duration() + + # The maximum number of delayed events a user may have scheduled at a time. + # (Defined here despite being experimental to be near the other MSC4140 config) + self.max_delayed_events_per_user: int = config.get( + "experimental_features", {} + ).get("msc4140_max_delayed_events_per_user", 100) + if ( + not isinstance(self.max_delayed_events_per_user, int) + or self.max_delayed_events_per_user < 0 + ): + raise ConfigError( + "'msc4140_max_delayed_events_per_user' must be a non-negative integer", + ("experimental", "msc4140_max_delayed_events_per_user"), + ) + + self.msc4140_enabled = bool( + self.max_delayed_events_per_user and self.max_event_delay_duration + ) def has_tls_listener(self) -> bool: return any(listener.is_tls() for listener in self.listeners) diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index f016d95e314..13d6a54de24 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -13,12 +13,13 @@ # import logging +from http import HTTPStatus from typing import TYPE_CHECKING, Optional from twisted.internet.interfaces import IDelayedCall from synapse.api.constants import EventTypes, StickyEvent, StickyEventField -from synapse.api.errors import ShadowBanError, SynapseError +from synapse.api.errors import Codes, ShadowBanError, SynapseError from synapse.api.ratelimiting import Ratelimiter from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME from synapse.http.site import SynapseRequest @@ -330,7 +331,7 @@ async def add( state_key: str | None, origin_server_ts: int | None, content: JsonDict, - delay: int, + delay: Duration, sticky_duration_ms: int | None, ) -> str: """ @@ -344,20 +345,37 @@ async def add( origin_server_ts: The custom timestamp to send the event with. If None, the timestamp will be the actual time when the event is sent. content: The content of the event to be sent. - delay: How long (in milliseconds) to wait before automatically sending the event. + delay: How long to wait before automatically sending the event. sticky_duration_ms: If an MSC4354 sticky event: the sticky duration (in milliseconds). The event will be attempted to be reliably delivered to clients and remote servers during its sticky period. Returns: The ID of the added delayed event. Raises: - SynapseError: if the delayed event fails validation checks. + SynapseError: if the delayed event fails validation checks, or + if the requested delay is longer than allowed, or + if sending delayed events has been disallowed entirely. """ # Use standard request limiter for scheduling new delayed events. # TODO: Instead apply ratelimiting based on the scheduled send time. # See https://github.com/element-hq/synapse/issues/18021 await self._request_ratelimiter.ratelimit(requester) + if not self._config.server.msc4140_enabled: + raise SynapseError( + HTTPStatus.FORBIDDEN, + "Sending delayed events has been disallowed", + Codes.FORBIDDEN, + ) + if delay > self._config.server.max_event_delay_duration: + requested_delay = delay.as_millis() + max_delay = self._config.server.max_event_delay_duration.as_millis() + raise SynapseError( + HTTPStatus.FORBIDDEN, + f"The requested delay ({requested_delay}ms) exceeds the allowed maximum ({max_delay}ms)", + Codes.FORBIDDEN, + ) + self._event_creation_handler.validator.validate_builder( self._event_creation_handler.event_builder_factory.for_room_version( await self._store.get_room_version(room_id), @@ -384,6 +402,7 @@ async def add( content=content, delay=delay, sticky_duration_ms=sticky_duration_ms, + limit=self._config.server.max_delayed_events_per_user, ) if self._repl_client is not None: diff --git a/synapse/rest/client/capabilities.py b/synapse/rest/client/capabilities.py index 2be5f5849d7..4ddaaeda74e 100644 --- a/synapse/rest/client/capabilities.py +++ b/synapse/rest/client/capabilities.py @@ -109,6 +109,11 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: "capabilities" ]["m.profile_fields"] + response["capabilities"]["org.matrix.msc4140.delayed_events"] = { + "max_delay_ms": self.config.server.max_event_delay_duration.as_millis(), + "max_scheduled": self.config.server.max_delayed_events_per_user, + } + if self.config.experimental.msc4267_enabled: response["capabilities"]["org.matrix.msc4267.forget_forced_upon_leave"] = { "enabled": self.config.room.forget_on_leave, diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py index 36f638e236f..951b4ff08b9 100644 --- a/synapse/rest/client/room.py +++ b/synapse/rest/client/room.py @@ -84,6 +84,7 @@ from synapse.types.state import StateFilter from synapse.util.cancellation import cancellable from synapse.util.clock import Clock +from synapse.util.duration import Duration from synapse.util.events import generate_fake_event_id from synapse.util.stringutils import parse_and_validate_server_name @@ -215,7 +216,6 @@ def __init__(self, hs: "HomeServer"): self.auth = hs.get_auth() self.clock = hs.get_clock() self._event_serializer = hs.get_event_client_serializer() - self._max_event_delay_ms = hs.config.server.max_event_delay_ms self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker self._msc4354_enabled = hs.config.experimental.msc4354_enabled @@ -343,7 +343,7 @@ async def on_PUT( if self._msc4354_enabled: sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) - delay = _parse_request_delay(request, self._max_event_delay_ms) + delay = _parse_request_for_delayed_event_delay(request) if delay is not None: delay_id = await self.delayed_events_handler.add( requester, @@ -416,7 +416,6 @@ def __init__(self, hs: "HomeServer"): self.event_creation_handler = hs.get_event_creation_handler() self.delayed_events_handler = hs.get_delayed_events_handler() self.auth = hs.get_auth() - self._max_event_delay_ms = hs.config.server.max_event_delay_ms self._msc4354_enabled = hs.config.experimental.msc4354_enabled def register(self, http_server: HttpServer) -> None: @@ -442,7 +441,7 @@ async def _do( if self._msc4354_enabled: sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) - delay = _parse_request_delay(request, self._max_event_delay_ms) + delay = _parse_request_for_delayed_event_delay(request) if delay is not None: delay_id = await self.delayed_events_handler.add( requester, @@ -515,47 +514,20 @@ async def on_PUT( ) -def _parse_request_delay( - request: SynapseRequest, - max_delay: int | None, -) -> int | None: +def _parse_request_for_delayed_event_delay(request: SynapseRequest) -> Duration | None: """Parses from the request string the delay parameter for delayed event requests, and checks it for correctness. Args: request: the twisted HTTP request. - max_delay: the maximum allowed value of the delay parameter, - or None if no delay parameter is allowed. Returns: The value of the requested delay, or None if it was absent. Raises: - SynapseError: if the delay parameter is present and forbidden, - or if it exceeds the maximum allowed value. + SynapseError: if the delay parameter is present and invalid. """ - delay = parse_integer(request, "org.matrix.msc4140.delay") - if delay is None: - return None - if max_delay is None: - raise SynapseError( - HTTPStatus.BAD_REQUEST, - "Delayed events are not supported on this server", - Codes.UNKNOWN, - { - "org.matrix.msc4140.errcode": "M_MAX_DELAY_UNSUPPORTED", - }, - ) - if delay > max_delay: - raise SynapseError( - HTTPStatus.BAD_REQUEST, - "The requested delay exceeds the allowed maximum.", - Codes.UNKNOWN, - { - "org.matrix.msc4140.errcode": "M_MAX_DELAY_EXCEEDED", - "org.matrix.msc4140.max_delay": max_delay, - }, - ) - return delay + delay_ms = parse_integer(request, "org.matrix.msc4140.delay") + return Duration(milliseconds=delay_ms) if delay_ms is not None else None # TODO: Needs unit testing for room ID + alias joins diff --git a/synapse/storage/databases/main/delayed_events.py b/synapse/storage/databases/main/delayed_events.py index 1727f589e2a..bb512611e4f 100644 --- a/synapse/storage/databases/main/delayed_events.py +++ b/synapse/storage/databases/main/delayed_events.py @@ -17,7 +17,7 @@ import attr -from synapse.api.errors import NotFoundError +from synapse.api.errors import LimitExceededError, NotFoundError from synapse.storage._base import SQLBaseStore, db_to_json from synapse.storage.database import ( DatabasePool, @@ -28,6 +28,7 @@ from synapse.storage.engines import PostgresEngine from synapse.types import JsonDict, RoomID from synapse.util import stringutils +from synapse.util.duration import Duration from synapse.util.json import json_encoder if TYPE_CHECKING: @@ -122,20 +123,84 @@ async def add_delayed_event( state_key: str | None, origin_server_ts: int | None, content: JsonDict, - delay: int, + delay: Duration, sticky_duration_ms: int | None, + limit: int, ) -> tuple[DelayID, Timestamp]: """ Inserts a new delayed event in the DB. + Args: + user_localpart: The localpart of the requester of the delayed event, who will be its owner. + device_id: The device ID of the requester. + creation_ts: The timestamp of when the request to add the delayed event was made. + room_id: The ID of the room where the event should be sent to. + event_type: The type of event to be sent. + state_key: The state key of the event to be sent, or None if it is not a state event. + origin_server_ts: The custom timestamp to send the event with. + If None, the timestamp will be the actual time when the event is sent. + content: The content of the event to be sent. + delay: How long to wait before automatically sending the event. + sticky_duration_ms: If an MSC4354 sticky event: the sticky duration (in milliseconds). + The event will be attempted to be reliably delivered to clients and remote servers + during its sticky period. + limit: The maximum number of delayed events the DB may store for the given requester. + Must be greater than 0. Returns: The generated ID assigned to the added delayed event, and the send time of the next delayed event to be sent, which is either the event just added or one added earlier. + + Raises: + LimitExceededError: if the DB has reached the limit of + how many delayed events it may store for the given requester. + AssertionError: if the limit is not greater than 0. """ + assert limit > 0, "limit must be greater than 0" + delay_id = _generate_delay_id() - send_ts = Timestamp(creation_ts + delay) + delay_ms = delay.as_millis() + send_ts = creation_ts + delay_ms def add_delayed_event_txn(txn: LoggingTransaction) -> Timestamp: + num_existing: int = self.db_pool.simple_select_one_onecol_txn( + txn, + table="delayed_events", + keyvalues={"user_localpart": user_localpart}, + retcol="COUNT(*)", + ) + if num_existing >= limit: + # Find the send_ts threshold that will bring the queue back under the limit. + # When the amount of existing delayed events has reached the limit, + # this will be the send time of the next delayed event to be sent. + # When the amount has exceeded the limit (e.g., due to config changes), + # this will be the send time of the delayed event that will be sent + # once all earlier events that exceed the limit have been sent. + # + # FIXME: Remove "AS subquery" after dropping support for PostgreSQL <16 + txn.execute( + """ + SELECT MAX(send_ts) FROM ( + SELECT * FROM delayed_events + WHERE user_localpart = ? + ORDER BY send_ts ASC + LIMIT ? + ) AS subquery + """, + ( + user_localpart, + num_existing - limit + 1, + ), + ) + row = txn.fetchone() + assert row + retry_after_ms = row[0] - self.clock.time_msec() + err = LimitExceededError( + limiter_name="add_delayed_event", + retry_after_ms=retry_after_ms if retry_after_ms > 0 else None, + ) + err.msg = "The maximum number of delayed events has been reached." + raise err + self.db_pool.simple_insert_txn( txn, table="delayed_events", @@ -143,7 +208,7 @@ def add_delayed_event_txn(txn: LoggingTransaction) -> Timestamp: "delay_id": delay_id, "user_localpart": user_localpart, "device_id": device_id, - "delay": delay, + "delay": delay_ms, "send_ts": send_ts, "room_id": room_id, "event_type": event_type, diff --git a/synapse/storage/schema/main/delta/88/01_add_delayed_events.sql b/synapse/storage/schema/main/delta/88/01_add_delayed_events.sql index 78ba5129af5..4abe0ccaf44 100644 --- a/synapse/storage/schema/main/delta/88/01_add_delayed_events.sql +++ b/synapse/storage/schema/main/delta/88/01_add_delayed_events.sql @@ -22,6 +22,8 @@ CREATE TABLE delayed_events ( state_key TEXT, origin_server_ts BIGINT, content bytea NOT NULL, + -- is_processed = TRUE means that the work of sending the delayed event has begun. + -- Once the send is complete, the delayed event is removed from this table. is_processed BOOLEAN NOT NULL DEFAULT FALSE, PRIMARY KEY (user_localpart, delay_id) ); diff --git a/tests/config/test_server.py b/tests/config/test_server.py index d3c59ae14ca..41ea8fb5b1b 100644 --- a/tests/config/test_server.py +++ b/tests/config/test_server.py @@ -18,11 +18,15 @@ # # + +from typing import Any + import yaml from synapse.config._base import ConfigError, RootConfig from synapse.config.homeserver import HomeServerConfig from synapse.config.server import ServerConfig, generate_ip_set, is_threepid_reserved +from synapse.types import JsonDict from tests import unittest @@ -189,6 +193,54 @@ def test_listeners_set_correctly_open_private_ports_true(self) -> None: self.assertEqual(conf["listeners"], expected_listeners) + def test_max_delayed_events_enforces_positive(self) -> None: + """ + Test that the configured maximum allowed delay must be a positive value if set, + as per documentation + """ + + def generate_config(value: int) -> JsonDict: + return {"max_event_delay_duration": value} + + _read_config(generate_config(1)) + + with self.assertRaises(ConfigError): + _read_config(generate_config(0)) + + with self.assertRaises(ConfigError): + _read_config(generate_config(-1)) + + def test_max_delayed_events_per_user_enforces_non_negative_int(self) -> None: + """ + Test that the configured maximum number of delayed events must be a non-negative value if set, + as a negative limit can never be satisfied + """ + + def generate_config(value: Any) -> JsonDict: + return { + "experimental_features": {"msc4140_max_delayed_events_per_user": value} + } + + for allowed_value in (0, 1): + _read_config(generate_config(allowed_value)) + + for disallowed_value in (-1, 0.5): + with self.assertRaises(ConfigError): + _read_config(generate_config(disallowed_value)) + + +def _read_config(config_values: JsonDict) -> None: + ServerConfig(RootConfig()).read_config( + yaml.safe_load( + HomeServerConfig().generate_config( + config_dir_path="CONFDIR", + data_dir_path="/data_dir_path", + server_name="che.org", + ) + ) + | config_values + ) + class GenerateIpSetTestCase(unittest.TestCase): def test_empty(self) -> None: diff --git a/tests/rest/client/test_capabilities.py b/tests/rest/client/test_capabilities.py index c28e0605b54..42926c43593 100644 --- a/tests/rest/client/test_capabilities.py +++ b/tests/rest/client/test_capabilities.py @@ -26,6 +26,7 @@ from synapse.rest.client import capabilities, login from synapse.server import HomeServer from synapse.util.clock import Clock +from synapse.util.duration import Duration from tests import unittest from tests.unittest import override_config, skip_unless @@ -203,6 +204,43 @@ def test_get_set_avatar_url_capabilities_avatar_url_disabled_msc4133(self) -> No ["avatar_url"], ) + def test_get_delayed_events_capabilities_default_config_msc4140(self) -> None: + access_token = self.login(self.localpart, self.password) + + channel = self.make_request("GET", self.url, access_token=access_token) + capabilities = channel.json_body["capabilities"] + + self.assertEqual(channel.code, HTTPStatus.OK) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_delay_ms"], 0 + ) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_scheduled"], 100 + ) + + @override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 50, + }, + } + ) + def test_get_delayed_events_capabilities_custom_config_msc4140(self) -> None: + access_token = self.login(self.localpart, self.password) + + channel = self.make_request("GET", self.url, access_token=access_token) + capabilities = channel.json_body["capabilities"] + + self.assertEqual(channel.code, HTTPStatus.OK) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_delay_ms"], + Duration(days=1).as_millis(), + ) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_scheduled"], 50 + ) + @override_config({"enable_3pid_changes": False}) def test_get_change_3pid_capabilities_3pid_disabled(self) -> None: """Test if change 3pid is disabled that the server responds it.""" diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py index 8d4892ae91f..90063ccb062 100644 --- a/tests/rest/client/test_rooms.py +++ b/tests/rest/client/test_rooms.py @@ -61,6 +61,7 @@ from synapse.server import HomeServer from synapse.types import JsonDict, JsonMapping, RoomAlias, UserID, create_requester from synapse.util.clock import Clock +from synapse.util.duration import Duration from synapse.util.stringutils import random_string from tests import unittest @@ -2503,7 +2504,12 @@ def test_send_delayed_invalid_event(self) -> None: {}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) - self.assertNotIn("org.matrix.msc4140.errcode", channel.json_body) + # Assert that the standard error response uses a valid errcode. + # The specific errcode is irrelevant for the purpose of this test. + self.assertIsInstance( + channel.json_body.get("errcode"), + str, + ) def test_delayed_event_unsupported_by_default(self) -> None: """Test that sending a delayed event is unsupported with the default config.""" @@ -2515,10 +2521,35 @@ def test_delayed_event_unsupported_by_default(self) -> None: ).encode("ascii"), {"body": "test", "msgtype": "m.text"}, ) - self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) + self.assertEqual( + Codes.FORBIDDEN, + channel.json_body.get("errcode"), + channel.json_body, + ) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 0, + }, + } + ) + def test_delayed_event_disabled_by_limit(self) -> None: + """Test that delayed events are disabled by configuring the per-user limit to 0.""" + channel = self.make_request( + "PUT", + ( + "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000" + % self.room_id + ).encode("ascii"), + {"body": "test", "msgtype": "m.text"}, + ) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) self.assertEqual( - "M_MAX_DELAY_UNSUPPORTED", - channel.json_body.get("org.matrix.msc4140.errcode"), + Codes.FORBIDDEN, + channel.json_body.get("errcode"), channel.json_body, ) @@ -2533,12 +2564,177 @@ def test_delayed_event_exceeds_max_delay(self) -> None: ).encode("ascii"), {"body": "test", "msgtype": "m.text"}, ) - self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) + self.assertEqual( + Codes.FORBIDDEN, + channel.json_body.get("errcode"), + channel.json_body, + ) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 1, + }, + } + ) + def test_delayed_event_user_limit_reached(self) -> None: + """Test that users cannot have more delayed events scheduled at once than allowed.""" + # Disable rate-limits for this user. We want to specifically test the storage-based limit, not the request limits + self.get_success( + self.hs.get_datastores().main.set_ratelimit_for_user(self.user_id, 0, 0) + ) + + make_delayed_event_request = lambda: self.make_request( + "POST", + ( + "rooms/%s/send/m.room.message?org.matrix.msc4140.delay=15000" + % self.room_id + ).encode("ascii"), + {"body": "test", "msgtype": "m.text"}, + ) + # Send a delayed event to eat up the limit + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + # Try to send another delayed event (we expect to hit the limit on the max number of delayed events that can be scheduled at once) + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) self.assertEqual( - "M_MAX_DELAY_EXCEEDED", - channel.json_body.get("org.matrix.msc4140.errcode"), + Codes.LIMIT_EXCEEDED, + channel.json_body["errcode"], channel.json_body, ) + # Confirm that the response includes the time remaining until the next of the user's + # delayed events to be sent, at which point another delayed event may be scheduled + # without exceeding the limit + retry_after_headers = channel.headers.getRawHeaders("Retry-After") + assert retry_after_headers + retry_after_sec = int(retry_after_headers[0]) + self.assertGreater(retry_after_sec, 0) + # Confirm that there is only a single value to the Retry-After header, as per RFC9110 + self.assertEqual(1, len(retry_after_headers)) + + # Wait until we're able to retry again (the retry time from the error response) + self.reactor.advance(retry_after_sec) + + # We should be able to send another delayed event again + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 1, + }, + } + ) + def test_delayed_event_processed_user_limit_reached(self) -> None: + """ + Test that delayed events in the midst of being sent still count towards the limit of + how many delayed events a user may have scheduled at once. + """ + send_after = Duration(seconds=1) + make_delayed_event_request = lambda: self.make_request( + "POST", + ( + f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after.as_millis()}" + % self.room_id + ).encode("ascii"), + {"body": "test", "msgtype": "m.text"}, + ) + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + # Simulate the server taking a long time to persist delayed events + simulated_send_lag = Duration(seconds=5) + event_creation_handler = self.hs.get_event_creation_handler() + orig_send_fn = event_creation_handler.create_and_send_nonmember_event + + async def slow_send_fn(*args: Any, **kwargs: Any) -> Any: + await self.clock.sleep(simulated_send_lag) + return await orig_send_fn(*args, **kwargs) + + with patch.object(event_creation_handler, orig_send_fn.__name__, slow_send_fn): + self.reactor.advance(send_after.as_secs()) + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) + self.assertEqual( + Codes.LIMIT_EXCEEDED, + channel.json_body["errcode"], + channel.json_body, + ) + # Confirm that the response lacks a Retry-After header, because the reason for this limit + # is the server taking an indeterminitely long time to process a delayed event, and the + # server doesn't know how much longer the client should wait before sending more requests + retry_after_headers = channel.headers.getRawHeaders("Retry-After") + assert not retry_after_headers + + # Wait until the delayed event gets persisted + self.reactor.advance(simulated_send_lag.as_secs()) + + # We should be able to send another delayed event again + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 5, + }, + } + ) + def test_delayed_event_user_limit_exceeded(self) -> None: + """ + Test that delayed event limits work properly when + the number of already scheduled events exceeds the configured limit. + + This can be invoked by the server admin lowering the configured limit & restarting the server + while a user has fewer scheduled delayed events than the old limit, but more than the new limit. + """ + send_after: Duration + make_delayed_event_request = lambda: self.make_request( + "POST", + ( + f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after.as_millis()}" + % self.room_id + ).encode("ascii"), + {"body": f"test (send after {send_after.as_secs()}s)", "msgtype": "m.text"}, + ) + + for i in range(4): + send_after = Duration(seconds=i) + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + # Simulate restarting the server after having reconfigured the limit + # to be lower than the number of delayed events we just scheduled. + # + # Set the limit > 1 to test not having to wait for _all_ delayed events + # to be sent before being able to schedule a new one. + self.hs.config.server.max_delayed_events_per_user = 2 + + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) + self.assertEqual( + Codes.LIMIT_EXCEEDED, + channel.json_body["errcode"], + channel.json_body, + ) + retry_after_header = channel.headers.getRawHeaders("Retry-After") + assert retry_after_header + retry_after_sec = int(retry_after_header[0]) + assert retry_after_sec > 0 + + # Wait until we're able to retry again (the retry time from the error response) + self.reactor.advance(retry_after_sec) + + # We should be able to send another delayed event again + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @unittest.override_config({"max_event_delay_duration": "24h"}) def test_delayed_event_with_negative_delay(self) -> None: @@ -2595,7 +2791,7 @@ def test_add_delayed_event_ratelimit(self) -> None: """ # Test that new delayed events are correctly ratelimited. - args = ( + make_delayed_event_request = lambda: self.make_request( "POST", ( "rooms/%s/send/m.room.message?org.matrix.msc4140.delay=2000" @@ -2603,9 +2799,9 @@ def test_add_delayed_event_ratelimit(self) -> None: ).encode("ascii"), {"body": "test", "msgtype": "m.text"}, ) - channel = self.make_request(*args) + channel = make_delayed_event_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - channel = self.make_request(*args) + channel = make_delayed_event_request() self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Add the current user to the ratelimit overrides, allowing them no ratelimiting. @@ -2614,7 +2810,7 @@ def test_add_delayed_event_ratelimit(self) -> None: ) # Test that the new delayed events aren't ratelimited anymore. - channel = self.make_request(*args) + channel = make_delayed_event_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result)