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.

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

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:

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,

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

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,28 +123,92 @@ 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",
                 values={
                     "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)
 );