From 85dc8974ebc69d9075bc4ca5a64b8d263ea8b78e Mon Sep 17 00:00:00 2001 From: Andrew Ferrazzutti Date: Tue, 7 Jul 2026 14:17:11 -0400 Subject: [PATCH 1/2] MSC4140: support getting a single delayed event See https://github.com/matrix-org/matrix-spec-proposals/blob/toger5/expiring-events-keep-alive/proposals/4140-delayed-events-futures.md#getting-a-single-delayed-event --- synapse/handlers/delayed_events.py | 13 +++++ synapse/rest/client/delayed_events.py | 20 +++++++ .../storage/databases/main/delayed_events.py | 58 +++++++++++++++---- tests/rest/client/test_delayed_events.py | 42 ++++++++++++++ 4 files changed, 121 insertions(+), 12 deletions(-) diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index f016d95e314..51c1a9d5d7e 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -530,6 +530,19 @@ def _schedule_next_at(self, next_send_ts: Timestamp) -> None: else: self._next_delayed_event_call.reset(delay_duration.as_secs()) + async def get_for_user(self, requester: Requester, delay_id: str) -> JsonDict: + """ + Return the specified pending delayed event requested by the given user. + + Raises: + NotFoundError: if no matching delayed event could be found. + """ + await self._delayed_event_mgmt_ratelimiter.ratelimit(requester) + return await self._store.get_delayed_event_for_user( + delay_id, + requester.user.localpart, + ) + async def get_all_for_user(self, requester: Requester) -> list[JsonDict]: """Return all pending delayed events requested by the given user.""" await self._delayed_event_mgmt_ratelimiter.ratelimit(requester) diff --git a/synapse/rest/client/delayed_events.py b/synapse/rest/client/delayed_events.py index 7afecffe2d0..7ce75e8acbf 100644 --- a/synapse/rest/client/delayed_events.py +++ b/synapse/rest/client/delayed_events.py @@ -134,6 +134,25 @@ async def on_POST( return 200, {} +class DelayedEventServlet(RestServlet): + PATTERNS = client_patterns( + r"/org\.matrix\.msc4140/delayed_events/(?P[^/]+)$", + releases=(), + ) + CATEGORY = "Delayed event management requests" + + def __init__(self, hs: "HomeServer"): + super().__init__() + self.auth = hs.get_auth() + self.delayed_events_handler = hs.get_delayed_events_handler() + + async def on_GET( + self, request: SynapseRequest, delay_id: str + ) -> tuple[int, JsonDict]: + requester = await self.auth.get_user_by_req(request) + return 200, await self.delayed_events_handler.get_for_user(requester, delay_id) + + class DelayedEventsServlet(RestServlet): PATTERNS = client_patterns( r"/org\.matrix\.msc4140/delayed_events$", @@ -162,4 +181,5 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: CancelDelayedEventServlet(hs).register(http_server) SendDelayedEventServlet(hs).register(http_server) RestartDelayedEventServlet(hs).register(http_server) + DelayedEventServlet(hs).register(http_server) DelayedEventsServlet(hs).register(http_server) diff --git a/synapse/storage/databases/main/delayed_events.py b/synapse/storage/databases/main/delayed_events.py index 1727f589e2a..e6f87efab51 100644 --- a/synapse/storage/databases/main/delayed_events.py +++ b/synapse/storage/databases/main/delayed_events.py @@ -225,6 +225,39 @@ def _get_count_of_delayed_events(txn: LoggingTransaction) -> int: _get_count_of_delayed_events, ) + async def get_delayed_event_for_user( + self, + delay_id: str, + user_localpart: str, + ) -> JsonDict: + """ + Returns the specified pending delayed event owned by the given user. + + Raises: + NotFoundError: if there is no matching delayed event. + """ + row = await self.db_pool.simple_select_one( + table="delayed_events", + keyvalues={ + "delay_id": delay_id, + "user_localpart": user_localpart, + "is_processed": False, + }, + retcols=( + "room_id", + "event_type", + "state_key", + "delay", + "send_ts", + "content", + ), + allow_none=True, + desc="get_delayed_event_for_user", + ) + if row is None: + raise NotFoundError("Delayed event not found") + return _row_to_delayed_event_dict((delay_id, *row)) + async def get_all_delayed_events_for_user( self, user_localpart: str, @@ -248,18 +281,7 @@ async def get_all_delayed_events_for_user( """, user_localpart, ) - return [ - { - "delay_id": DelayID(row[0]), - "room_id": str(RoomID.from_string(row[1])), - "type": EventType(row[2]), - **({"state_key": StateKey(row[3])} if row[3] is not None else {}), - "delay": Delay(row[4]), - "running_since": Timestamp(row[5] - row[4]), - "content": db_to_json(row[6]), - } - for row in rows - ] + return [_row_to_delayed_event_dict(row) for row in rows] async def process_timeout_delayed_events( self, current_ts: Timestamp, reprocess_events: bool = False @@ -573,6 +595,18 @@ def _get_next_delayed_event_send_ts_txn( return Timestamp(result) if result is not None else None +def _row_to_delayed_event_dict(row: tuple) -> JsonDict: + return { + "delay_id": DelayID(row[0]), + "room_id": str(RoomID.from_string(row[1])), + "type": EventType(row[2]), + **({"state_key": StateKey(row[3])} if row[3] is not None else {}), + "delay": Delay(row[4]), + "running_since": Timestamp(row[5] - row[4]), + "content": db_to_json(row[6]), + } + + def _generate_delay_id() -> DelayID: """Generates an opaque string, for use as a delay ID""" diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py index 75d716244a8..deeded5f915 100644 --- a/tests/rest/client/test_delayed_events.py +++ b/tests/rest/client/test_delayed_events.py @@ -127,6 +127,48 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: def test_delayed_events_empty_on_startup(self) -> None: self.assertListEqual([], self._get_delayed_events()) + def test_delayed_event_lookup(self) -> None: + channel = self.make_request( + "POST", + _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 100000), + {}, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + delay_id = channel.json_body["delay_id"] + + # Test that the scheduled delayed event can be retrieved + channel = self.make_request( + "GET", + f"{PATH_PREFIX}/{delay_id}", + access_token=self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + event = channel.json_body + self.assertEqual(delay_id, event["delay_id"]) + + # Test that the list lookup retrieves the same item + events = self._get_delayed_events() + self.assertEqual(1, len(events), events) + self.assertDictEqual(events[0], event) + + # Test that a non-existent delayed event cannot be found + channel = self.make_request( + "GET", + f"{PATH_PREFIX}/{delay_id}-fake", + access_token=self.user1_access_token, + ) + self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result) + + # Test that other users cannot access this delayed event + channel = self.make_request( + "GET", + f"{PATH_PREFIX}/{delay_id}", + access_token=self.user2_access_token, + ) + self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result) + def test_delayed_state_events_are_sent_on_timeout(self) -> None: state_key = "to_send_on_timeout" From ffb8dcf0869c593ffbaf9834a1a7373006cc8219 Mon Sep 17 00:00:00 2001 From: Andrew Ferrazzutti Date: Tue, 7 Jul 2026 14:22:49 -0400 Subject: [PATCH 2/2] Add changelog --- changelog.d/19926.feature | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19926.feature diff --git a/changelog.d/19926.feature b/changelog.d/19926.feature new file mode 100644 index 00000000000..88f31110790 --- /dev/null +++ b/changelog.d/19926.feature @@ -0,0 +1 @@ +[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): Add an endpoint for getting a single delayed event.