From a5c56b2340c1739ffb9e5f9657d201db5e2b8f48 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Mon, 22 Dec 2025 14:34:24 +0000 Subject: [PATCH] Allow clients to send sticky events Including delayed events --- synapse/handlers/delayed_events.py | 13 ++++++++--- synapse/rest/client/room.py | 23 ++++++++++++++++++- .../storage/databases/main/delayed_events.py | 9 +++++++- 3 files changed, 40 insertions(+), 5 deletions(-) diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index c58d1d42bc..a9b32b6fb0 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -17,7 +17,7 @@ from typing import TYPE_CHECKING, Optional from twisted.internet.interfaces import IDelayedCall -from synapse.api.constants import EventTypes +from synapse.api.constants import EventTypes, StickyEvent from synapse.api.errors import ShadowBanError, SynapseError from synapse.api.ratelimiting import Ratelimiter from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME @@ -333,6 +333,7 @@ class DelayedEventsHandler: origin_server_ts: int | None, content: JsonDict, delay: int, + sticky_duration_ms: int | None, ) -> str: """ Creates a new delayed event and schedules its delivery. @@ -346,7 +347,9 @@ class DelayedEventsHandler: 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. - + 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: @@ -382,6 +385,7 @@ class DelayedEventsHandler: origin_server_ts=origin_server_ts, content=content, delay=delay, + sticky_duration_ms=sticky_duration_ms, ) if self._repl_client is not None: @@ -570,7 +574,10 @@ class DelayedEventsHandler: if event.state_key is not None: event_dict["state_key"] = event.state_key - + if event.sticky_duration_ms is not None: + event_dict[StickyEvent.FIELD_NAME] = { + "duration_ms": event.sticky_duration_ms, + } ( sent_event, _, diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py index 5e7dcb0191..d09425a186 100644 --- a/synapse/rest/client/room.py +++ b/synapse/rest/client/room.py @@ -34,7 +34,7 @@ from prometheus_client.core import Histogram from twisted.web.server import Request from synapse import event_auth -from synapse.api.constants import Direction, EventTypes, Membership +from synapse.api.constants import Direction, EventTypes, Membership, StickyEvent from synapse.api.errors import ( AuthError, Codes, @@ -210,6 +210,7 @@ class RoomStateEventRestServlet(RestServlet): self.clock = hs.get_clock() 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 def register(self, http_server: HttpServer) -> None: # /rooms/$roomid/state/$eventtype @@ -331,6 +332,10 @@ class RoomStateEventRestServlet(RestServlet): if requester.app_service: origin_server_ts = parse_integer(request, "ts") + sticky_duration_ms: int | None = None + if self._msc4354_enabled: + sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) + delay = _parse_request_delay(request, self._max_event_delay_ms) if delay is not None: delay_id = await self.delayed_events_handler.add( @@ -341,6 +346,7 @@ class RoomStateEventRestServlet(RestServlet): origin_server_ts=origin_server_ts, content=content, delay=delay, + sticky_duration_ms=sticky_duration_ms, ) set_tag("delay_id", delay_id) @@ -368,6 +374,10 @@ class RoomStateEventRestServlet(RestServlet): "room_id": room_id, "sender": requester.user.to_string(), } + if sticky_duration_ms is not None: + event_dict[StickyEvent.FIELD_NAME] = { + "duration_ms": sticky_duration_ms, + } if state_key is not None: event_dict["state_key"] = state_key @@ -400,6 +410,7 @@ class RoomSendEventRestServlet(TransactionRestServlet): 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: # /rooms/$roomid/send/$event_type[/$txn_id] @@ -420,6 +431,10 @@ class RoomSendEventRestServlet(TransactionRestServlet): if requester.app_service: origin_server_ts = parse_integer(request, "ts") + sticky_duration_ms: int | None = None + if self._msc4354_enabled: + sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) + delay = _parse_request_delay(request, self._max_event_delay_ms) if delay is not None: delay_id = await self.delayed_events_handler.add( @@ -430,6 +445,7 @@ class RoomSendEventRestServlet(TransactionRestServlet): origin_server_ts=origin_server_ts, content=content, delay=delay, + sticky_duration_ms=sticky_duration_ms, ) set_tag("delay_id", delay_id) @@ -446,6 +462,11 @@ class RoomSendEventRestServlet(TransactionRestServlet): if origin_server_ts is not None: event_dict["origin_server_ts"] = origin_server_ts + if sticky_duration_ms is not None: + event_dict[StickyEvent.FIELD_NAME] = { + "duration_ms": sticky_duration_ms, + } + try: ( event, diff --git a/synapse/storage/databases/main/delayed_events.py b/synapse/storage/databases/main/delayed_events.py index 5547150515..1727f589e2 100644 --- a/synapse/storage/databases/main/delayed_events.py +++ b/synapse/storage/databases/main/delayed_events.py @@ -54,6 +54,7 @@ class EventDetails: origin_server_ts: Timestamp | None content: JsonDict device_id: DeviceID | None + sticky_duration_ms: int | None @attr.s(slots=True, frozen=True, auto_attribs=True) @@ -122,6 +123,7 @@ class DelayedEventsStore(SQLBaseStore): origin_server_ts: int | None, content: JsonDict, delay: int, + sticky_duration_ms: int | None, ) -> tuple[DelayID, Timestamp]: """ Inserts a new delayed event in the DB. @@ -148,6 +150,7 @@ class DelayedEventsStore(SQLBaseStore): "state_key": state_key, "origin_server_ts": origin_server_ts, "content": json_encoder.encode(content), + "sticky_duration_ms": sticky_duration_ms, }, ) @@ -299,6 +302,7 @@ class DelayedEventsStore(SQLBaseStore): "send_ts", "content", "device_id", + "sticky_duration_ms", ) ) sql_update = "UPDATE delayed_events SET is_processed = TRUE" @@ -344,6 +348,7 @@ class DelayedEventsStore(SQLBaseStore): Timestamp(row[5] if row[5] is not None else row[6]), db_to_json(row[7]), DeviceID(row[8]) if row[8] is not None else None, + int(row[9]) if row[9] is not None else None, DelayID(row[0]), UserLocalpart(row[1]), ) @@ -392,6 +397,7 @@ class DelayedEventsStore(SQLBaseStore): origin_server_ts, content, device_id, + sticky_duration_ms, user_localpart """, (delay_id,), @@ -407,8 +413,9 @@ class DelayedEventsStore(SQLBaseStore): Timestamp(row[3]) if row[3] is not None else None, db_to_json(row[4]), DeviceID(row[5]) if row[5] is not None else None, + int(row[6]) if row[6] is not None else None, DelayID(delay_id), - UserLocalpart(row[6]), + UserLocalpart(row[7]), ) return event, self._get_next_delayed_event_send_ts_txn(txn)