diff --git a/synapse/federation/sender/__init__.py b/synapse/federation/sender/__init__.py index f5938b20c4..18f5aab102 100644 --- a/synapse/federation/sender/__init__.py +++ b/synapse/federation/sender/__init__.py @@ -181,6 +181,7 @@ from synapse.types import ( from synapse.util.clock import Clock from synapse.util.metrics import Measure from synapse.util.retryutils import filter_destinations_by_retry_limiter +from synapse.visibility import filter_events_for_server if TYPE_CHECKING: from synapse.events.presence_router import PresenceRouter @@ -496,7 +497,58 @@ class FederationSender(AbstractFederationSender): return queue def notify_new_server_joined(self, server: str, room_id: str) -> None: - print(f"FEDSENDER: new server joined: server={server} room={room_id}") + # We currently only use this notification for MSC4354: Sticky Events. + if not self.hs.config.experimental.msc4354_enabled: + return + # fire off a processing loop in the background + run_as_background_process( + "process_new_server_joined_over_federation", + self.server_name, + self._process_new_server_joined_over_federation, + server, + room_id, + ) + + async def _process_new_server_joined_over_federation( + self, new_server: str, room_id: str + ) -> None: + sticky_event_ids = await self.store.get_sticky_event_ids_sent_by_self( + room_id, + 0, + ) + sticky_events = await self.store.get_events_as_list(sticky_event_ids) + + # We must not send events that are outliers / lack a stream ordering, else we won't be able to + # satisfy /get_missing_events requests + sticky_events = [ + ev + for ev in sticky_events + if ev.internal_metadata.stream_ordering is not None + and not ev.internal_metadata.is_outlier() + ] + # order by stream ordering so we present things in the right timeline order on the receiver + sticky_events = sorted( + sticky_events, + key=lambda ev: ev.internal_metadata.stream_ordering + or 0, # not possible to be 0 + ) + + sticky_events = await filter_events_for_server( + self._storage_controllers, + new_server, + self.server_name, + sticky_events, + redact=False, + filter_out_erased_senders=True, + filter_out_remote_partial_state_events=True, + ) + if sticky_events: + logger.info("sending %d sticky events to newly joined server %s in room %s", len(sticky_events), new_server, room_id) + # we don't track that we sent up to this stream position since it won't make any difference + # since notify_new_server_joined is only called initially. + await self._transaction_manager.send_new_transaction( + new_server, sticky_events, [] + ) def notify_new_events(self, max_token: RoomStreamToken) -> None: """This gets called when we have some new events we might want to diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 41fb3076c3..be4d173e70 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -67,6 +67,7 @@ from synapse.events import EventBase from synapse.events.snapshot import EventContext, UnpersistedEventContextBase from synapse.events.validator import EventValidator from synapse.federation.federation_client import InvalidResponseError +from synapse.federation.federation_server import _INBOUND_EVENT_HANDLING_LOCK_NAME from synapse.handlers.pagination import PURGE_PAGINATION_LOCK_NAME from synapse.http.servlet import assert_params_in_dict from synapse.logging.context import nested_logging_context @@ -75,6 +76,7 @@ from synapse.metrics import SERVER_NAME_LABEL from synapse.metrics.background_process_metrics import run_as_background_process from synapse.module_api import NOT_SPAM from synapse.storage.databases.main.events_worker import EventRedactBehaviour +from synapse.storage.databases.main.lock import Lock from synapse.storage.invite_rule import InviteRule from synapse.types import JsonDict, StrCollection, get_domain_from_id from synapse.types.state import StateFilter @@ -647,125 +649,158 @@ class FederationHandler: except ValueError: pass + lock: Optional[Lock] = None async with self._is_partial_state_room_linearizer.queue(room_id): - already_partial_state_room = await self.store.is_partial_state_room( - room_id - ) - - ret = await self.federation_client.send_join( - host_list, - event, - room_version_obj, - # Perform a full join when we are already in the room and it is a - # full state room, since we are not allowed to persist a partial - # state join event in a full state room. In the future, we could - # optimize this by always performing a partial state join and - # computing the state ourselves or retrieving it from the remote - # homeserver if necessary. - # - # There's a race where we leave the room, then perform a full join - # anyway. This should end up being fast anyway, since we would - # already have the full room state and auth chain persisted. - partial_state=not is_host_joined or already_partial_state_room, - ) - - event = ret.event - origin = ret.origin - state = ret.state - auth_chain = ret.auth_chain - auth_chain.sort(key=lambda e: e.depth) - - logger.debug("do_invite_join auth_chain: %s", auth_chain) - logger.debug("do_invite_join state: %s", state) - - logger.debug("do_invite_join event: %s", event) - - # if this is the first time we've joined this room, it's time to add - # a row to `rooms` with the correct room version. If there's already a - # row there, we should override it, since it may have been populated - # based on an invite request which lied about the room version. - # - # federation_client.send_join has already checked that the room - # version in the received create event is the same as room_version_obj, - # so we can rely on it now. - # - await self.store.upsert_room_on_join( - room_id=room_id, - room_version=room_version_obj, - state_events=state, - ) - - if ret.partial_state and not already_partial_state_room: - # Mark the room as having partial state. - # The background process is responsible for unmarking this flag, - # even if the join fails. - # TODO(faster_joins): - # We may want to reset the partial state info if it's from an - # old, failed partial state join. - # https://github.com/matrix-org/synapse/issues/13000 - - # FIXME: Ideally, we would store the full stream token here - # not just the minimum stream ID, so that we can compute an - # accurate list of device changes when un-partial-ing the - # room. The only side effect of this is that we may send - # extra unecessary device list outbound pokes through - # federation, which is harmless. - device_lists_stream_id = self.store.get_device_stream_token().stream - - await self.store.store_partial_state_room( - room_id=room_id, - servers=ret.servers_in_room, - device_lists_stream_id=device_lists_stream_id, - joined_via=origin, - ) - try: - max_stream_id = ( - await self._federation_event_handler.process_remote_join( - origin, - room_id, - auth_chain, - state, - event, - room_version_obj, - partial_state=ret.partial_state, + # MSC4354: Sticky Events causes existing servers in the room to send sticky events + # to the newly joined server as soon as they realise the new server is in the room. + # If they do this before we've persisted the /send_join response we will be unable to + # process those PDUs. Therefore, we take a lock out now for this room, and release it + # once we have processed the /send_join response, to buffer up these inbound messages. + # This may be useful to do even without MSC4354, but it's gated behind an + # experimental flag check to reduce the chance of this having unintended side-effects + # e.g accidental deadlocks. Once we're confident of this behaviour, we can probably + # drop the flag check. We take the lock AFTER we have been queued by the linearizer + # else we would just hold the lock for no reason whilst in the queue: we want to hold + # the lock for the smallest amount of time possible. + if self.config.experimental.msc4354_enabled: + lock = await self.store.try_acquire_lock( + _INBOUND_EVENT_HANDLING_LOCK_NAME, room_id ) - ) - except PartialStateConflictError: - # This should be impossible, since we hold the lock on the room's - # partial statedness. - logger.error( - "Room %s was un-partial stated while processing remote join.", - room_id, - ) - raise - else: - # Record the join event id for future use (when we finish the full - # join). We have to do this after persisting the event to keep - # foreign key constraints intact. - if ret.partial_state and not already_partial_state_room: - # TODO(faster_joins): - # We may want to reset the partial state info if it's from - # an old, failed partial state join. - # https://github.com/matrix-org/synapse/issues/13000 - await self.store.write_partial_state_rooms_join_event_id( - room_id, event.event_id - ) - finally: - # Always kick off the background process that asynchronously fetches - # state for the room. - # If the join failed, the background process is responsible for - # cleaning up — including unmarking the room as a partial state - # room. - if ret.partial_state: - # Kick off the process of asynchronously fetching the state for - # this room. - self._start_partial_state_room_sync( - initial_destination=origin, - other_destinations=ret.servers_in_room, + # Insert the room into the rooms table now so we can process potential incoming + # /send transactions enough to be able to insert into the federation staging + # area. We won't process the staging area until we release the lock above. + await self.store.upsert_room_on_join( room_id=room_id, + room_version=room_version_obj, + state_events=None, ) + already_partial_state_room = await self.store.is_partial_state_room( + room_id + ) + + ret = await self.federation_client.send_join( + host_list, + event, + room_version_obj, + # Perform a full join when we are already in the room and it is a + # full state room, since we are not allowed to persist a partial + # state join event in a full state room. In the future, we could + # optimize this by always performing a partial state join and + # computing the state ourselves or retrieving it from the remote + # homeserver if necessary. + # + # There's a race where we leave the room, then perform a full join + # anyway. This should end up being fast anyway, since we would + # already have the full room state and auth chain persisted. + partial_state=not is_host_joined or already_partial_state_room, + ) + + event = ret.event + origin = ret.origin + state = ret.state + auth_chain = ret.auth_chain + auth_chain.sort(key=lambda e: e.depth) + + logger.debug("do_invite_join auth_chain: %s", auth_chain) + logger.debug("do_invite_join state: %s", state) + + logger.debug("do_invite_join event: %s", event) + + # if this is the first time we've joined this room, it's time to add + # a row to `rooms` with the correct room version. If there's already a + # row there, we should override it, since it may have been populated + # based on an invite request which lied about the room version. + # + # federation_client.send_join has already checked that the room + # version in the received create event is the same as room_version_obj, + # so we can rely on it now. + # + await self.store.upsert_room_on_join( + room_id=room_id, + room_version=room_version_obj, + state_events=state, + ) + + if ret.partial_state and not already_partial_state_room: + # Mark the room as having partial state. + # The background process is responsible for unmarking this flag, + # even if the join fails. + # TODO(faster_joins): + # We may want to reset the partial state info if it's from an + # old, failed partial state join. + # https://github.com/matrix-org/synapse/issues/13000 + + # FIXME: Ideally, we would store the full stream token here + # not just the minimum stream ID, so that we can compute an + # accurate list of device changes when un-partial-ing the + # room. The only side effect of this is that we may send + # extra unecessary device list outbound pokes through + # federation, which is harmless. + device_lists_stream_id = ( + self.store.get_device_stream_token().stream + ) + + await self.store.store_partial_state_room( + room_id=room_id, + servers=ret.servers_in_room, + device_lists_stream_id=device_lists_stream_id, + joined_via=origin, + ) + + try: + max_stream_id = ( + await self._federation_event_handler.process_remote_join( + origin, + room_id, + auth_chain, + state, + event, + room_version_obj, + partial_state=ret.partial_state, + ) + ) + except PartialStateConflictError: + # This should be impossible, since we hold the lock on the room's + # partial statedness. + logger.error( + "Room %s was un-partial stated while processing remote join.", + room_id, + ) + raise + else: + # Record the join event id for future use (when we finish the full + # join). We have to do this after persisting the event to keep + # foreign key constraints intact. + if ret.partial_state and not already_partial_state_room: + # TODO(faster_joins): + # We may want to reset the partial state info if it's from + # an old, failed partial state join. + # https://github.com/matrix-org/synapse/issues/13000 + await self.store.write_partial_state_rooms_join_event_id( + room_id, event.event_id + ) + finally: + # Always kick off the background process that asynchronously fetches + # state for the room. + # If the join failed, the background process is responsible for + # cleaning up — including unmarking the room as a partial state + # room. + if ret.partial_state: + # Kick off the process of asynchronously fetching the state for + # this room. + self._start_partial_state_room_sync( + initial_destination=origin, + other_destinations=ret.servers_in_room, + room_id=room_id, + ) + finally: + # allow inbound events which happened during the join to be processed. + # Also ensures we release the lock on unexpected errors e.g db errors from + # upsert_room_on_join or network errors from send_join. + if lock: + await lock.release() # We wait here until this instance has seen the events come down # replication (if we're using replication) as the below uses caches. await self._replication.wait_for_stream_position( diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index f12e3f3ede..77008f1607 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -1187,7 +1187,9 @@ class PersistEventsStore: ) if self.msc4354_sticky_events: - self.store.insert_sticky_events_txn(txn, events_and_contexts) + self.store.insert_sticky_events_txn( + txn, [ev for ev, _ in events_and_contexts] + ) for ev, _ in events_and_contexts: if ev.type == "m.room.member" and ev.membership == "join": print(f"GOT JOIN FOR {ev.state_key}") @@ -2658,6 +2660,11 @@ class PersistEventsStore: # event isn't an outlier any more. self._update_backward_extremeties(txn, [event]) + if self.msc4354_sticky_events and event.sticky_duration(): + # The de-outliered event is sticky. Update the sticky events table to ensure + # we delivery this down /sync. + self.store.insert_sticky_events_txn(txn, [event]) + return [ec for ec in events_and_contexts if ec[0] not in to_remove] def _store_event_txn( diff --git a/synapse/storage/databases/main/room.py b/synapse/storage/databases/main/room.py index 9f03c084a5..d67b018ee7 100644 --- a/synapse/storage/databases/main/room.py +++ b/synapse/storage/databases/main/room.py @@ -2460,7 +2460,10 @@ class RoomStore(RoomBackgroundUpdateStore, RoomWorkerStore): self._instance_name = hs.get_instance_name() async def upsert_room_on_join( - self, room_id: str, room_version: RoomVersion, state_events: List[EventBase] + self, + room_id: str, + room_version: RoomVersion, + state_events: Optional[List[EventBase]], ) -> None: """Ensure that the room is stored in the table @@ -2472,36 +2475,46 @@ class RoomStore(RoomBackgroundUpdateStore, RoomWorkerStore): # mark the room as having an auth chain cover index. has_auth_chain_index = await self.has_auth_chain_index(room_id) - create_event = None - for e in state_events: - if (e.type, e.state_key) == (EventTypes.Create, ""): - create_event = e - break + # We may want to insert a row into the rooms table BEFORE having the state events in the + # room, in order to correctly handle the race condition where the /send_join is processed + # remotely which causes remote servers to send us events before we've processed the /send_join + # response. Therefore, we allow state_events (and thus the creator column) to be optional. + # When we get the /send_join response, we'll patch this up. + room_creator: Optional[str] = None + if state_events: + create_event = None + for e in state_events: + if (e.type, e.state_key) == (EventTypes.Create, ""): + create_event = e + break - if create_event is None: - # If the state doesn't have a create event then the room is - # invalid, and it would fail auth checks anyway. - raise StoreError(400, "No create event in state") - - # Before MSC2175, the room creator was a separate field. - if not room_version.implicit_room_creator: - room_creator = create_event.content.get(EventContentFields.ROOM_CREATOR) - - if not isinstance(room_creator, str): - # If the create event does not have a creator then the room is + if create_event is None: + # If the state doesn't have a create event then the room is # invalid, and it would fail auth checks anyway. - raise StoreError(400, "No creator defined on the create event") - else: - room_creator = create_event.sender + raise StoreError(400, "No create event in state") + + # Before MSC2175, the room creator was a separate field. + if not room_version.implicit_room_creator: + room_creator = create_event.content.get(EventContentFields.ROOM_CREATOR) + + if not isinstance(room_creator, str): + # If the create event does not have a creator then the room is + # invalid, and it would fail auth checks anyway. + raise StoreError(400, "No creator defined on the create event") + else: + room_creator = create_event.sender + + update_with = {"room_version": room_version.identifier} + if room_creator: + update_with["creator"] = room_creator await self.db_pool.simple_upsert( desc="upsert_room_on_join", table="rooms", keyvalues={"room_id": room_id}, - values={"room_version": room_version.identifier}, + values=update_with, insertion_values={ "is_public": False, - "creator": room_creator, "has_auth_chain_index": has_auth_chain_index, }, ) diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index 077518cfbb..2578aad3e1 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/synapse/storage/databases/main/sticky_events.py @@ -212,7 +212,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor async def get_sticky_event_ids_sent_by_self( self, room_id: str, from_stream_pos: int ) -> List[str]: - """Get sticky event IDs which have been sent by users on this homeserver. + """Get unexpired sticky event IDs which have been sent by users on this homeserver. Used when sending sticky events eagerly to newly joined servers, or when catching up over federation. @@ -284,12 +284,12 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor def insert_sticky_events_txn( self, txn: LoggingTransaction, - events_and_contexts: List[EventPersistencePair], + events: List[EventBase], ) -> None: now_ms = self._now() # event, expires_at, stream_id sticky_events: List[Tuple[EventBase, int, int]] = [] - for ev, _ in events_and_contexts: + for ev in events: # MSC: Note: policy servers and other similar antispam techniques still apply to these events. if ev.internal_metadata.policy_server_spammy: continue