From 40acb9702294516e3c5e37a0ac0345b08bde1c2b Mon Sep 17 00:00:00 2001 From: Andrew Ferrazzutti Date: Mon, 12 Oct 2026 05:33:13 -0400 Subject: [PATCH] MSC4140: drop delayed state interference (#19495) No longer cancel a user's delayed state events by a more recent state event sent from a different user. This behaviour has been delegated to an "alternative" in the delayed event MSC for some time now, and is not known to be used by any module other than experimental iterations of MatrixRTC. See also: https://github.com/matrix-org/complement/pull/845 Opens: https://github.com/element-hq/synapse/issues/20311 --------- Co-authored-by: Olivier 'reivilibre' Co-authored-by: Paul Chobert --- changelog.d/19495.removal | 1 + changelog.d/20316.bugfix | 1 - synapse/handlers/delayed_events.py | 223 +----------------- .../storage/databases/main/delayed_events.py | 85 ------- synapse/storage/schema/__init__.py | 7 +- tests/rest/client/test_delayed_events.py | 86 ------- 6 files changed, 8 insertions(+), 395 deletions(-) create mode 100644 changelog.d/19495.removal delete mode 100644 changelog.d/20316.bugfix diff --git a/changelog.d/19495.removal b/changelog.d/19495.removal new file mode 100644 index 0000000000..06e53d3969 --- /dev/null +++ b/changelog.d/19495.removal @@ -0,0 +1 @@ +[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): No longer cancel a user's delayed state events due to a more recent state event sent from a different user. diff --git a/changelog.d/20316.bugfix b/changelog.d/20316.bugfix deleted file mode 100644 index 3c49f942a9..0000000000 --- a/changelog.d/20316.bugfix +++ /dev/null @@ -1 +0,0 @@ -Only run delayed event background jobs when a new room event is processed. diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index f652f9b429..4ee5c89b52 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -25,7 +25,6 @@ from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME from synapse.http.site import SynapseRequest from synapse.logging.context import make_deferred_yieldable from synapse.logging.opentracing import set_tag -from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions from synapse.replication.http.delayed_events import ( ReplicationAddedDelayedEventRestServlet, ReplicationCancelDelayedEventsForUserRestServlet, @@ -38,9 +37,7 @@ from synapse.storage.databases.main.delayed_events import ( StateKey, Timestamp, ) -from synapse.storage.databases.main.state_deltas import StateDelta from synapse.types import ( - Absent, JsonDict, Requester, RoomID, @@ -49,7 +46,6 @@ from synapse.types import ( ) from synapse.util.duration import Duration from synapse.util.events import generate_fake_event_id -from synapse.util.metrics import Measure if TYPE_CHECKING: from synapse.server import HomeServer @@ -81,28 +77,11 @@ class DelayedEventsHandler: self._next_delayed_event_call: Optional[IDelayedCall] = None - # The current position in the current_state_delta stream - self._event_pos: int | None = None - - # Guard to ensure we only process event deltas one at a time - self._event_processing = False - if hs.config.worker.worker_app is None: self._repl_client = None async def _schedule_db_events() -> None: - # We kick this off to pick up outstanding work from before the last restart. - # Block until we're up to date. - await self._unsafe_process_new_event() - hs.get_notifier().add_replication_callback(self.notify_new_event) - # Kick off again (without blocking) to catch any missed notifications - # that may have fired before the callback was added. - self._clock.call_later( - Duration(seconds=0), - self.notify_new_event, - ) - - # Now process any delayed events that are due to be sent. + # Process any delayed events that are due to be sent. # # We set `reprocess_events` to True in case any events had been # marked as processed, but had not yet actually been sent, @@ -139,206 +118,6 @@ class DelayedEventsHandler: def _is_master(self) -> bool: return self._repl_client is None - def notify_new_event(self) -> None: - """ - Called when there may be more state event deltas to process, - which should cancel pending delayed events for the same state. - """ - if self._event_processing: - return - - # This is called for updates on any stream (typing, receipts, to-device, - # etc), but only new room events can produce the state deltas we - # process. If the room stream hasn't advanced past what we've already - # handled, there's nothing to do, so skip the database round-trips. - if ( - self._event_pos is not None - and self._event_pos >= self._store.get_room_max_stream_ordering() - ): - return - - self._event_processing = True - - async def process() -> None: - try: - await self._unsafe_process_new_event() - finally: - self._event_processing = False - - self.hs.run_as_background_process("delayed_events.notify_new_event", process) - - async def _unsafe_process_new_event(self) -> None: - # We purposefully fetch the current max room stream ordering before - # doing anything else, as it could increment duing processing of state - # deltas. We want to avoid updating `delayed_events_stream_pos` past - # the stream ordering of the state deltas we've processed. Otherwise - # we'll leave gaps in our processing. - room_max_stream_ordering = self._store.get_room_max_stream_ordering() - - # Check that there are actually any delayed events to process. If not, bail early. - delayed_events_count = await self._store.get_count_of_delayed_events() - if delayed_events_count == 0: - # There are no delayed events to process. Update the - # `delayed_events_stream_pos` to the latest `events` stream pos and - # exit early. - self._event_pos = room_max_stream_ordering - - logger.debug( - "No delayed events to process. Updating `delayed_events_stream_pos` to max stream ordering (%s)", - room_max_stream_ordering, - ) - - await self._store.update_delayed_events_stream_pos(room_max_stream_ordering) - - event_processing_positions.labels( - name="delayed_events", **{SERVER_NAME_LABEL: self.server_name} - ).set(room_max_stream_ordering) - - return - - # If self._event_pos is None then means we haven't fetched it from the DB yet - if self._event_pos is None: - self._event_pos = await self._store.get_delayed_events_stream_pos() - if self._event_pos > room_max_stream_ordering: - # apparently, we've processed more events than exist in the database! - # this can happen if events are removed with history purge or similar. - logger.warning( - "Event stream ordering appears to have gone backwards (%i -> %i): " - "rewinding delayed events processor", - self._event_pos, - room_max_stream_ordering, - ) - self._event_pos = room_max_stream_ordering - - # Loop round handling deltas until we're up to date - while True: - with Measure( - self._clock, name="delayed_events_delta", server_name=self.server_name - ): - room_max_stream_ordering = self._store.get_room_max_stream_ordering() - if self._event_pos >= room_max_stream_ordering: - return - - logger.debug( - "Processing delayed events %s->%s", - self._event_pos, - room_max_stream_ordering, - ) - ( - max_pos, - deltas, - ) = await self._storage_controllers.state.get_current_state_deltas( - self._event_pos, room_max_stream_ordering - ) - - logger.debug( - "Handling %d state deltas for delayed events processing", - len(deltas), - ) - await self._handle_state_deltas(deltas) - - self._event_pos = max_pos - - # Expose current event processing position to prometheus - event_processing_positions.labels( - name="delayed_events", **{SERVER_NAME_LABEL: self.server_name} - ).set(max_pos) - - await self._store.update_delayed_events_stream_pos(max_pos) - - async def _handle_state_deltas(self, deltas: list[StateDelta]) -> None: - """ - Process current state deltas to cancel other users' pending delayed events - that target the same state. - """ - # Get the senders of each delta's state event (as sender information is - # not currently stored in the `current_state_deltas` table). - event_id_and_sender_dict = await self._store.get_senders_for_event_ids( - [delta.event_id for delta in deltas if delta.event_id is not None] - ) - - # Note: No need to batch as `get_current_state_deltas` will only ever - # return 100 rows at a time. - for delta in deltas: - logger.debug( - "Handling: %r %r, %s", delta.event_type, delta.state_key, delta.event_id - ) - - # `delta.event_id` and `delta.sender` can be `None` in a few valid - # cases (see the docstring of - # `get_current_state_delta_membership_changes_for_user` for details). - if delta.event_id is None: - # TODO: Differentiate between this being caused by a state reset - # which removed a user from a room, or the homeserver - # purposefully having left the room. We can do so by checking - # whether there are any local memberships still left in the - # room. If so, then this is the result of a state reset. - # - # If it is a state reset, we should avoid cancelling new, - # delayed state events due to old state resurfacing. So we - # should skip and log a warning in this case. - # - # If the homeserver has left the room, then we should cancel all - # delayed state events intended for this room, as there is no - # need to try and send a delayed event into a room we've left. - logger.warning( - "Skipping state delta (%r, %r) without corresponding event ID. " - "This can happen if the homeserver has left the room (in which " - "case this can be ignored), or if there has been a state reset " - "which has caused the sender to be kicked out of the room", - delta.event_type, - delta.state_key, - ) - continue - - sender_str = event_id_and_sender_dict.get(delta.event_id, Absent) - if sender_str is None: - # An event exists, but the `sender` field was "null" and Synapse - # incorrectly accepted the event. This is not expected. - logger.error( - "Skipping state delta with event ID '%s' as 'sender' was None. " - "This is unexpected - please report it as a bug!", - delta.event_id, - ) - continue - if sender_str is Absent: - # We have an event ID, but the event was not found in the - # datastore. This can happen if a room, or its history, is - # purged. State deltas related to the room are left behind, but - # the event no longer exists. - # - # As we cannot get the sender of this event, we can't calculate - # whether to cancel delayed events related to this one. So we skip. - logger.debug( - "Skipping state delta with event ID '%s' - the room, or its history, may have been purged", - delta.event_id, - ) - continue - - try: - sender = UserID.from_string(sender_str) - except SynapseError as e: - logger.error( - "Skipping state delta with Matrix User ID '%s' that failed to parse: %s", - sender_str, - e, - ) - continue - - next_send_ts = await self._store.cancel_delayed_state_events( - room_id=delta.room_id, - event_type=delta.event_type, - state_key=delta.state_key, - not_from_localpart=( - sender.localpart - if sender.domain == self._config.server.server_name - else "" - ), - ) - - if self._next_send_ts_changed(next_send_ts): - self._schedule_next_at_or_none(next_send_ts) - async def add( self, requester: Requester, diff --git a/synapse/storage/databases/main/delayed_events.py b/synapse/storage/databases/main/delayed_events.py index 1cd06dc14e..01a3915af6 100644 --- a/synapse/storage/databases/main/delayed_events.py +++ b/synapse/storage/databases/main/delayed_events.py @@ -113,32 +113,6 @@ class DelayedEventsStore(SQLBaseStore): unique=True, ) - async def get_delayed_events_stream_pos(self) -> int: - """ - Gets the stream position of the background process to watch for state events - that target the same piece of state as any pending delayed events. - """ - return await self.db_pool.simple_select_one_onecol( - table="delayed_events_stream_pos", - keyvalues={}, - retcol="stream_id", - desc="get_delayed_events_stream_pos", - ) - - async def update_delayed_events_stream_pos(self, stream_id: int | None) -> None: - """ - Updates the stream position of the background process to watch for state events - that target the same piece of state as any pending delayed events. - - Must only be used by the worker running the background process. - """ - await self.db_pool.simple_update_one( - table="delayed_events_stream_pos", - keyvalues={}, - updatevalues={"stream_id": stream_id}, - desc="update_delayed_events_stream_pos", - ) - async def add_delayed_event( self, *, @@ -302,21 +276,6 @@ class DelayedEventsStore(SQLBaseStore): "restart_delayed_event", restart_delayed_event_txn ) - async def get_count_of_delayed_events(self) -> int: - """Returns the number of pending delayed events in the DB.""" - - def _get_count_of_delayed_events(txn: LoggingTransaction) -> int: - sql = "SELECT count(*) FROM delayed_events" - - txn.execute(sql) - resp = txn.fetchone() - return resp[0] if resp is not None else 0 - - return await self.db_pool.runInteraction( - "get_count_of_delayed_events", - _get_count_of_delayed_events, - ) - async def get_delayed_event_for_user( self, delay_id: str, @@ -577,50 +536,6 @@ class DelayedEventsStore(SQLBaseStore): "cancel_delayed_event", cancel_delayed_event_txn ) - async def cancel_delayed_state_events( - self, - *, - room_id: str, - event_type: str, - state_key: str, - not_from_localpart: str, - ) -> Timestamp | None: - """ - Cancels all matching delayed state events, i.e. remove them as long as they haven't been processed. - - Args: - room_id: The room ID to match against. - event_type: The event type to match against. - state_key: The state key to match against. - not_from_localpart: The localpart of a user whose delayed events to not cancel. - If set to the empty string, any users' delayed events may be cancelled. - - Returns: The send time of the next delayed event to be sent, if any. - """ - - def cancel_delayed_state_events_txn( - txn: LoggingTransaction, - ) -> Timestamp | None: - txn.execute( - """ - DELETE FROM delayed_events - WHERE room_id = ? AND event_type = ? AND state_key = ? - AND user_localpart <> ? - AND NOT is_processed - """, - ( - room_id, - event_type, - state_key, - not_from_localpart, - ), - ) - return self._get_next_delayed_event_send_ts_txn(txn) - - return await self.db_pool.runInteraction( - "cancel_delayed_state_events", cancel_delayed_state_events_txn - ) - async def cancel_delayed_events_for_user( self, user_localpart: str ) -> Timestamp | None: diff --git a/synapse/storage/schema/__init__.py b/synapse/storage/schema/__init__.py index 926121d8e1..424946d8d5 100644 --- a/synapse/storage/schema/__init__.py +++ b/synapse/storage/schema/__init__.py @@ -19,7 +19,7 @@ # # -SCHEMA_VERSION = 94 # remember to update the list below when updating +SCHEMA_VERSION = 95 # remember to update the list below when updating """Represents the expectations made by the codebase about the database schema This should be incremented whenever the codebase changes its requirements on the @@ -177,6 +177,11 @@ Changes in SCHEMA_VERSION = 94 - MSC4242: Add state DAG tables. - MSC4429/MSC4262: Track updates to user profile fields via a new stream. - Add an `inserted_ts` column to the `state_groups_persisting` table. + +Changes in SCHEMA_VERSION = 95 + - MSC4140: Stop using the `delayed_events_stream_pos` table, now that + delayed state events are no longer canceled by more recent state events + sent from a different user. """ diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py index 8405dedcf9..4c66c08da0 100644 --- a/tests/rest/client/test_delayed_events.py +++ b/tests/rest/client/test_delayed_events.py @@ -17,7 +17,6 @@ import sqlite3 from http import HTTPStatus from typing import Literal, overload -from unittest.mock import AsyncMock from parameterized import parameterized @@ -980,91 +979,6 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase): 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_delayed_state_is_cancelled_by_new_state_from_other_user( - self, - ) -> None: - state_key = "to_be_cancelled_by_other_user" - - setter_key = "setter" - 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: "on_timeout", - }, - 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) - - setter_expected = "other_user" - self.helper.send_state( - self.room_id, - _EVENT_TYPE, - { - setter_key: setter_expected, - }, - self.user2_access_token, - state_key=state_key, - ) - 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, - ) - self.assertEqual(setter_expected, content.get(setter_key), content) - - 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) - - def test_new_state_processing_skipped_without_new_room_events(self) -> None: - """Ensure that we only process state deltas (and hit the database) when - new room events have been persisted, rather than on every replication - notification.""" - state_key = "to_be_cancelled_by_other_user" - - channel = self._make_delayed_event_request( - room_id=self.room_id, - delay=Duration(milliseconds=900), - event_type=_EVENT_TYPE, - state_key=state_key, - content={}, - access_token=self.user1_access_token, - ) - self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - self.assertEqual(1, len(self._get_delayed_events())) - - store = self.hs.get_datastores().main - store.get_count_of_delayed_events = AsyncMock(return_value=1) # type: ignore[method-assign] - - # Notifications for other streams (typing, receipts, to-device, etc) - # don't advance the room stream, so shouldn't cause any processing. - self.hs.get_notifier().notify_replication() - self.pump() - store.get_count_of_delayed_events.assert_not_called() - - # A new state event from another user should still be processed, and - # cancel the pending delayed event. - self.helper.send_state( - self.room_id, - _EVENT_TYPE, - {}, - self.user2_access_token, - state_key=state_key, - ) - store.get_count_of_delayed_events.assert_called() - self.assertListEqual([], self._get_delayed_events()) - @parameterized.expand((("client_api", False), ("admin_api", True))) def test_delayed_events_are_cancelled_on_deactivation( self, _name: str, by_admin: bool