MSC4140: update error responses (#19539)

- Impose limit of scheduled delayed events
- Update error codes to match latest draft of MSC4140 

---------

Co-authored-by: Eric Eastwood <madlittlemods@gmail.com>

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.

rust/src/config/mod.rs

@@ -47,7 +47,7 @@ pub struct AuthConfig {
 }
 #[derive(FromPyObject, Clone)]
 pub struct ServerConfig {
-    pub max_event_delay_ms: Option<u64>,
+    pub msc4140_enabled: bool,
 }
 
 #[derive(FromPyObject, Clone)]

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,

synapse/config/server.py

@@ -37,6 +37,7 @@ from twisted.conch.ssh.keys import Key
 
 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 @@ class ServerConfig(Config):
         # 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)

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 @@ class DelayedEventsHandler:
         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 @@ class DelayedEventsHandler:
             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 @@ class DelayedEventsHandler:
             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:

synapse/rest/client/capabilities.py

@@ -109,6 +109,11 @@ class CapabilitiesRestServlet(RestServlet):
                 "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,

synapse/rest/client/room.py

@@ -84,6 +84,7 @@ from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID
 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 @@ class RoomStateEventRestServlet(RestServlet):
         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 @@ class RoomStateEventRestServlet(RestServlet):
         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 @@ class RoomSendEventRestServlet(TransactionRestServlet):
         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 @@ class RoomSendEventRestServlet(TransactionRestServlet):
         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 @@ class RoomSendEventRestServlet(TransactionRestServlet):
         )
 
 
-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

synapse/storage/databases/main/delayed_events.py

@@ -17,7 +17,7 @@ from typing import TYPE_CHECKING, NewType
 
 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.database import (
 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 @@ class DelayedEventsStore(SQLBaseStore):
         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 @@ class DelayedEventsStore(SQLBaseStore):
                     "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,

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)
 );

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 @@ class ServerConfigTestCase(unittest.TestCase):
 
         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

@@ -26,6 +26,7 @@ from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
 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 @@ class CapabilitiesTestCase(unittest.HomeserverTestCase):
             ["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."""

tests/rest/client/test_rooms.py

@@ -61,6 +61,7 @@ from synapse.rest.client import (
 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 @@ class RoomDelayedEventTestCase(RoomBase):
             {},
         )
         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 @@ class RoomDelayedEventTestCase(RoomBase):
             ).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(
-            "M_MAX_DELAY_UNSUPPORTED",
-            channel.json_body.get("org.matrix.msc4140.errcode"),
+            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(
+            Codes.FORBIDDEN,
+            channel.json_body.get("errcode"),
             channel.json_body,
         )
 
@@ -2533,13 +2564,178 @@ class RoomDelayedEventTestCase(RoomBase):
             ).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(
-            "M_MAX_DELAY_EXCEEDED",
-            channel.json_body.get("org.matrix.msc4140.errcode"),
+            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(
+            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:
         """Test that sending a delayed event fails if its delay is negative."""
@@ -2595,7 +2791,7 @@ class RoomDelayedEventTestCase(RoomBase):
         """
 
         # 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 @@ class RoomDelayedEventTestCase(RoomBase):
             ).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 @@ class RoomDelayedEventTestCase(RoomBase):
         )
 
         # 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)