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: