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 <oliverw@matrix.org>

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.

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.
 

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,

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.

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
 

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:

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 <server> <room_id>
+    """
+
+    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,
 )
 

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.
 

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]:

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

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,

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"],
+        )

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: