MSC4140: add dedicated endpoint for delayed events (#19354)

Use a new, dedicated endpoint for scheduling delayed events. This should
deprecate the `delay` query parameter on the `/send` and `/state`
endpoints.

Co-authored-by: Paul Chobert <paul@chobert.fr>
Co-authored-by: Eric Eastwood <erice@element.io>

changelog.d/19354.feature

@@ -0,0 +1 @@
+MSC4140: Add a dedicated endpoint for scheduling delayed events.

docker/configure_workers_and_start.py

@@ -203,6 +203,7 @@ WORKERS_CONFIG: dict[str, dict[str, Any]] = {
             "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/device_signing/upload$",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$",
             "^/_matrix/client/unstable/org.matrix.msc4140/delayed_events(/[^/]+(/restart)?)?$",
+            "^/_matrix/client/unstable/org.matrix.msc4140/rooms/.*/delayed_event/",
         ],
         "shared_extra_conf": {},
         "worker_extra_conf": "",

docs/workers.md

@@ -288,7 +288,8 @@ information.
     # User directory search requests
     ^/_matrix/client/(r0|v3|unstable)/user_directory/search$
 
-    # Unstable MSC4140 support
+    # Unstable MSC4140 support (delayed event scheduling & management)
+    ^/_matrix/client/unstable/org.matrix.msc4140/rooms/.*/delayed_event/
     ^/_matrix/client/unstable/org.matrix.msc4140/delayed_events(/[^/]+(/restart)?)?$
 
     # Stabilised Delegated Authentication support (`matrix_authentication_service.enabled: true`)

synapse/api/constants.py

@@ -477,13 +477,16 @@ class StickyEventField(TypedDict):
 
 
 class StickyEvent:
-    QUERY_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms"
+    REQUEST_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms"
     """
-    Query parameter used by clients for setting the sticky duration of an event they are sending.
+    Parameter used by clients for setting the sticky duration of an event they are sending.
 
-    Applies to:
+    Used as a query parameter for:
         - /rooms/.../send/...
         - /rooms/.../state/...
+
+    Used as a request body field for:
+        - /rooms/.../delayed_event/...
     """
 
     EVENT_FIELD_NAME: Final = "msc4354_sticky"

synapse/handlers/delayed_events.py

@@ -68,6 +68,7 @@ class DelayedEventsHandler:
         self._auth = hs.get_auth()
         self._event_creation_handler = hs.get_event_creation_handler()
         self._room_member_handler = hs.get_room_member_handler()
+        self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker
 
         self._request_ratelimiter = hs.get_request_ratelimiter()
 
@@ -369,20 +370,13 @@ class DelayedEventsHandler:
 
         Raises:
             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.
+                if the requested delay is longer than allowed.
         """
         # 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()
@@ -625,6 +619,7 @@ class DelayedEventsHandler:
     ) -> None:
         user_id = UserID(event.user_localpart, self._config.server.server_name)
         user_id_str = user_id.to_string()
+        room_id_str = event.room_id.to_string()
         # Create a new requester from what data is currently available
         requester = create_requester(
             user_id,
@@ -633,13 +628,31 @@ class DelayedEventsHandler:
         )
 
         try:
+            if event.state_key is not None:
+                is_requester_admin = await self._auth.is_server_admin(requester)
+                if not is_requester_admin:
+                    spam_check = await self._spam_checker_module_callbacks.user_may_send_state_event(
+                        user_id=user_id_str,
+                        room_id=room_id_str,
+                        event_type=event.type,
+                        state_key=event.state_key,
+                        content=event.content,
+                    )
+                    if spam_check != self._spam_checker_module_callbacks.NOT_SPAM:
+                        raise SynapseError(
+                            403,
+                            "You are not permitted to send the state event",
+                            errcode=spam_check[0],
+                            additional_fields=spam_check[1],
+                        )
+
             if event.state_key is not None and event.type == EventTypes.Member:
                 membership = event.content.get("membership")
                 assert membership is not None
                 event_id, _ = await self._room_member_handler.update_membership(
                     requester,
                     target=UserID.from_string(event.state_key),
-                    room_id=event.room_id.to_string(),
+                    room_id=room_id_str,
                     action=membership,
                     content=event.content,
                     origin_server_ts=event.origin_server_ts,
@@ -649,7 +662,7 @@ class DelayedEventsHandler:
                 event_dict: JsonDict = {
                     "type": event.type,
                     "content": event.content,
-                    "room_id": event.room_id.to_string(),
+                    "room_id": room_id_str,
                     "sender": user_id_str,
                 }
 

synapse/rest/client/room.py

@@ -25,11 +25,12 @@ import logging
 import re
 from enum import Enum
 from http import HTTPStatus
-from typing import TYPE_CHECKING, Awaitable
+from typing import TYPE_CHECKING, Awaitable, NoReturn
 from urllib import parse as urlparse
 
 import attr
 from prometheus_client.core import Histogram
+from pydantic import Field, PositiveInt, StrictStr
 
 from twisted.web.server import Request
 
@@ -63,6 +64,7 @@ from synapse.http.servlet import (
     ResolveRoomIdMixin,
     RestServlet,
     assert_params_in_dict,
+    parse_and_validate_json_object_from_request,
     parse_boolean,
     parse_enum,
     parse_integer,
@@ -80,7 +82,16 @@ from synapse.rest.client.transactions import HttpTransactionCache
 from synapse.state import CREATE_KEY, POWER_KEY
 from synapse.storage.databases.main import DataStore
 from synapse.streams.config import PaginationConfig
-from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID, UserID
+from synapse.types import (
+    Absent,
+    AbsentType,
+    JsonDict,
+    Requester,
+    StreamToken,
+    ThirdPartyInstanceID,
+    UserID,
+)
+from synapse.types.rest import RequestBodyModel
 from synapse.types.state import StateFilter
 from synapse.util.cancellation import cancellable
 from synapse.util.clock import Clock
@@ -217,6 +228,7 @@ class RoomStateEventRestServlet(RestServlet):
         self.clock = hs.get_clock()
         self._event_serializer = hs.get_event_client_serializer()
         self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker
+        self._msc4140_enabled = hs.config.server.msc4140_enabled
 
     def register(self, http_server: HttpServer) -> None:
         # /rooms/$roomid/state/$eventtype
@@ -315,6 +327,28 @@ class RoomStateEventRestServlet(RestServlet):
 
         content = parse_json_object_from_request(request)
 
+        origin_server_ts = None
+        if requester.app_service_id:
+            origin_server_ts = parse_integer(request, "ts")
+
+        # FIXME(MSC4140): remove by 2026-12-31, once RoomDelayedEventRestServlet has been stable for a suitable amount of time.
+        delay = _parse_request_for_delayed_event_delay(request, self._msc4140_enabled)
+        if delay is not None:
+            delay_id = await self.delayed_events_handler.add(
+                requester,
+                room_id=room_id,
+                event_type=event_type,
+                state_key=state_key,
+                origin_server_ts=origin_server_ts,
+                content=content,
+                delay=delay,
+                sticky_duration_ms=None,
+            )
+
+            set_tag("delay_id", delay_id)
+            ret = {"delay_id": delay_id}
+            return 200, ret
+
         is_requester_admin = await self.auth.is_server_admin(requester)
         if not is_requester_admin:
             spam_check = (
@@ -334,27 +368,6 @@ class RoomStateEventRestServlet(RestServlet):
                     additional_fields=spam_check[1],
                 )
 
-        origin_server_ts = None
-        if requester.app_service_id:
-            origin_server_ts = parse_integer(request, "ts")
-
-        delay = _parse_request_for_delayed_event_delay(request)
-        if delay is not None:
-            delay_id = await self.delayed_events_handler.add(
-                requester,
-                room_id=room_id,
-                event_type=event_type,
-                state_key=state_key,
-                origin_server_ts=origin_server_ts,
-                content=content,
-                delay=delay,
-                sticky_duration_ms=None,
-            )
-
-            set_tag("delay_id", delay_id)
-            ret = {"delay_id": delay_id}
-            return 200, ret
-
         try:
             if event_type == EventTypes.Member:
                 membership = content.get("membership", None)
@@ -407,6 +420,7 @@ 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._msc4140_enabled = hs.config.server.msc4140_enabled
         self._msc4354_enabled = hs.config.experimental.msc4354_enabled
 
     def register(self, http_server: HttpServer) -> None:
@@ -430,9 +444,10 @@ class RoomSendEventRestServlet(TransactionRestServlet):
 
         sticky_duration_ms: int | None = None
         if self._msc4354_enabled:
-            sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME)
+            sticky_duration_ms = parse_integer(request, StickyEvent.REQUEST_PARAM_NAME)
 
-        delay = _parse_request_for_delayed_event_delay(request)
+        # FIXME(MSC4140): remove by 2026-12-31, once RoomDelayedEventRestServlet has been stable for a suitable amount of time.
+        delay = _parse_request_for_delayed_event_delay(request, self._msc4140_enabled)
         if delay is not None:
             delay_id = await self.delayed_events_handler.add(
                 requester,
@@ -505,7 +520,118 @@ class RoomSendEventRestServlet(TransactionRestServlet):
         )
 
 
-def _parse_request_for_delayed_event_delay(request: SynapseRequest) -> Duration | None:
+class RoomDelayedEventRestServlet(TransactionRestServlet):
+    CATEGORY = "Delayed event management requests"
+    PATTERNS = client_patterns(
+        "/org.matrix.msc4140/rooms/(?P<room_id>[^/]*)/delayed_event/(?P<event_type>[^/]*)/(?P<txn_id>[^/]*)$",
+        releases=(),
+        unstable=True,
+    )
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__(hs)
+        self.delayed_events_handler = hs.get_delayed_events_handler()
+        self.auth = hs.get_auth()
+        self._msc4140_enabled = hs.config.server.msc4140_enabled
+        self._msc4354_enabled = hs.config.experimental.msc4354_enabled
+
+    async def on_PUT(
+        self, request: SynapseRequest, room_id: str, event_type: str, txn_id: str
+    ) -> tuple[int, JsonDict]:
+        requester = await self.auth.get_user_by_req(request, allow_guest=True)
+        set_tag("txn_id", txn_id)
+
+        return await self.txns.fetch_or_execute_request(
+            request,
+            requester,
+            self._do,
+            request,
+            requester,
+            room_id,
+            event_type,
+        )
+
+    class DelayedEventBodyModel(RequestBodyModel):
+        delay_ms: PositiveInt
+        content: JsonDict
+        state_key: StrictStr | AbsentType = Absent
+
+    class StickyDelayedEventBodyModel(DelayedEventBodyModel):
+        # MSC4354 allows any integer value in the request, not just positives
+        sticky_duration_ms: int | AbsentType = Field(
+            Absent, alias=StickyEvent.REQUEST_PARAM_NAME
+        )
+
+    async def _do(
+        self,
+        request: SynapseRequest,
+        requester: Requester,
+        room_id: str,
+        event_type: str,
+    ) -> tuple[int, JsonDict]:
+        if not self._msc4140_enabled:
+            _raise_delayed_events_unsupported()
+
+        if not self._msc4354_enabled:
+            # When the MSC is disabled, skip validating the sticky duration field
+            request_body = parse_and_validate_json_object_from_request(
+                request, self.DelayedEventBodyModel
+            )
+            sticky_duration_ms = None
+        else:
+            # Support the parameter in both the query & body until the MSC settles on which one to use
+            request_body = parse_and_validate_json_object_from_request(
+                request, self.StickyDelayedEventBodyModel
+            )
+            sticky_duration_ms = parse_integer(
+                request,
+                StickyEvent.REQUEST_PARAM_NAME,
+                negative=True,  # MSC4354 allows any integer value in the request, not just positives
+            )
+            if request_body.sticky_duration_ms is not Absent:
+                if sticky_duration_ms is None:
+                    sticky_duration_ms = request_body.sticky_duration_ms
+                elif request_body.sticky_duration_ms != sticky_duration_ms:
+                    raise SynapseError(
+                        HTTPStatus.BAD_REQUEST,
+                        f"Conflicting values given for {StickyEvent.REQUEST_PARAM_NAME}",
+                        Codes.INVALID_PARAM,
+                    )
+
+        if request_body.state_key is not Absent:
+            if sticky_duration_ms is not None:
+                raise SynapseError(
+                    HTTPStatus.BAD_REQUEST,
+                    "Sticky state events are not allowed. Cannot specify both `state_key` and `sticky_duration_ms`",
+                    Codes.INVALID_PARAM,
+                )
+            state_key = request_body.state_key
+        else:
+            state_key = None
+
+        origin_server_ts = None
+        if requester.app_service_id:
+            origin_server_ts = parse_integer(request, "ts")
+
+        delay_id = await self.delayed_events_handler.add(
+            requester,
+            room_id=room_id,
+            event_type=event_type,
+            state_key=state_key,
+            origin_server_ts=origin_server_ts,
+            content=request_body.content,
+            delay=Duration(milliseconds=request_body.delay_ms),
+            sticky_duration_ms=sticky_duration_ms,
+        )
+
+        set_tag("delay_id", delay_id)
+        ret = {"delay_id": delay_id}
+        return 200, ret
+
+
+def _parse_request_for_delayed_event_delay(
+    request: SynapseRequest, msc4140_enabled: bool
+) -> Duration | None:
     """Parses from the request string the delay parameter for
         delayed event requests, and checks it for correctness.
 
@@ -517,8 +643,29 @@ def _parse_request_for_delayed_event_delay(request: SynapseRequest) -> Duration
     Raises:
         SynapseError: if the delay parameter is present and invalid.
     """
-    delay_ms = parse_integer(request, "org.matrix.msc4140.delay")
-    return Duration(milliseconds=delay_ms) if delay_ms is not None else None
+    delay_param_name = "org.matrix.msc4140.delay"
+    # Allow negatives here to validate the delay only if delayed events are enabled,
+    # and so that any non-positive value is rejected with the same error
+    delay_ms = parse_integer(request, delay_param_name, negative=True)
+    if delay_ms is None:
+        return None
+    if not msc4140_enabled:
+        _raise_delayed_events_unsupported()
+    if delay_ms <= 0:
+        raise SynapseError(
+            HTTPStatus.BAD_REQUEST,
+            f"Query parameter {delay_param_name} must be an integer greater than zero.",
+            Codes.INVALID_PARAM,
+        )
+    return Duration(milliseconds=delay_ms)
+
+
+def _raise_delayed_events_unsupported() -> NoReturn:
+    raise SynapseError(
+        HTTPStatus.FORBIDDEN,
+        "Sending delayed events is disabled on this homeserver",
+        Codes.FORBIDDEN,
+    )
 
 
 # TODO: Needs unit testing for room ID + alias joins
@@ -1750,6 +1897,7 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     SearchRestServlet(hs).register(http_server)
     RoomCreateRestServlet(hs).register(http_server)
     TimestampLookupRestServlet(hs).register(http_server)
+    RoomDelayedEventRestServlet(hs).register(http_server)
 
     # Some servlets only get registered for the main process.
     if hs.config.worker.worker_app is None:

tests/module_api/test_spamchecker.py

@@ -15,6 +15,8 @@
 from http import HTTPStatus
 from typing import Literal
 
+from parameterized import parameterized
+
 from twisted.internet.testing import MemoryReactor
 
 from synapse.api.constants import (
@@ -26,10 +28,11 @@ from synapse.api.room_versions import RoomVersions
 from synapse.config.server import DEFAULT_ROOM_VERSION
 from synapse.events import make_event_from_dict
 from synapse.module_api import EventBase
-from synapse.rest import admin, login, room, room_upgrade_rest_servlet
+from synapse.rest import admin, delayed_events, login, room, room_upgrade_rest_servlet
 from synapse.server import HomeServer
 from synapse.types import Codes, JsonDict
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests import unittest
 from tests.server import FakeChannel
@@ -42,6 +45,7 @@ class SpamCheckerTestCase(HomeserverTestCase):
     servlets = [
         room.register_servlets,
         admin.register_servlets,
+        delayed_events.register_servlets,
         login.register_servlets,
         room_upgrade_rest_servlet.register_servlets,
     ]
@@ -296,6 +300,144 @@ class SpamCheckerTestCase(HomeserverTestCase):
         self.assertEqual(channel.code, 403)
         self.assertEqual(channel.json_body["errcode"], Codes.FORBIDDEN)
 
+    @parameterized.expand((False, True))
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_user_may_send_state_event_with_delay(self, send_manually: bool) -> None:
+        """Test that the user_may_send_state_event callback is called when a delayed state event
+        is sent (not when it is scheduled), and that it receives the correct parameters.
+        """
+
+        async def user_may_send_state_event(
+            user_id: str,
+            room_id: str,
+            event_type: str,
+            state_key: str,
+            content: JsonDict,
+        ) -> Literal["NOT_SPAM"] | Codes:
+            self.did_spam_check = True
+            self.last_user_id = user_id
+            self.last_room_id = room_id
+            self.last_event_type = event_type
+            self.last_state_key = state_key
+            self.last_content = content
+            return "NOT_SPAM"
+
+        self._module_api.register_spam_checker_callbacks(
+            user_may_send_state_event=user_may_send_state_event
+        )
+        self.did_spam_check = False
+
+        channel = self.create_room({})
+
+        self.assertEqual(channel.code, 200)
+
+        room_id = channel.json_body["room_id"]
+
+        event_type = "test.event.type"
+        state_key = "test.state.key"
+        delay = Duration(seconds=10)
+        channel = self.make_request(
+            "PUT",
+            f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/mid",
+            content={
+                "delay_ms": delay.as_millis(),
+                "state_key": state_key,
+                "content": {"foo": "bar"},
+            },
+            access_token=self.token,
+        )
+
+        self.assertEqual(channel.code, 200)
+        self.assertFalse(self.did_spam_check)
+
+        delay_id = channel.json_body["delay_id"]
+
+        if send_manually:
+            channel = self.make_request(
+                "POST",
+                f"/_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/send",
+                access_token=self.token,
+            )
+        else:
+            # Advance time enough so the delayed event is sent
+            self.reactor.advance(delay.as_secs())
+
+        self.assertEqual(channel.code, 200)
+        self.assertTrue(self.did_spam_check)
+        self.assertEqual(self.last_user_id, self.user_id)
+        self.assertEqual(self.last_room_id, room_id)
+        self.assertEqual(self.last_event_type, event_type)
+        self.assertEqual(self.last_state_key, state_key)
+        self.assertEqual(self.last_content, {"foo": "bar"})
+
+    @parameterized.expand((False, True))
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_user_may_send_state_event_disallows_with_delay(
+        self, send_manually: bool
+    ) -> None:
+        """Test that the user_may_send_state_event callback is called when a delayed state event
+        is sent (not when it is scheduled), and that the response is honoured.
+        """
+
+        async def user_may_send_state_event(
+            user_id: str,
+            room_id: str,
+            event_type: str,
+            state_key: str,
+            content: JsonDict,
+        ) -> Literal["NOT_SPAM"] | Codes:
+            return Codes.FORBIDDEN
+
+        self._module_api.register_spam_checker_callbacks(
+            user_may_send_state_event=user_may_send_state_event
+        )
+        self.did_spam_check = False
+
+        channel = self.create_room({})
+
+        self.assertEqual(channel.code, 200)
+
+        room_id = channel.json_body["room_id"]
+
+        event_type = "test.event.type"
+        state_key = "test.state.key"
+        delay = Duration(seconds=10)
+        channel = self.make_request(
+            "PUT",
+            f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/mid",
+            content={
+                "delay_ms": delay.as_millis(),
+                "state_key": state_key,
+                "content": {"foo": "bar"},
+            },
+            access_token=self.token,
+        )
+
+        self.assertEqual(channel.code, 200)
+        self.assertFalse(self.did_spam_check)
+
+        delay_id = channel.json_body["delay_id"]
+
+        if send_manually:
+            channel = self.make_request(
+                "POST",
+                f"/_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/send",
+                access_token=self.token,
+            )
+            self.assertEqual(channel.code, 403)
+            self.assertEqual(channel.json_body["errcode"], Codes.FORBIDDEN)
+        else:
+            # Advance time enough so the delayed event is sent
+            self.reactor.advance(delay.as_secs())
+
+        # The state should not have been set by the delayed event
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/r0/rooms/{room_id}/state/{event_type}/{state_key}",
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 404)
+
 
 class FederatedEventSpamCheckMetadataTestCase(unittest.FederatingHomeserverTestCase):
     servlets = [

tests/rest/client/test_rooms.py

@@ -24,6 +24,7 @@
 """Tests REST events for /rooms paths."""
 
 import json
+import time
 from http import HTTPStatus
 from typing import Any, Iterable, Literal
 from unittest.mock import AsyncMock, Mock, call, create_autospec, patch
@@ -67,6 +68,7 @@ from synapse.util.stringutils import random_string
 
 from tests import unittest
 from tests.http.server._base import make_request_with_cancellation_test
+from tests.server import FakeChannel
 from tests.storage.test_stream import PaginationTestCase
 from tests.test_utils import FakeResponse
 from tests.test_utils.event_injection import (
@@ -1589,6 +1591,37 @@ class RoomAppserviceTsParamTestCase(unittest.HomeserverTestCase):
         res = self.get_success(self.main_store.get_event(event_id))
         self.assertEqual(ts, res.origin_server_ts)
 
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_send_delayed_event_ts(self) -> None:
+        """Test scheduling a delayed event with a custom timestamp."""
+        ts = 1
+
+        url_params = {
+            "user_id": self.appservice_user,
+            "ts": ts,
+        }
+        channel = self.make_request(
+            "PUT",
+            path=f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{self.room}/delayed_event/m.room.message/1234?"
+            + urlparse.urlencode(url_params),
+            content={
+                "delay_ms": 2000,
+                "content": {"body": "test", "msgtype": "m.text"},
+            },
+            access_token=self.appservice.token,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Wait for the delayed event to be sent, then ensure it was persisted
+        # with the custom timestamp.
+        self.reactor.advance(Duration(seconds=2).as_secs())
+        (event_id,) = self.get_success(
+            self.main_store.get_latest_event_ids_in_room(self.room)
+        )
+        res = self.get_success(self.main_store.get_event(event_id))
+        self.assertEqual("m.room.message", res.type, res)
+        self.assertEqual(ts, res.origin_server_ts)
+
     def test_send_membership_event_ts(self) -> None:
         """Test sending a membership event with a custom timestamp."""
         ts = 1
@@ -2494,6 +2527,42 @@ class RoomDelayedEventTestCase(RoomBase):
     """Tests delayed events."""
 
     user_id = "@sid1:red"
+    invalid_delay_error_type = Codes.INVALID_PARAM
+
+    def make_delayed_event_request(
+        self,
+        *,
+        room_id: str,
+        delay: Duration,
+        event_type: str,
+        state_key: str | None = None,
+        content: JsonDict,
+        txn_id: str | None = None,
+    ) -> FakeChannel:
+        """
+        Build and send a request for scheduling a delayed event.
+
+        Args:
+            room_id: The room to send the event to.
+            delay: How long to wait before sending the event.
+            event_type: The type of the event to send.
+            state_key: The state key of the event, or None for a message event.
+            content: The content of the event.
+            txn_id: The transaction ID for a delayed message event request.
+                Ignored for a delayed state event request.
+                Otherwise, generated if not given.
+
+        Returns:
+            The `FakeChannel` object which stores the result of the request.
+        """
+        if state_key is not None:
+            path = f"rooms/{room_id}/state/{event_type}/{state_key}"
+        else:
+            if txn_id is None:
+                txn_id = "m%s" % (str(time.time()))
+            path = f"rooms/{room_id}/send/{event_type}/{txn_id}"
+        path += f"?org.matrix.msc4140.delay={delay.as_millis()}"
+        return self.make_request("PUT", path, content)
 
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.room_id = self.helper.create_room_as(self.user_id)
@@ -2501,13 +2570,11 @@ class RoomDelayedEventTestCase(RoomBase):
     @unittest.override_config({"max_event_delay_duration": "24h"})
     def test_send_delayed_invalid_event(self) -> None:
         """Test sending a delayed event with invalid content."""
-        channel = self.make_request(
-            "PUT",
-            (
-                "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000"
-                % self.room_id
-            ).encode("ascii"),
-            {},
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={},
         )
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
         # Assert that the standard error response uses a valid errcode.
@@ -2519,13 +2586,11 @@ class RoomDelayedEventTestCase(RoomBase):
 
     def test_delayed_event_unsupported_by_default(self) -> None:
         """Test that sending a delayed event is unsupported with the default config."""
-        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"},
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result)
         self.assertEqual(
@@ -2544,13 +2609,11 @@ class RoomDelayedEventTestCase(RoomBase):
     )
     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"},
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result)
         self.assertEqual(
@@ -2562,17 +2625,15 @@ class RoomDelayedEventTestCase(RoomBase):
     @unittest.override_config({"max_event_delay_duration": "1000"})
     def test_delayed_event_exceeds_max_delay(self) -> None:
         """Test that sending a delayed event fails if its delay is longer than allowed."""
-        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"},
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
         self.assertEqual(
-            "ORG.MATRIX.MSC4140_DELAY_TOO_LARGE",
+            Codes.DELAY_TOO_LARGE,
             channel.json_body.get("errcode"),
             channel.json_body,
         )
@@ -2592,13 +2653,11 @@ class RoomDelayedEventTestCase(RoomBase):
             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"},
+        make_delayed_event_request = lambda: self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=15000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         # Send a delayed event to eat up the limit
         channel = make_delayed_event_request()
@@ -2643,13 +2702,11 @@ class RoomDelayedEventTestCase(RoomBase):
         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"},
+        make_delayed_event_request = lambda: self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=send_after,
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         channel = make_delayed_event_request()
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
@@ -2702,16 +2759,18 @@ class RoomDelayedEventTestCase(RoomBase):
         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"},
+
+        make_delayed_event_request = lambda: self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=send_after,
+            event_type="m.room.message",
+            content={
+                "body": f"test (send after {send_after.as_secs()}s)",
+                "msgtype": "m.text",
+            },
         )
 
-        for i in range(4):
+        for i in range(1, 5):
             send_after = Duration(seconds=i)
             channel = make_delayed_event_request()
             self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
@@ -2742,45 +2801,100 @@ class RoomDelayedEventTestCase(RoomBase):
         channel = make_delayed_event_request()
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
 
+    @parameterized.expand(
+        (
+            # A negative delay would put the send time in the past.
+            Duration(milliseconds=-2000),
+            # A zero delay is no delay at all; the regular send endpoints are for that.
+            Duration(),
+        )
+    )
     @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."""
-        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"},
+    def test_delayed_event_with_invalid_delay(self, invalid_delay: Duration) -> None:
+        """Test that sending a delayed event fails if its delay is not positive."""
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=invalid_delay,
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
         self.assertEqual(
-            Codes.INVALID_PARAM, channel.json_body["errcode"], channel.json_body
+            self.invalid_delay_error_type,
+            channel.json_body["errcode"],
+            channel.json_body,
+        )
+
+    @parameterized.expand(
+        (
+            # A negative delay would put the send time in the past.
+            Duration(milliseconds=-2000),
+            # A zero delay is no delay at all; the regular send endpoints are for that.
+            Duration(),
+        )
+    )
+    @unittest.override_config({"max_event_delay_duration": None})
+    def test_delayed_event_unsupported_with_invalid_delay(
+        self,
+        invalid_delay: Duration,
+    ) -> None:
+        """Test that delayed events being unsupported takes precedence over an invalid delay."""
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=invalid_delay,
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
+        )
+        self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result)
+        self.assertEqual(
+            Codes.FORBIDDEN,
+            channel.json_body["errcode"],
+            channel.json_body,
         )
 
     @unittest.override_config({"max_event_delay_duration": "24h"})
     def test_send_delayed_message_event(self) -> None:
         """Test sending a valid delayed message event."""
-        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"},
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
 
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_send_delayed_message_event_idempotent(self) -> None:
+        """
+        Test that repeated requests to schedule a delayed message event
+        with the same transaction ID are idempotent.
+        """
+        make_delayed_event_request = lambda: self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
+            txn_id="txn_id_0",
+        )
+        channel = make_delayed_event_request()
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+
+        delay_id = channel.json_body["delay_id"]
+
+        # Repeat the request, which reuses its txn_id
+        channel = make_delayed_event_request()
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        self.assertEqual(delay_id, channel.json_body.get("delay_id"))
+
     @unittest.override_config({"max_event_delay_duration": "24h"})
     def test_send_delayed_state_event(self) -> None:
         """Test sending a valid delayed state event."""
-        channel = self.make_request(
-            "PUT",
-            (
-                "rooms/%s/state/m.room.topic/?org.matrix.msc4140.delay=2000"
-                % self.room_id
-            ).encode("ascii"),
-            {"topic": "This is a topic"},
+        channel = self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.topic",
+            state_key="",
+            content={"topic": "This is a topic"},
         )
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
 
@@ -2796,15 +2910,13 @@ class RoomDelayedEventTestCase(RoomBase):
         exempt from ratelimiting.
         """
 
-        # Test that new delayed events are correctly ratelimited.
-        make_delayed_event_request = lambda: self.make_request(
-            "POST",
-            (
-                "rooms/%s/send/m.room.message?org.matrix.msc4140.delay=2000"
-                % self.room_id
-            ).encode("ascii"),
-            {"body": "test", "msgtype": "m.text"},
+        make_delayed_event_request = lambda: self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.message",
+            content={"body": "test", "msgtype": "m.text"},
         )
+        # Test that new delayed events are correctly ratelimited.
         channel = make_delayed_event_request()
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
         channel = make_delayed_event_request()
@@ -2820,6 +2932,152 @@ class RoomDelayedEventTestCase(RoomBase):
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
 
 
+class RoomDelayedEventDedicatedEndpointTestCase(RoomDelayedEventTestCase):
+    """Tests the dedicated request endpoint for scheduling delayed events."""
+
+    invalid_delay_error_type = Codes.BAD_JSON
+
+    def make_delayed_event_request(
+        self,
+        *,
+        room_id: str,
+        delay: Duration,
+        event_type: str,
+        state_key: str | None = None,
+        content: JsonDict,
+        txn_id: str | None = None,
+    ) -> FakeChannel:
+        return self.make_request(
+            "PUT",
+            self.helper.build_delayed_event_request_path(
+                room_id=room_id,
+                event_type=event_type,
+                txn_id=txn_id,
+            ),
+            self.helper.build_delayed_event_request_body(
+                delay=delay,
+                state_key=state_key,
+                content=content,
+            ),
+        )
+
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_delayed_event_with_content_at_top(self) -> None:
+        """Test that the dedicated endpoint fails with event content keys at the top level of the body."""
+        channel = self.make_request(
+            "PUT",
+            self.helper.build_delayed_event_request_path(
+                room_id=self.room_id,
+                event_type="m.room.message",
+            ),
+            content={
+                "delay_ms": 2000,
+                "body": "test",
+                "msgtype": "m.text",
+            },
+        )
+        self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
+        self.assertEqual(
+            Codes.MISSING_PARAM,
+            channel.json_body["errcode"],
+            channel.json_body,
+        )
+
+    @parameterized.expand(("delay_ms", "content"))
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_delayed_event_with_missing_key(self, missing_key: str) -> None:
+        """Test that the dedicated endpoint fails with a body missing a required key."""
+        body = self.helper.build_delayed_event_request_body(
+            delay=Duration(milliseconds=2000),
+            content={"body": "test", "msgtype": "m.text"},
+        )
+        del body[missing_key]
+        channel = self.make_request(
+            "PUT",
+            self.helper.build_delayed_event_request_path(
+                room_id=self.room_id,
+                event_type="m.room.message",
+            ),
+            content=body,
+        )
+        self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
+        self.assertEqual(
+            Codes.MISSING_PARAM,
+            channel.json_body["errcode"],
+            channel.json_body,
+        )
+
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_delayed_event_with_no_keys(self) -> None:
+        """Test that the dedicated endpoint fails with a body missing all required keys."""
+        channel = self.make_request(
+            "PUT",
+            self.helper.build_delayed_event_request_path(
+                room_id=self.room_id,
+                event_type="m.room.topic",
+            ),
+            {},
+        )
+        self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
+        self.assertEqual(
+            Codes.BAD_JSON,
+            channel.json_body["errcode"],
+            channel.json_body,
+        )
+
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_delayed_event_with_null_state_key(self) -> None:
+        """
+        Test that the dedicated endpoint fails to schedule a message event
+        with a "state_key" of `None` in the request body.
+        """
+        body = self.helper.build_delayed_event_request_body(
+            delay=Duration(milliseconds=2000),
+            content={"body": "test", "msgtype": "m.text"},
+        )
+        body["state_key"] = None
+        channel = self.make_request(
+            "PUT",
+            self.helper.build_delayed_event_request_path(
+                room_id=self.room_id,
+                event_type="m.room.message",
+            ),
+            body,
+        )
+        self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
+        self.assertEqual(
+            Codes.BAD_JSON,
+            channel.json_body["errcode"],
+            channel.json_body,
+        )
+
+    @unittest.override_config({"max_event_delay_duration": "24h"})
+    def test_send_delayed_state_event_idempotent(self) -> None:
+        """
+        Test that repeated requests to schedule a delayed state event
+        with the same transaction ID are idempotent.
+        Note that only the dedicated endpoint can test this, as the one based on /state
+        does not support idempotent requests by using a transaction ID.
+        """
+        make_delayed_event_request = lambda: self.make_delayed_event_request(
+            room_id=self.room_id,
+            delay=Duration(milliseconds=2000),
+            event_type="m.room.topic",
+            state_key="",
+            content={"topic": "This is a topic"},
+            txn_id="txn_id_0",
+        )
+        channel = make_delayed_event_request()
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+
+        delay_id = channel.json_body["delay_id"]
+
+        # Repeat the request, which reuses its txn_id
+        channel = make_delayed_event_request()
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        self.assertEqual(delay_id, channel.json_body.get("delay_id"))
+
+
 class RoomSearchTestCase(unittest.HomeserverTestCase):
     servlets = [
         synapse.rest.admin.register_servlets_for_client_rest_resource,

tests/rest/client/utils.py

@@ -636,6 +636,34 @@ class RestHelper:
             room_id, event_type, body, tok, expect_code, state_key, method="PUT"
         )
 
+    @staticmethod
+    def build_delayed_event_request_path(
+        *,
+        room_id: str,
+        event_type: str,
+        txn_id: str | None = None,
+    ) -> str:
+        """Build the request path for scheduling a delayed event via the dedicated endpoint."""
+        if txn_id is None:
+            txn_id = "m%s" % (str(time.time()))
+        return f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/{txn_id}"
+
+    @staticmethod
+    def build_delayed_event_request_body(
+        *,
+        delay: Duration,
+        content: JsonDict,
+        state_key: str | None = None,
+    ) -> JsonDict:
+        """Build the request body for scheduling a delayed event via the dedicated endpoint."""
+        body = {
+            "delay_ms": delay.as_millis(),
+            "content": content,
+        }
+        if state_key is not None:
+            body["state_key"] = state_key
+        return body
+
     def upload_media(
         self,
         image_data: bytes,