From b9b74e88928fc4011081d9a14abf45e3bc731ad5 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Thu, 15 Oct 2026 09:53:02 +0100 Subject: [PATCH] Catch up historical MSC4354 Sticky Events to remote homeservers that are newly joining the room. (#20337) Part of: MSC4354 (Experimental Feature tracking issue #19409) Follows: #20165 (Sticky Event federation backlog + catch-up transactions) The MSC says that we need to tell a newly-joining server about all of our sticky events (yes, even the ones before the server joined the room). This PR introduces the mechanism, building on the 'Sticky Event backlog' mechanism introduced in #20165, to make that happen. Most of the logic is taken from the original Sticky Events PR #18968, but the Sticky Event backlog system did not exist back then. (The benefit of having the persistent Sticky Event backlog is that we don't need to rely on the federation senders not crashing to make sure we get around to sending the Sticky Events out.) - Add NEW_SERVER_JOINED replication command. (This is a cross-worker 'wake-up call' to check for a sticky event backlog) We send it when persisting a state change that causes a new server to be in the room. We handle it by waking up federation senders for the destination, including checking the sticky event backlog tables. - When a server newly joins a room, we insert a sticky event backlog row for that server, if there are any Sticky Events needing to be sent. --------- Signed-off-by: Olivier 'reivilibre --- changelog.d/20337.feature | 1 + docs/tcp_replication.md | 4 + synapse/federation/send_queue.py | 5 ++ synapse/federation/sender/__init__.py | 23 ++++++ .../sender/per_destination_queue.py | 8 ++ synapse/notifier.py | 7 ++ synapse/replication/tcp/commands.py | 29 +++++++ synapse/replication/tcp/handler.py | 10 +++ synapse/storage/controllers/persist_events.py | 82 +++++++++++++++++++ synapse/storage/databases/main/events.py | 14 ++++ .../storage/databases/main/sticky_events.py | 60 ++++++++++++++ tests/federation/test_federation_catch_up.py | 26 ++++++ .../test_federation_sender_shard.py | 75 ++++++++++++++++- 13 files changed, 343 insertions(+), 1 deletion(-) create mode 100644 changelog.d/20337.feature diff --git a/changelog.d/20337.feature b/changelog.d/20337.feature new file mode 100644 index 0000000000..b2050b8548 --- /dev/null +++ b/changelog.d/20337.feature @@ -0,0 +1 @@ +Catch up historical [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) to remote homeservers that are newly joining the room. \ No newline at end of file diff --git a/docs/tcp_replication.md b/docs/tcp_replication.md index c3b8c76609..789879ae14 100644 --- a/docs/tcp_replication.md +++ b/docs/tcp_replication.md @@ -225,6 +225,10 @@ Asks the server for the current position of all streams. Inform other processes that a remote server may have come back online. +### NEW_SERVER_JOINED (S, C) + + Informs other processes that a remote server has just joined a room. + See `synapse/replication/tcp/commands.py` for a detailed description and the format of each command. diff --git a/synapse/federation/send_queue.py b/synapse/federation/send_queue.py index 4a6d155217..d49f33179a 100644 --- a/synapse/federation/send_queue.py +++ b/synapse/federation/send_queue.py @@ -212,6 +212,11 @@ class FederationRemoteSendQueue(AbstractFederationSender): # This should never get called. raise NotImplementedError() + def notify_new_server_joined(self, *, server: str, room_id: str) -> None: + """As per FederationSender""" + # This should never get called. + raise NotImplementedError() + def build_and_send_edu( self, destination: str, diff --git a/synapse/federation/sender/__init__.py b/synapse/federation/sender/__init__.py index 9b48bf33fb..a218ca74cb 100644 --- a/synapse/federation/sender/__init__.py +++ b/synapse/federation/sender/__init__.py @@ -241,6 +241,13 @@ class AbstractFederationSender(metaclass=abc.ABCMeta): """ raise NotImplementedError() + @abc.abstractmethod + def notify_new_server_joined(self, *, server: str, room_id: str) -> None: + """This gets called when we a new server has joined a room. We might + want to send out some events to this server. + """ + raise NotImplementedError() + @abc.abstractmethod async def send_read_receipt(self, receipt: ReadReceipt) -> None: """Send a RR to any other servers in the room @@ -503,6 +510,22 @@ class FederationSender(AbstractFederationSender): self._per_destination_queues[destination] = queue return queue + def notify_new_server_joined(self, *, server: str, room_id: str) -> None: + # We currently only use this notification for MSC4354: Sticky Events. + if not self.hs.config.experimental.msc4354_enabled: + return + + if not self._federation_shard_config.should_handle(self._instance_name, server): + # Another federation sender instance is responsible for this destination + return + + queue = self._get_per_destination_queue(server) + if not queue: + # Destination not allowed + return + + queue.notify_sticky_event_backlog() + def notify_new_events(self, max_token: RoomStreamToken) -> None: """This gets called when we have some new events we might want to send out to other servers. diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index 02e69bf778..605a0b1b50 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -392,6 +392,14 @@ class PerDestinationQueue: self._new_data_to_send = True + def notify_sticky_event_backlog(self) -> None: + """Marks that there may be a new backlog of sticky events to send to + this destination (e.g. because it newly joined a room containing sticky + events), and tries to start a new transaction to send them. + """ + self._sticky_event_backlog_tracker.notify_potential_new_backlog() + self.attempt_new_transaction() + def attempt_new_transaction(self) -> None: """Try to start a new transaction to this destination diff --git a/synapse/notifier.py b/synapse/notifier.py index be24cf4058..40a4a1a621 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -1012,6 +1012,13 @@ class Notifier: # that any in flight requests can be immediately retried. self._federation_client.wake_destination(server) + def notify_new_server_joined(self, *, server: str, room_id: str) -> None: + # Inform the federation_sender that it may need to send events to the new server. + if self.federation_sender: + self.federation_sender.notify_new_server_joined( + server=server, room_id=room_id + ) + def add_lock_released_callback( self, callback: Callable[[str, str, str], None] ) -> None: diff --git a/synapse/replication/tcp/commands.py b/synapse/replication/tcp/commands.py index 0c85fb36fc..394b8a224d 100644 --- a/synapse/replication/tcp/commands.py +++ b/synapse/replication/tcp/commands.py @@ -462,6 +462,32 @@ class RemoteServerUpCommand(_SimpleCommand): NAME = "REMOTE_SERVER_UP" +class NewServerJoinedCommand(Command): + """Sent when a worker has detected that a new remote server has joined a room. + + Format:: + + NEW_SERVER_JOINED + """ + + NAME = "NEW_SERVER_JOINED" + __slots__ = ["server", "room_id"] + + def __init__(self, *, server: str, room_id: str): + self.server = server + self.room_id = room_id + + @classmethod + def from_line( + cls: type["NewServerJoinedCommand"], line: str + ) -> "NewServerJoinedCommand": + server, room_id = line.split(" ", 1) + return cls(server=server, room_id=room_id) + + def to_line(self) -> str: + return "%s %s" % (self.server, self.room_id) + + class LockReleasedCommand(Command): """Sent to inform other instances that a given lock has been dropped. @@ -529,6 +555,7 @@ _COMMANDS: tuple[type[Command], ...] = ( FederationAckCommand, UserIpCommand, RemoteServerUpCommand, + NewServerJoinedCommand, ClearUserSyncsCommand, LockReleasedCommand, NewActiveTaskCommand, @@ -546,6 +573,7 @@ VALID_SERVER_COMMANDS = ( ErrorCommand.NAME, PingCommand.NAME, RemoteServerUpCommand.NAME, + NewServerJoinedCommand.NAME, LockReleasedCommand.NAME, ) @@ -560,6 +588,7 @@ VALID_CLIENT_COMMANDS = ( UserIpCommand.NAME, ErrorCommand.NAME, RemoteServerUpCommand.NAME, + NewServerJoinedCommand.NAME, LockReleasedCommand.NAME, ) diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py index b63902b3ec..1c5bff57b0 100644 --- a/synapse/replication/tcp/handler.py +++ b/synapse/replication/tcp/handler.py @@ -41,6 +41,7 @@ from synapse.replication.tcp.commands import ( FederationAckCommand, LockReleasedCommand, NewActiveTaskCommand, + NewServerJoinedCommand, PositionCommand, RdataCommand, RemoteServerUpCommand, @@ -757,6 +758,12 @@ class ReplicationCommandHandler: """Called when get a new REMOTE_SERVER_UP command.""" self._notifier.notify_remote_server_up(cmd.data) + def on_NEW_SERVER_JOINED( + self, conn: IReplicationConnection, cmd: NewServerJoinedCommand + ) -> None: + """Called when get a new NEW_SERVER_JOINED command.""" + self._notifier.notify_new_server_joined(server=cmd.server, room_id=cmd.room_id) + def on_LOCK_RELEASED( self, conn: IReplicationConnection, cmd: LockReleasedCommand ) -> None: @@ -886,6 +893,9 @@ class ReplicationCommandHandler: def send_remote_server_up(self, server: str) -> None: self.send_command(RemoteServerUpCommand(server)) + def send_new_server_joined(self, *, server: str, room_id: str) -> None: + self.send_command(NewServerJoinedCommand(server=server, room_id=room_id)) + def stream_update(self, stream_name: str, token: int | None, data: Any) -> None: """Called when a new update is available to stream to Redis subscribers. diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 13f958d998..53cf5300bf 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -703,6 +703,31 @@ class EventsPersistenceStorageController: async with self._state_deletion_store.persisting_state_group_references( events_and_contexts ): + newly_joined_servers: set[str] = set() + if self.hs.config.experimental.msc4354_enabled and state_delta_for_room: + # We specifically only consider events in `chunk` to reduce the risk of state rollbacks + # causing servers to appear to repeatedly rejoin rooms. This works because we only + # persist events once, whereas the state delta may unreliably flap between joined members + # on unrelated events. This means we may miss cases where the /first/ join event for a server + # is as a result of a state rollback and not as a result of a new join event. That is fine + # because the chance of that happening is vanishingly rare because the join event would need to be + # persisted without it affecting the current state (e.g there's a concurrent ban for that user) + # which is then revoked concurrently by a later event (e.g the user is unbanned). + # If state resolution were more reliable (in terms of state resets) then we could feasibly only + # consider the events in the state_delta_for_room, but we aren't there yet. + new_event_ids_in_current_state = set( + state_delta_for_room.to_insert.values() + ) + newly_joined_servers = ( + await self._newly_joined_servers_after_state_change( + room_id, + [ + ev + for (ev, _) in chunk + if ev.event_id in new_event_ids_in_current_state + ], + ) + ) await self.persist_events_store._persist_events_and_state_updates( room_id, chunk, @@ -712,10 +737,67 @@ class EventsPersistenceStorageController: inhibit_local_membership_updates=backfilled, new_event_links=new_event_links, new_state_dag_forward_extremities=new_state_dag_extrems, + newly_joined_servers=newly_joined_servers, ) + if newly_joined_servers: + # Notify federation senders after the server has joined, + # so we can start to send historical sticky events. + # + # As per MSC4354: + # > When a new server joins the room, existing servers MUST attempt to **push** all of their own sticky events[^newjoiner]. + # > + # > — https://github.com/matrix-org/matrix-spec-proposals/blob/4ad14b0cd3b09205dcba59e45cbf1cab1e75edf7/proposals/4354-sticky-events.md#L91 + for server_name in newly_joined_servers: + self.hs.get_notifier().notify_new_server_joined( + server=server_name, room_id=room_id + ) + + if self.hs.config.redis.redis_enabled: + self.hs.get_replication_command_handler().send_new_server_joined( + server=server_name, room_id=room_id + ) + return replaced_events + async def _newly_joined_servers_after_state_change( + self, room_id: str, new_events_in_current_state: list[EventBase] + ) -> set[str]: + """ + Returns the set of servers that, after persisting `new_events_in_current_state` into the current state, + will be considered to have newly-joined the given room. + + This function must be called BEFORE the current_state_events table is updated. + + Args: + room_id: The room in question + new_events_in_current_state: A list of events that will become part of the current state, + but have not yet been persisted. + """ + # filter to only join events from other servers. We're obviously joined if we are getting full events + # so needn't consider ourselves. + join_events = [ + ev + for ev in new_events_in_current_state + if ev.type == EventTypes.Member + and ev.is_state() + and not self.is_mine_id(ev.state_key) + and ev.membership == Membership.JOIN + ] + if not join_events: + return set() + + joining_domains = {get_domain_from_id(ev.state_key) for ev in join_events} + + # We know the cache is up to date here because we are the event persister for the room, + # the only writer of current state for the room and so we'd have received our own + # cache invalidations immediately with no replication delay. + joined_domains = await self.main_store.get_current_hosts_in_room(room_id) + + newly_joined_domains = joining_domains.difference(joined_domains) + + return newly_joined_domains + async def _calculate_new_forward_extremities_and_state_delta( self, room_id: str, ev_ctx_rm: list[EventPersistencePair] ) -> tuple[set[str] | None, DeltaState | None]: diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index ced620f4c1..ef2110292d 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -317,6 +317,7 @@ class PersistEventsStore: use_negative_stream_ordering: bool = False, inhibit_local_membership_updates: bool = False, new_state_dag_forward_extremities: set[str] | None = None, + newly_joined_servers: Set[str] = frozenset(), ) -> None: """Persist a set of events alongside updates to the current state and forward extremities tables. @@ -339,6 +340,8 @@ class PersistEventsStore: not affect the current local state. new_state_dag_forward_extremities: A set of event IDs that are the new forward extremities for the state DAG for this room. MSC4242 only. + newly_joined_servers: set of server names that have newly joined this room + as a result of these events. Only calculated with MSC4354 enabled. Returns: Resolves when the events have been persisted @@ -424,6 +427,7 @@ class PersistEventsStore: sliding_sync_table_changes=sliding_sync_table_changes, new_state_dag_forward_extremities=new_state_dag_forward_extremities, sticky_events_to_un_soft_fail=sticky_events_to_un_soft_fail, + newly_joined_servers=newly_joined_servers, ) persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc( len(events_and_contexts) @@ -1090,6 +1094,7 @@ class PersistEventsStore: sliding_sync_table_changes: SlidingSyncTableChanges | None, new_state_dag_forward_extremities: set[str] | None = None, sticky_events_to_un_soft_fail: Set[str] = frozenset(), + newly_joined_servers: Set[str] = frozenset(), ) -> None: """Insert some number of room events into the necessary database tables. @@ -1120,6 +1125,8 @@ class PersistEventsStore: `_calculate_sliding_sync_table_changes(...)`) sticky_events_to_un_soft_fail: Sticky events which will be un-soft-failed when persisting the events. + newly_joined_servers: set of server names that have newly joined this room + as a result of these events. Only calculated with MSC4354 enabled. Raises: PartialStateConflictError: if attempting to persist a partial state event in @@ -1274,6 +1281,13 @@ class PersistEventsStore: txn, sticky_events_to_un_soft_fail ) + # If any servers joined the room, then we should write down the fact that we + # need to send them all historical sticky events + if newly_joined_servers: + self.store.mark_backlogged_sticky_events_for_newly_joined_servers_txn( + txn, room_id=room_id, destinations=newly_joined_servers + ) + # We only update the sliding sync tables for non-backfilled events. self._update_sliding_sync_tables_with_new_persisted_events_txn( txn, room_id, events_and_contexts diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index 26e7bc7000..4f95736709 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/synapse/storage/databases/main/sticky_events.py @@ -1006,6 +1006,66 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor _get_backlogged_sticky_events_for_destination_txn, ) + def mark_backlogged_sticky_events_for_newly_joined_servers_txn( + self, + txn: LoggingTransaction, + room_id: str, + destinations: Set[str], + ) -> None: + """ + For the given `destinations`, update the `destination_rooms_sticky_events_backlog` + table to mark this room as backlogged for those destinations if it contains + any sticky events. + + This lets us remember that we need to send historical sticky events to the server. + """ + + if not destinations: + return + + now_millis = self.clock.time_msec() + + # Find the first unexpired, locally-originating sticky event in the room; + # the newly-joined servers need to be sent everything from there onwards. + txn.execute( + """ + SELECT MIN(stream_id) + FROM sticky_events + WHERE room_id = ? + AND ? < expires_at + -- filter to locally-originating sticky events + AND sender LIKE ? + """, + (room_id, now_millis, user_is_local_like_pattern(self.hs)), + ) + row = txn.fetchone() + # MIN() returns NULL over no rows + assert row is not None + min_sticky_events_stream_id: int | None + (min_sticky_events_stream_id,) = row + + if min_sticky_events_stream_id is None: + # No sticky events in the room to send + return + + txn.execute_batch( + """ + INSERT INTO destination_rooms_sticky_events_backlog AS backlog + (destination, room_id, sticky_events_stream_position) + VALUES (?, ?, ?) + ON CONFLICT (destination, room_id) + DO + UPDATE SET sticky_events_stream_position = EXCLUDED.sticky_events_stream_position + -- Only move the position _backwards_ + -- (Not sure there is a concrete race this fixes, but seems like good hygiene.) + WHERE EXCLUDED.sticky_events_stream_position < backlog.sticky_events_stream_position + """, + [ + (destination, room_id, min_sticky_events_stream_id) + for destination in destinations + ], + ) + async def mark_backlogged_sticky_events_after_catchup_transaction( self, destination: str, diff --git a/tests/federation/test_federation_catch_up.py b/tests/federation/test_federation_catch_up.py index 95a4fa57c1..9174be576e 100644 --- a/tests/federation/test_federation_catch_up.py +++ b/tests/federation/test_federation_catch_up.py @@ -946,3 +946,29 @@ class FederationStickyEventCatchUpTestCase(_FederationCatchUpTestCaseBase): transaction = self.run_transaction(per_dest_queue) assert transaction is not None self.assertEqual([pdu.event_id for pdu in transaction.pdus], [sticky_id]) + + def test_historical_sticky_events_sent_to_newly_joined_server(self) -> None: + """ + Tests that historical sticky events are sent (via Federation `/send`) to a newly-joined server. + """ + self.register_user("u1", "you the one") + u1_token = self.login("u1", "you the one") + room_id = self.helper.create_room_as("u1", tok=u1_token) + self._send_sticky(room_id, "sticky before join", u1_token) + + # (Sanity-check) The federation sender does nothing because the server isn't joined the the room + # Advance so that the federation sender wakes up and has chance to send something (if it was going to). + self.reactor.advance(0) + self.assertEqual(self.pdus, []) + + # host2 joins the room + self.get_success( + event_injection.inject_member_event(self.hs, room_id, "@user:host2", "join") + ) + + # The event is sent to the newly joined server via /send + # (PDUs don't carry an `event_id` in room v3+, so match on content) + self.assertEqual( + [pdu["content"].get("body") for pdu in self.pdus], + ["sticky before join"], + ) diff --git a/tests/replication/test_federation_sender_shard.py b/tests/replication/test_federation_sender_shard.py index 929b1d028e..d15ae70da2 100644 --- a/tests/replication/test_federation_sender_shard.py +++ b/tests/replication/test_federation_sender_shard.py @@ -19,7 +19,7 @@ # # import logging -from unittest.mock import AsyncMock, Mock +from unittest.mock import AsyncMock, Mock, patch from netaddr import IPSet from signedjson.key import ( @@ -34,6 +34,7 @@ from synapse.api.constants import EventTypes, Membership from synapse.api.room_versions import RoomVersion from synapse.crypto.event_signing import add_hashes_and_signatures from synapse.events import EventBase, make_event_from_dict +from synapse.federation.sender.per_destination_queue import PerDestinationQueue from synapse.handlers.typing import TypingWriterHandler from synapse.http.federation.matrix_federation_agent import MatrixFederationAgent from synapse.rest.admin import register_servlets_for_client_rest_resource @@ -45,6 +46,7 @@ from synapse.util.clock import Clock from tests.replication._base import BaseMultiWorkerStreamTestCase from tests.server import get_clock +from tests.unittest import override_config logger = logging.getLogger(__name__) @@ -261,6 +263,77 @@ class FederationSenderTestCase(BaseMultiWorkerStreamTestCase): self.assertTrue(sent_on_1) self.assertTrue(sent_on_2) + @override_config({"experimental_features": {"msc4354_enabled": True}}) + def test_new_server_joined_command_sent_to_federation_sender(self) -> None: + """ + Tests that the NEW_SERVER_JOINED replication command is sent to the federation sender + when a server newly joins a room. + """ + worker_hs = self.make_worker_hs( + "synapse.app.generic_worker", + { + "worker_name": "federation_sender1", + "federation_sender_instances": ["federation_sender1"], + }, + ) + + user = self.register_user("user", "pass") + token = self.login("user", "pass") + + with patch.object( + worker_hs.get_notifier(), "notify_new_server_joined" + ) as mock_notify: + room_id = self.create_room_with_remote_server(user, token) + + mock_notify.assert_called_once_with(server="other_server", room_id=room_id) + + @override_config({"experimental_features": {"msc4354_enabled": True}}) + def test_new_server_joined_handled_by_one_shard(self) -> None: + """ + Tests that the NEW_SERVER_JOINED replication command is handled by exactly one + shard, when sharded federation senders are in use. + """ + shard_config = { + "federation_sender_instances": [ + "federation_sender1", + "federation_sender2", + ], + } + self.make_worker_hs( + "synapse.app.generic_worker", + {"worker_name": "federation_sender1", **shard_config}, + ) + self.make_worker_hs( + "synapse.app.generic_worker", + {"worker_name": "federation_sender2", **shard_config}, + ) + + user = self.register_user("user", "pass") + token = self.login("user", "pass") + + # These destinations are known to hash to different shards. + expected_shards = { + "other_server_0": "federation_sender1", + "other_server_3": "federation_sender2", + } + + with patch.object( + PerDestinationQueue, "notify_sticky_event_backlog", autospec=True + ) as mock_notify: + for server_name, expected_instance in expected_shards.items(): + mock_notify.reset_mock() + + self.create_room_with_remote_server(user, token, server_name) + self.replicate() + + # Assert that exactly one federation sender notified its per-destination queue + # about the potential sticky event backlog from the newly-joining server. + notified = [ + (call.args[0]._instance_name, call.args[0]._destination) + for call in mock_notify.call_args_list + ] + self.assertEqual(notified, [(expected_instance, server_name)]) + def create_fake_event_from_remote_server( self, remote_server_name: str, event_dict: JsonDict, room_version: RoomVersion ) -> EventBase: