MSC4140: update error responses

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.

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.

synapse/config/server.py

@@ -914,10 +914,27 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
                 max_event_delay_duration
             )
             if self.max_event_delay_ms <= 0:
-                raise ConfigError("max_event_delay_duration must be a positive value")
+                raise ConfigError(
+                    "'max_event_delay_duration' must be a positive value if set",
+                    ("max_event_delay_duration",),
+                )
         else:
             self.max_event_delay_ms = None
 
+        # 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"),
+            )
+
     def has_tls_listener(self) -> bool:
         return any(listener.is_tls() for listener in self.listeners)
 

synapse/handlers/delayed_events.py

@@ -385,6 +385,7 @@ async def add(
             origin_server_ts=origin_server_ts,
             content=content,
             delay=delay,
+            limit=self._config.server.max_delayed_events_per_user,
             sticky_duration_ms=sticky_duration_ms,
         )
 

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": self.config.server.max_event_delay_ms or 0,
+            "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,

synapse/rest/client/room.py

@@ -539,21 +539,13 @@ def _parse_request_delay(
     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",
-            },
+            "Sending delayed events has been disallowed",
         )
     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,
-            },
+            Codes.INVALID_PARAM,
         )
     return delay
 

synapse/storage/databases/main/delayed_events.py

@@ -13,11 +13,12 @@
 #
 
 import logging
+from http import HTTPStatus
 from typing import TYPE_CHECKING, NewType
 
 import attr
 
-from synapse.api.errors import NotFoundError
+from synapse.api.errors import LimitExceededError, NotFoundError, SynapseError
 from synapse.storage._base import SQLBaseStore, db_to_json
 from synapse.storage.database import (
     DatabasePool,
@@ -123,6 +124,7 @@ async def add_delayed_event(
         origin_server_ts: int | None,
         content: JsonDict,
         delay: int,
+        limit: int,
         sticky_duration_ms: int | None,
     ) -> tuple[DelayID, Timestamp]:
         """
@@ -131,11 +133,44 @@ async def add_delayed_event(
         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 user has reached the limit of
+                how many delayed events they may have scheduled at once.
         """
+        if limit <= 0:
+            raise SynapseError(
+                HTTPStatus.BAD_REQUEST,
+                "Sending delayed events has been disallowed",
+            )
         delay_id = _generate_delay_id()
         send_ts = Timestamp(creation_ts + delay)
 
         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(*)",
+            )
+            next_send_ts: Timestamp | None
+            if num_existing >= limit:
+                next_send_ts = self.db_pool.simple_select_one_onecol_txn(
+                    txn,
+                    table="delayed_events",
+                    keyvalues={
+                        "is_processed": False,
+                        "user_localpart": user_localpart,
+                    },
+                    retcol="MIN(send_ts)",
+                )
+                e = LimitExceededError(
+                    limiter_name="add_delayed_event",
+                    retry_after_ms=next_send_ts - creation_ts,
+                )
+                e.msg = "The maximum number of delayed events has been reached."
+                raise e
+
             self.db_pool.simple_insert_txn(
                 txn,
                 table="delayed_events",

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:

tests/rest/client/test_capabilities.py

@@ -203,6 +203,42 @@ 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"], 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"], 86400000
+        )
+        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."""

tests/rest/client/test_rooms.py

@@ -24,6 +24,7 @@
 """Tests REST events for /rooms paths."""
 
 import json
+import math
 from http import HTTPStatus
 from typing import Any, Iterable, Literal
 from unittest.mock import AsyncMock, Mock, call, patch
@@ -2503,7 +2504,10 @@ 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)
+        self.assertTrue(
+            channel.json_body.get("errcode", "").startswith("M_"),
+            channel.json_body,
+        )
 
     def test_delayed_event_unsupported_by_default(self) -> None:
         """Test that sending a delayed event is unsupported with the default config."""
@@ -2516,11 +2520,26 @@ def test_delayed_event_unsupported_by_default(self) -> None:
             {"body": "test", "msgtype": "m.text"},
         )
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
-        self.assertEqual(
-            "M_MAX_DELAY_UNSUPPORTED",
-            channel.json_body.get("org.matrix.msc4140.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.BAD_REQUEST, channel.code, channel.result)
 
     @unittest.override_config({"max_event_delay_duration": "1000"})
     def test_delayed_event_exceeds_max_delay(self) -> None:
@@ -2535,10 +2554,66 @@ def test_delayed_event_exceeds_max_delay(self) -> None:
         )
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
         self.assertEqual(
-            "M_MAX_DELAY_EXCEEDED",
-            channel.json_body.get("org.matrix.msc4140.errcode"),
+            Codes.INVALID_PARAM,
+            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_exceeded(self) -> None:
+        """Test that users cannot have more delayed events scheduled at once than allowed."""
+        send_after_ms = 15000
+        args = (
+            "POST",
+            (
+                f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after_ms}"
+                % self.room_id
+            ).encode("ascii"),
+            {"body": "test", "msgtype": "m.text"},
+        )
+        channel = self.make_request(*args)
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+
+        channel = self.make_request(*args)
+        self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result)
+        self.assertEqual(
+            Codes.LIMIT_EXCEEDED,
+            channel.json_body["errcode"],
             channel.json_body,
         )
+        step_ms = 100  # This is the amount of time advanced by a call to make_request
+        expected_retry_after_ms = send_after_ms - step_ms
+        self.assertEqual(
+            expected_retry_after_ms,
+            channel.json_body["retry_after_ms"],
+            channel.json_body,
+        )
+        retry_header = channel.headers.getRawHeaders("Retry-After")
+        assert retry_header
+        self.assertSequenceEqual(
+            [str(math.ceil(expected_retry_after_ms / 1000))],
+            retry_header,
+        )
+
+        # Confirm that ratelimit overrides do not unblock this kind of limit
+        self.get_success(
+            self.hs.get_datastores().main.set_ratelimit_for_user(self.user_id, 0, 0)
+        )
+        channel = self.make_request(*args)
+        self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result)
+        self.assertIn("retry_after_ms", channel.json_body)
+        assert channel.headers.getRawHeaders("Retry-After")
+
+        self.reactor.advance(expected_retry_after_ms)
+        channel = self.make_request(*args)
+        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: