MSC4140: update error responses

README.rst

@@ -104,7 +104,7 @@ Copyright and Licensing
   | Copyright 2014–2017 OpenMarket Ltd
   | Copyright 2017 Vector Creations Ltd
   | Copyright 2017–2025 New Vector Ltd
-  | Copyright 2025 Element Creations Ltd
+  | Copyright 2025-2026 Element Creations Ltd
 
 This software is dual-licensed by Element Creations Ltd (Element). It can be
 used either:

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

@@ -2,7 +2,8 @@
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
 # Copyright 2014-2021 The Matrix.org Foundation C.I.C.
-# Copyright (C) 2023-2024 New Vector, Ltd
+# Copyright (C) 2023-2025 New Vector Ltd
+# Copyright (C) 2025-2026 Element Creations Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -914,10 +915,26 @@ 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(
+                    "Expected a positive value", ("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(
+                "Expected a positive value",
+                ("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

@@ -1,7 +1,8 @@
 #
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
-# Copyright (C) 2024 New Vector, Ltd
+# Copyright (C) 2024-2025 New Vector Ltd
+# Copyright (C) 2025-2026 Element Creations Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -385,6 +386,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/room.py

@@ -2,7 +2,8 @@
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright (C) 2023-2024 New Vector, Ltd
+# Copyright (C) 2023-2025 New Vector Ltd
+# Copyright (C) 2025-2026 Element Creations Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -539,19 +540,18 @@ def _parse_request_delay(
         raise SynapseError(
             HTTPStatus.BAD_REQUEST,
             "Delayed events are not supported on this server",
-            Codes.UNKNOWN,
+            "ORG.MATRIX.MSC4140_MAX_DELAY_EXCEEDED",
             {
-                "org.matrix.msc4140.errcode": "M_MAX_DELAY_UNSUPPORTED",
+                "max_delay": 0,
             },
         )
     if delay > max_delay:
         raise SynapseError(
             HTTPStatus.BAD_REQUEST,
             "The requested delay exceeds the allowed maximum.",
-            Codes.UNKNOWN,
+            "ORG.MATRIX.MSC4140_MAX_DELAY_EXCEEDED",
             {
-                "org.matrix.msc4140.errcode": "M_MAX_DELAY_EXCEEDED",
-                "org.matrix.msc4140.max_delay": max_delay,
+                "max_delay": max_delay,
             },
         )
     return delay

synapse/storage/databases/main/delayed_events.py

@@ -1,7 +1,8 @@
 #
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
-# Copyright (C) 2024 New Vector, Ltd
+# Copyright (C) 2024-2025 New Vector Ltd
+# Copyright (C) 2025-2026 Element Creations Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -17,7 +18,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,
@@ -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,39 @@ 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.
         """
+        assert limit > 0  # Should be enforced at config read time
         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(*)",
+            )
+            if num_existing >= limit:
+                next_send_ms: int = 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_ms - 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

@@ -1,7 +1,8 @@
 #
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
-# Copyright (C) 2023 New Vector, Ltd
+# Copyright (C) 2023-2025 New Vector Ltd
+# Copyright (C) 2025-2026 Element Creations Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -18,11 +19,13 @@
 #
 #
 
+
 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 +192,45 @@ 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:
+        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_positive(self) -> None:
+        def generate_config(value: int) -> JsonDict:
+            return {
+                "experimental_features": {"msc4140_max_delayed_events_per_user": 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 _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_rooms.py

@@ -4,7 +4,8 @@
 # Copyright 2019 The Matrix.org Foundation C.I.C.
 # Copyright 2017 Vector Creations Ltd
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright (C) 2023-2024 New Vector, Ltd
+# Copyright (C) 2023-2025 New Vector Ltd
+# Copyright (C) 2025-2026 Element Creations Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -24,6 +25,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
@@ -2419,7 +2421,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."""
@@ -2433,8 +2438,8 @@ def test_delayed_event_unsupported_by_default(self) -> None:
         )
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
         self.assertEqual(
-            "M_MAX_DELAY_UNSUPPORTED",
-            channel.json_body.get("org.matrix.msc4140.errcode"),
+            "ORG.MATRIX.MSC4140_MAX_DELAY_EXCEEDED",
+            channel.json_body.get("errcode"),
             channel.json_body,
         )
 
@@ -2451,11 +2456,69 @@ 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"),
+            "ORG.MATRIX.MSC4140_MAX_DELAY_EXCEEDED",
+            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)
+
+        wait_ms = 2000
+        self.reactor.advance(wait_ms / 1000.0)
+        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  # The simulated duration of each make_request
+        expected_retry_after_ms = send_after_ms - wait_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:
         """Test that sending a delayed event fails if its delay is negative."""