# # This file is licensed under the Affero General Public License (AGPL) version 3. # # Copyright (C) 2024 New Vector, Ltd # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # See the GNU Affero General Public License for more details: # . # """Tests REST events for /delayed_events paths.""" import sqlite3 from http import HTTPStatus from typing import Literal, overload from parameterized import parameterized from twisted.internet.testing import MemoryReactor from synapse.api.constants import ( EventContentFields, EventTypes, EventUnsignedContentFields, GuestAccess, StickyEvent, ) from synapse.api.errors import Codes, NotFoundError from synapse.rest import admin from synapse.rest.client import ( account, delayed_events, login, register, room, sync, versions, ) from synapse.rest.client.delayed_events import _UpdateDelayedEventAction from synapse.server import HomeServer from synapse.types import JsonDict, create_requester from synapse.util.clock import Clock from synapse.util.duration import Duration from tests import unittest from tests.replication._base import BaseMultiWorkerStreamTestCase from tests.server import FakeChannel, make_request from tests.unittest import HomeserverTestCase from tests.utils import USE_POSTGRES_FOR_TESTS _MANAGEMENT_PATH_PREFIX = "/_matrix/client/unstable/org.matrix.msc4140/delayed_events" _EVENT_TYPE = "com.example.test" class DelayedEventsUnstableSupportTestCase(HomeserverTestCase): servlets = [versions.register_servlets] def tearDown(self) -> None: # MemoryReactor doesn't trigger the shutdown phases, and we want the # Tokio thread pool to be stopped # XXX: This logic should probably get moved somewhere else shutdown_triggers = self.reactor.triggers.get("shutdown", {}) for phase in ["before", "during", "after"]: triggers = shutdown_triggers.get(phase, []) for callbable, args, kwargs in triggers: callbable(*args, **kwargs) def test_false_by_default(self) -> None: channel = self.make_request("GET", "/_matrix/client/versions") self.assertEqual(channel.code, 200, channel.result) self.assertFalse(channel.json_body["unstable_features"]["org.matrix.msc4140"]) @unittest.override_config({"max_event_delay_duration": "24h"}) def test_true_if_enabled(self) -> None: channel = self.make_request("GET", "/_matrix/client/versions") self.assertEqual(channel.code, 200, channel.result) self.assertTrue(channel.json_body["unstable_features"]["org.matrix.msc4140"]) class DelayedEventsHelperMixin(HomeserverTestCase): room_id: str 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, access_token: str, ) -> FakeChannel: """ Build and send a request for scheduling a delayed event via the dedicated endpoint. 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. access_token: The access token of the user to make the request for. Returns: The `FakeChannel` object which stores the result of the request. """ 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, ), access_token, ) @overload def _check_for_delayed_event_in_sync( self, access_token: str, delay_id: str, should_find: Literal[True] ) -> JsonDict: ... @overload def _check_for_delayed_event_in_sync( self, access_token: str, delay_id: str, should_find: Literal[False] ) -> None: ... def _check_for_delayed_event_in_sync( self, access_token: str, delay_id: str, should_find: bool ) -> JsonDict | None: """Call /sync and look for a synced event with a specified delay_id. At most one event will ever have a matching delay_id. Args: access_token: The access token of the user to call /sync for. delay_id: The delay_id to search for in synced events. should_find: Whether /sync should include an event with a matching delay_id. Returns: The synced event with the matching delay_id, if any. """ channel = self.make_request("GET", "/sync", access_token=access_token) self.assertEqual(HTTPStatus.OK, channel.code) rooms = channel.json_body["rooms"] events = [] for membership in "join", "leave": if membership in rooms: events += rooms[membership][self.room_id]["timeline"]["events"] found: JsonDict | None = None for event in events: if event["unsigned"].get("org.matrix.msc4140.delay_id") == delay_id: if not should_find: self.fail( f"Found event {event['event_id']} with matching delay_id {delay_id}, but expected to not find one" ) if found is not None: self.fail( f"Events {found['event_id']} & {event['event_id']} have matching delay_id {delay_id}, but delay_id should be unique" ) found = event if should_find and found is None: self.fail(f"Did not find any event with matching delay_id {delay_id}") return found class DelayedEventsTestCaseBase(DelayedEventsHelperMixin): """Room and user fixtures, and request helpers, for the delayed events tests.""" servlets = [ admin.register_servlets, account.register_servlets, delayed_events.register_servlets, login.register_servlets, register.register_servlets, room.register_servlets, sync.register_servlets, ] def default_config(self) -> JsonDict: config = super().default_config() config["max_event_delay_duration"] = "24h" return config def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.user1_user_id = self.register_user("user1", "pass") self.user1_access_token = self.login("user1", "pass") self.user2_user_id = self.register_user("user2", "pass") self.user2_access_token = self.login("user2", "pass") self.room_id = self.helper.create_room_as( self.user1_user_id, tok=self.user1_access_token, extra_content={ "preset": "public_chat", "power_level_content_override": { "events": { _EVENT_TYPE: 0, } }, }, ) self.helper.join( room=self.room_id, user=self.user2_user_id, tok=self.user2_access_token ) # Advance enough time where any requests we made during `prepare(...)` doesn't # affect the rate-limits in the test itself self.reactor.advance(Duration(days=1).as_secs()) def _get_delayed_events(self) -> list[JsonDict]: channel = self.make_request( "GET", _MANAGEMENT_PATH_PREFIX, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) events = channel.json_body["delayed_events"] self.assertIsInstance(events, list) return events def _get_delayed_event_content(self, event: JsonDict) -> JsonDict: content = event["content"] self.assertIsInstance(content, dict) return content def _update_delayed_event( self, delay_id: str, action: _UpdateDelayedEventAction, action_in_path: bool, access_token: str | None = None, ) -> FakeChannel: """ Call the endpoint to cancel/restart/send a scheduled delayed event. Args: delay_id: The ID of the delayed event to manage. action: The management action to take on the target delayed event. action_in_path: If False, uses the deprecated version of this endpoint that expects the action in the request body instead of the path. access_token: The access token of the user to call the endpoint for. """ path = f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}" body = {} if action_in_path: path += f"/{action.value}" else: body["action"] = action.value return self.make_request("POST", path, body, access_token) class DelayedEventsTestCase(DelayedEventsTestCaseBase): """Tests getting and managing delayed events.""" def test_delayed_events_empty_on_startup(self) -> None: self.assertListEqual([], self._get_delayed_events()) def test_delayed_event_lookup(self) -> None: # Schedule a message event delay = Duration(milliseconds=100000) content: JsonDict = {"message": "hello"} delayed_since_ts = self.hs.get_clock().time_msec() channel = self._make_delayed_event_request( room_id=self.room_id, delay=delay, event_type=_EVENT_TYPE, content=content, access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) delay_id = channel.json_body["delay_id"] # Test that the scheduled delayed event can be retrieved channel = self.make_request( "GET", f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}", access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) # Assert the stored properties of the delayed event event = channel.json_body self.assertDictEqual( event, { "delay_id": delay_id, "room_id": self.room_id, "type": _EVENT_TYPE, "delay_ms": delay.as_millis(), "delayed_since_ts": delayed_since_ts, "content": content, }, ) # Test that a non-existent delayed event cannot be found channel = self.make_request( "GET", f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}-fake", access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.NOT_FOUND, channel.result) # Test that other users cannot access this delayed event channel = self.make_request( "GET", f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}", access_token=self.user2_access_token, ) self.assertEqual(channel.code, HTTPStatus.NOT_FOUND, channel.result) # Now schedule a state event. # Do it in this test, as opposed to a new one, to confirm that the correct delayed event # is retrieved when multiple delayed events have been scheduled. delay += Duration(milliseconds=2000) state_key = "" state_event_type = _EVENT_TYPE + "_state" content = {"state_message": "greetings"} delayed_since_ts = self.hs.get_clock().time_msec() channel = self._make_delayed_event_request( room_id=self.room_id, delay=delay, event_type=state_event_type, state_key=state_key, content=content, access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) delay_id_2 = channel.json_body["delay_id"] # Test that the new delayed event has a different ID from the previous one self.assertNotEqual(delay_id, delay_id_2) # Test that the scheduled delayed event can be retrieved channel = self.make_request( "GET", f"{_MANAGEMENT_PATH_PREFIX}/{delay_id_2}", access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) # Assert the stored properties of the delayed event state_event = channel.json_body self.assertDictEqual( state_event, { "delay_id": delay_id_2, "room_id": self.room_id, "type": state_event_type, "state_key": state_key, "delay_ms": delay.as_millis(), "delayed_since_ts": delayed_since_ts, "content": content, }, ) # Test that the list lookup retrieves the same items (with legacy fields included) self.assertEqual( self._get_delayed_events(), [ event | { "delay": event["delay_ms"], "running_since": event["delayed_since_ts"], }, state_event | { "delay": state_event["delay_ms"], "running_since": state_event["delayed_since_ts"], }, ], ) def test_delayed_state_events_are_sent_on_timeout(self) -> None: state_key = "to_send_on_timeout" setter_key = "setter" setter_expected = "on_timeout" channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type=_EVENT_TYPE, state_key=state_key, content={ setter_key: setter_expected, }, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) self.assertEqual(setter_expected, content.get(setter_key), content) self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, ) self.assertEqual(setter_expected, content.get(setter_key), content) self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @unittest.override_config({"allow_guest_access": True}) def test_guest_can_schedule_delayed_event(self) -> None: """Test that a guest user can schedule a delayed event.""" channel = self.make_request("POST", "/register?kind=guest", {}) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) guest_user_id = channel.json_body["user_id"] guest_access_token = channel.json_body["access_token"] self.helper.send_state( self.room_id, EventTypes.GuestAccess, {EventContentFields.GUEST_ACCESS: GuestAccess.CAN_JOIN}, tok=self.user1_access_token, ) self.helper.join(room=self.room_id, user=guest_user_id, tok=guest_access_token) channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type=_EVENT_TYPE, content={"body": "from a guest"}, access_token=guest_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) event = self._check_for_delayed_event_in_sync( guest_access_token, delay_id, True ) self.assertEqual(guest_user_id, event["sender"], event) def test_delayed_member_events_are_sent_on_timeout(self) -> None: channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type="m.room.member", state_key=self.user2_user_id, content={ "membership": "leave", "reason": "Delayed kick", }, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) self.assertEqual("leave", content.get("membership"), content) self.assertEqual("Delayed kick", content.get("reason"), content) content = self.helper.get_state( self.room_id, "m.room.member", self.user1_access_token, state_key=self.user2_user_id, ) self.assertEqual("join", content.get("membership"), content) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, "m.room.member", self.user1_access_token, state_key=self.user2_user_id, ) self.assertEqual("leave", content.get("membership"), content) self.assertEqual("Delayed kick", content.get("reason"), content) self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) def test_get_delayed_events_auth(self) -> None: channel = self.make_request("GET", _MANAGEMENT_PATH_PREFIX) self.assertEqual(HTTPStatus.UNAUTHORIZED, channel.code, channel.result) @unittest.override_config( {"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}} ) def test_get_delayed_events_ratelimit(self) -> None: args = ("GET", _MANAGEMENT_PATH_PREFIX, b"", self.user1_access_token) channel = self.make_request(*args) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) channel = self.make_request(*args) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Add the current user to the ratelimit overrides, allowing them no ratelimiting. self.get_success( self.hs.get_datastores().main.set_ratelimit_for_user( self.user1_user_id, 0, 0 ) ) # Test that the request isn't ratelimited anymore. channel = self.make_request(*args) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) def test_update_delayed_event_without_id(self) -> None: channel = self.make_request( "POST", f"{_MANAGEMENT_PATH_PREFIX}/", ) self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result) def test_update_delayed_event_without_body(self) -> None: channel = self.make_request( "POST", f"{_MANAGEMENT_PATH_PREFIX}/abc", ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual( Codes.NOT_JSON, channel.json_body["errcode"], ) def test_update_delayed_event_without_action(self) -> None: channel = self.make_request( "POST", f"{_MANAGEMENT_PATH_PREFIX}/abc", {}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual( Codes.MISSING_PARAM, channel.json_body["errcode"], ) def test_update_delayed_event_with_invalid_action(self) -> None: channel = self.make_request( "POST", f"{_MANAGEMENT_PATH_PREFIX}/abc", {"action": "oops"}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual( Codes.INVALID_PARAM, channel.json_body["errcode"], ) @parameterized.expand( ( (action, action_in_path) for action in _UpdateDelayedEventAction for action_in_path in (True, False) ) ) def test_update_delayed_event_without_match( self, action: _UpdateDelayedEventAction, action_in_path: bool, ) -> None: channel = self._update_delayed_event("abc", action, action_in_path) self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result) @parameterized.expand((True, False)) def test_cancel_delayed_state_event(self, action_in_path: bool) -> None: state_key = "to_never_send" setter_key = "setter" setter_expected = "none" channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=1500), event_type=_EVENT_TYPE, state_key=state_key, content={ setter_key: setter_expected, }, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) self.assertEqual(setter_expected, content.get(setter_key), content) self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertListEqual([], self._get_delayed_events()) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, False) self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((True, False)) @unittest.override_config( {"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}} ) def test_cancel_delayed_event_ratelimit(self, action_in_path: bool) -> None: delay_ids = [] for _ in range(3): channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=100000), event_type=_EVENT_TYPE, content={}, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None delay_ids.append(delay_id) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Using auth should bypass ratelimit applied against source IP channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path, self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path, self.user1_access_token, ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Add the current user to the ratelimit overrides, allowing them no ratelimiting. self.get_success( self.hs.get_datastores().main.set_ratelimit_for_user( self.user1_user_id, 0, 0 ) ) # Test that the request isn't ratelimited anymore. channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path, self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @parameterized.expand( ( (content_property_value, action_in_path) for content_property_value in ("test", "tест") for action_in_path in (True, False) ) ) def test_send_delayed_state_event( self, content_value: str, action_in_path: bool ) -> None: state_key = "to_send_on_request" content_property_name = "key" channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=100000), event_type=_EVENT_TYPE, state_key=state_key, content={ content_property_name: content_value, }, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) self.assertEqual(content_value, content.get(content_property_name), content) self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.SEND, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, ) self.assertEqual(content_value, content.get(content_property_name), content) self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((True, False)) @unittest.override_config({"rc_message": {"per_second": 2.5, "burst_count": 3}}) def test_send_delayed_event_ratelimit(self, action_in_path: bool) -> None: delay_ids = [] for _ in range(2): channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=100000), event_type=_EVENT_TYPE, content={}, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None delay_ids.append(delay_id) channel = self._update_delayed_event( delay_ids.pop(0), _UpdateDelayedEventAction.SEND, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) channel = self._update_delayed_event( delay_ids.pop(0), _UpdateDelayedEventAction.SEND, action_in_path ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) @parameterized.expand((True, False)) def test_restart_delayed_state_event(self, action_in_path: bool) -> None: state_key = "to_send_on_restarted_timeout" setter_key = "setter" setter_expected = "on_timeout" channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=1500), event_type=_EVENT_TYPE, state_key=state_key, content={ setter_key: setter_expected, }, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) self.assertEqual(setter_expected, content.get(setter_key), content) self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.RESTART, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) self.assertEqual(setter_expected, content.get(setter_key), content) self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, ) self.assertEqual(setter_expected, content.get(setter_key), content) self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((True, False)) @unittest.override_config( {"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}} ) def test_restart_delayed_event_ratelimit(self, action_in_path: bool) -> None: delay_ids = [] for _ in range(3): channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=100000), event_type=_EVENT_TYPE, content={}, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None delay_ids.append(delay_id) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.RESTART, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.RESTART, action_in_path ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Using auth should bypass ratelimit applied against source IP channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.RESTART, action_in_path, self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.RESTART, action_in_path, self.user1_access_token, ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Add the current user to the ratelimit overrides, allowing them no ratelimiting. self.get_success( self.hs.get_datastores().main.set_ratelimit_for_user( self.user1_user_id, 0, 0 ) ) # Test that the request isn't ratelimited anymore. channel = self._update_delayed_event( delay_id, _UpdateDelayedEventAction.RESTART, action_in_path, self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) def test_delayed_state_is_not_cancelled_by_new_state_from_same_user( self, ) -> None: state_key = "to_not_be_cancelled_by_same_user" setter_key = "setter" setter_expected = "on_timeout" channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type=_EVENT_TYPE, state_key=state_key, content={ setter_key: setter_expected, }, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None events = self._get_delayed_events() self.assertEqual(1, len(events), events) self.helper.send_state( self.room_id, _EVENT_TYPE, { setter_key: "manual", }, self.user1_access_token, state_key=state_key, ) events = self._get_delayed_events() self.assertEqual(1, len(events), events) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, self.user1_access_token, state_key=state_key, ) self.assertEqual(setter_expected, content.get(setter_key), content) self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((("client_api", False), ("admin_api", True))) def test_delayed_events_are_cancelled_on_deactivation( self, _name: str, by_admin: bool ) -> None: store = self.hs.get_datastores().main handler = self.hs.get_delayed_events_handler() # user1 schedules a delayed message event and a delayed state event. user1_delay_ids = [] state_key = "to_cancel_on_deactivation" for state_event in True, False: channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type=_EVENT_TYPE, state_key=state_key if state_event else None, content={"key": "value"}, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) user1_delay_ids.append(channel.json_body["delay_id"]) # user2 schedules a delayed event to be sent later than user1's. channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=2000), event_type=_EVENT_TYPE, content={"key": "value"}, access_token=self.user2_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) user2_delay_id = channel.json_body["delay_id"] self.assertEqual(2, len(self._get_delayed_events())) # The next send is scheduled for user1's delayed events. assert handler._next_delayed_event_call is not None self.assertLess( handler._next_delayed_event_call.getTime() - self.reactor.seconds(), 1 ) if by_admin: self.register_user("admin", "pass", admin=True) channel = self.make_request( "POST", f"/_synapse/admin/v1/deactivate/{self.user1_user_id}", {}, self.login("admin", "pass"), ) else: channel = self.make_request( "POST", "account/deactivate", { "auth": { "type": "m.login.password", "user": self.user1_user_id, "password": "pass", }, }, self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) # user1's delayed events are gone... self.assertListEqual( [], self.get_success(store.get_all_delayed_events_for_user("user1")) ) for delay_id in user1_delay_ids: self.get_failure( store.get_delayed_event_for_user(delay_id, "user1"), NotFoundError ) # ...and the next send has been rescheduled for user2's delayed event. assert handler._next_delayed_event_call is not None self.assertGreater( handler._next_delayed_event_call.getTime() - self.reactor.seconds(), 1 ) # Nothing gets sent when user1's delayed events would have timed out. # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) for delay_id in user1_delay_ids: self._check_for_delayed_event_in_sync( self.user2_access_token, delay_id, False ) self.helper.get_state( self.room_id, _EVENT_TYPE, self.user2_access_token, state_key=state_key, expect_code=HTTPStatus.NOT_FOUND, ) # user2's delayed event is unaffected, and still gets sent. self._check_for_delayed_event_in_sync( self.user2_access_token, user2_delay_id, False ) # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) for delay_id in user1_delay_ids: self._check_for_delayed_event_in_sync( self.user2_access_token, delay_id, False ) self._check_for_delayed_event_in_sync( self.user2_access_token, user2_delay_id, True ) def test_deactivation_cancels_the_timer_when_no_delayed_events_remain( self, ) -> None: handler = self.hs.get_delayed_events_handler() channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type=_EVENT_TYPE, content={}, access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertIsNotNone(handler._next_delayed_event_call) self.get_success( self.hs.get_deactivate_account_handler().deactivate_account( self.user1_user_id, erase_data=False, requester=create_requester(self.user1_user_id), ) ) self.assertIsNone(handler._next_delayed_event_call) class DelayedStickyEventsTestCase(DelayedEventsTestCaseBase): """Tests scheduling delayed sticky events (MSC4354).""" if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0): # We need the JSON functionality in SQLite # (the sticky events store refuses to start without it, so this has to be # a class-level skip: the homeserver is built before the test body runs) skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)" def default_config(self) -> JsonDict: config = super().default_config() config["experimental_features"] = {"msc4354_enabled": True} return config @parameterized.expand( ( (False, True), (True, False), (True, True), ) ) def test_delayed_sticky_event_is_sent_with_sticky_duration( self, param_in_body: bool, param_in_query: bool ) -> None: """Test that the sticky duration given when scheduling a delayed event is applied to the event once it is sent.""" path = self.helper.build_delayed_event_request_path( room_id=self.room_id, event_type=_EVENT_TYPE, ) body = self.helper.build_delayed_event_request_body( delay=Duration(milliseconds=900), content={"body": "sticky"}, ) sticky_duration = Duration(minutes=1) if param_in_body: body[StickyEvent.REQUEST_PARAM_NAME] = sticky_duration.as_millis() if param_in_query: path += f"?{StickyEvent.REQUEST_PARAM_NAME}={sticky_duration.as_millis()}" channel = self.make_request("PUT", path, body, self.user1_access_token) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) event = self._check_for_delayed_event_in_sync( self.user1_access_token, delay_id, True ) self.assertGreater( event["unsigned"].get(EventUnsignedContentFields.STICKY_TTL, 0), 0, event, ) @parameterized.expand((False, True)) def test_delayed_sticky_state_event(self, param_in_body: bool) -> None: """Test that delayed event requests forbid scheduling a sticky state event.""" path = self.helper.build_delayed_event_request_path( room_id=self.room_id, event_type="m.room.topic", ) body = self.helper.build_delayed_event_request_body( delay=Duration(milliseconds=900), content={"topic": "This is a topic"}, state_key="", ) sticky_duration = Duration(minutes=1) if param_in_body: body[StickyEvent.REQUEST_PARAM_NAME] = sticky_duration.as_millis() else: path += f"?{StickyEvent.REQUEST_PARAM_NAME}={sticky_duration.as_millis()}" channel = self.make_request("PUT", path, body, self.user1_access_token) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) @parameterized.expand( ( (sticky_duration, param_in_body) for sticky_duration in ( Duration(), Duration(seconds=-2000), Duration(days=1), ) for param_in_body in (False, True) ) ) def test_delayed_sticky_event_with_extreme_duration( self, sticky_duration: Duration, param_in_body: bool ) -> None: """Test that delayed event requests permit a zero, negative, or very long sticky duration. (The actual sticky duration is clamped at event send time.)""" path = self.helper.build_delayed_event_request_path( room_id=self.room_id, event_type=_EVENT_TYPE, ) body = self.helper.build_delayed_event_request_body( delay=Duration(milliseconds=900), content={"body": "sticky"}, ) if param_in_body: body[StickyEvent.REQUEST_PARAM_NAME] = sticky_duration.as_millis() else: path += f"?{StickyEvent.REQUEST_PARAM_NAME}={sticky_duration.as_millis()}" channel = self.make_request("PUT", path, body, self.user1_access_token) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None def test_delayed_sticky_event_with_conflicting_sticky_durations(self) -> None: """Test that delayed event requests reject having a different sticky duration in the query parameter & request body.""" path = self.helper.build_delayed_event_request_path( room_id=self.room_id, event_type=_EVENT_TYPE, ) body = self.helper.build_delayed_event_request_body( delay=Duration(milliseconds=900), content={"body": "sticky"}, ) path += f"?{StickyEvent.REQUEST_PARAM_NAME}=10000" body[StickyEvent.REQUEST_PARAM_NAME] = 20000 channel = self.make_request("PUT", path, body, self.user1_access_token) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) class DelayedEventsWorkerTestCase( BaseMultiWorkerStreamTestCase, DelayedEventsHelperMixin ): """Tests delayed events when some of the work happens off the main process.""" servlets = [ admin.register_servlets, delayed_events.register_servlets, login.register_servlets, room.register_servlets, sync.register_servlets, ] def default_config(self) -> JsonDict: config = super().default_config() config["max_event_delay_duration"] = "24h" return config def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.user_id = self.register_user("user", "pass") self.access_token = self.login("user", "pass") self.room_id = self.helper.create_room_as(self.user_id, tok=self.access_token) def test_delayed_event_scheduled_on_worker_is_sent(self) -> None: """Test that a delayed event scheduled on a worker is sent by the main process.""" worker_hs = self.make_worker_hs("synapse.app.generic_worker") worker_site = self._hs_to_site[worker_hs] channel = make_request( self.reactor, worker_site, "PUT", self.helper.build_delayed_event_request_path( room_id=self.room_id, event_type=_EVENT_TYPE, ), self.helper.build_delayed_event_request_body( delay=Duration(milliseconds=900), content={"body": "from a worker"}, ), self.access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None # Advance time enough so the delayed event is sent self.reactor.advance(Duration(seconds=1).as_secs()) self._check_for_delayed_event_in_sync(self.access_token, delay_id, True) def test_delayed_events_are_cancelled_on_deactivation_from_worker(self) -> None: worker_hs = self.make_worker_hs("synapse.app.generic_worker") store = self.hs.get_datastores().main handler = self.hs.get_delayed_events_handler() channel = self._make_delayed_event_request( room_id=self.room_id, delay=Duration(milliseconds=900), event_type=_EVENT_TYPE, content={}, access_token=self.access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertIsNotNone(handler._next_delayed_event_call) # Deactivate the account on the worker. The main process, which is the # one to send delayed events, must get to know about it. self.get_success( worker_hs.get_deactivate_account_handler().deactivate_account( self.user_id, erase_data=False, requester=create_requester(self.user_id), ) ) self.assertListEqual( [], self.get_success(store.get_all_delayed_events_for_user("user")) ) self.assertIsNone(handler._next_delayed_event_call)