From 542554c093b4803ce661ea28b9fac797b9e7c145 Mon Sep 17 00:00:00 2001 From: Kegan Dougal <7190048+kegsay@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:17:08 +0100 Subject: [PATCH] MSC4242: State DAGs (inbound joins) (#20194) Process inbound requests to join a state DAG room. Split out from https://github.com/element-hq/synapse/pull/19425 Part of a series of 5x PRs to land the federation part of MSC4242 (https://github.com/element-hq/synapse/pull/19718, https://github.com/element-hq/synapse/pull/20127, https://github.com/element-hq/synapse/pull/20133, inbound-joins (this PR), inbound-pulls). Has some Complement tests: ``` PASS TestMSC4242JoinLeaveRejoinPublicRoom 9.19s PASS TestMSC4242JoinPublicRoomWithRejectedStateDAGEvents 0.88s ``` ### Pull Request Checklist * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [x] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --------- Co-authored-by: Eric Eastwood Co-authored-by: Eric Eastwood --- changelog.d/20194.misc | 1 + .../conf/workers-shared-extra.yaml.j2 | 1 + scripts-dev/complement.sh | 1 + synapse/api/constants.py | 4 + synapse/app/generic_worker.py | 2 + synapse/event_auth.py | 13 +- synapse/events/builder.py | 14 +- synapse/federation/federation_server.py | 32 + synapse/handlers/federation.py | 24 +- synapse/handlers/federation_event.py | 665 +++++++++++++++++- synapse/storage/controllers/persist_events.py | 49 +- synapse/storage/controllers/state.py | 9 +- synapse/storage/databases/main/rejections.py | 14 + synapse/storage/databases/main/state.py | 6 + tests/federation/test_federation_server.py | 42 +- tests/handlers/test_federation_event.py | 194 ++++- 16 files changed, 1018 insertions(+), 53 deletions(-) create mode 100644 changelog.d/20194.misc diff --git a/changelog.d/20194.misc b/changelog.d/20194.misc new file mode 100644 index 0000000000..40cdf30f80 --- /dev/null +++ b/changelog.d/20194.misc @@ -0,0 +1 @@ +Add support for joining [MSC4242](https://github.com/matrix-org/matrix-spec-proposals/pull/4242) rooms over federation. diff --git a/docker/complement/conf/workers-shared-extra.yaml.j2 b/docker/complement/conf/workers-shared-extra.yaml.j2 index 3cf8615b75..b845370988 100644 --- a/docker/complement/conf/workers-shared-extra.yaml.j2 +++ b/docker/complement/conf/workers-shared-extra.yaml.j2 @@ -145,6 +145,7 @@ experimental_features: msc4306_enabled: true # Sticky Events msc4354_enabled: true + msc4242_enabled: true server_notices: system_mxid_localpart: _server diff --git a/scripts-dev/complement.sh b/scripts-dev/complement.sh index d8e553f446..ec2e26961c 100755 --- a/scripts-dev/complement.sh +++ b/scripts-dev/complement.sh @@ -287,6 +287,7 @@ main() { ./tests/msc4306 ./tests/msc4222 ./tests/msc4429 + ./tests/msc4242 ) # Export the list of test packages as a space-separated environment variable, so other diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 9d3e86a0d4..48c6f95bd1 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -518,3 +518,7 @@ class StateDag: Arbitrary, servers usually never get this high so it's mostly to bound the max size of a single /get_missing_events response. """ + MAX_PREV_STATE_EVENTS: Final = 20 + """ + Max number of prev_state_events allowed on an MSC4242 event as per the MSC. + """ diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py index 159cd44237..6014d328ba 100644 --- a/synapse/app/generic_worker.py +++ b/synapse/app/generic_worker.py @@ -93,6 +93,7 @@ from synapse.storage.databases.main.push_rule import PushRulesWorkerStore from synapse.storage.databases.main.pusher import PusherWorkerStore from synapse.storage.databases.main.receipts import ReceiptsWorkerStore from synapse.storage.databases.main.registration import RegistrationWorkerStore +from synapse.storage.databases.main.rejections import RejectionsStore from synapse.storage.databases.main.relations import RelationsWorkerStore from synapse.storage.databases.main.room import RoomWorkerStore from synapse.storage.databases.main.roommember import RoomMemberWorkerStore @@ -153,6 +154,7 @@ class GenericWorkerStore( EventFederationWorkerStore, EventPushActionsWorkerStore, PurgeEventsStore, + RejectionsStore, StateGroupWorkerStore, SignatureWorkerStore, UserErasureWorkerStore, diff --git a/synapse/event_auth.py b/synapse/event_auth.py index 81a454b544..c2a18889a3 100644 --- a/synapse/event_auth.py +++ b/synapse/event_auth.py @@ -164,6 +164,7 @@ async def check_state_independent_auth_rules( store: _EventSourceStore, event: "EventBase", batched_auth_events: Mapping[str, "EventBase"] | None = None, + batched_rejections: Mapping[str, str] | None = None, ) -> None: """Check that an event complies with auth rules that are independent of room state @@ -175,6 +176,11 @@ async def check_state_independent_auth_rules( event: the event being checked. batched_auth_events: if the event being authed is part of a batch, any events from the same batch that may be necessary to auth the current event + batched_rejections: event ID to rejection reason for any events in `batched_auth_events` + which have been rejected. Events in the batch have not been persisted + yet, so their rejection is recorded neither on the event nor in the database, + and the caller has to tell us. Needed for MSC4242 State DAG rooms, where + rejection cascades through `prev_state_events`. Raises: AuthError if the checks fail @@ -244,11 +250,14 @@ async def check_state_independent_auth_rules( ) # 2.3 If there are entries which were themselves rejected under the checks performed on # receipt of a PDU, reject. - if prev_state_event.rejected_reason is not None: + rejected_reason = prev_state_event.rejected_reason + if rejected_reason is None and batched_rejections is not None: + rejected_reason = batched_rejections.get(prev_state_event.event_id) + if rejected_reason is not None: raise AuthError( 403, f"During auth for event {event.event_id} in room {event.room_id}, event has a " - + f"prev_state_event which is rejected ({prev_state_event.rejected_reason}): " + + f"prev_state_event which is rejected ({rejected_reason}): " + f"{prev_state_event.event_id}", ) diff --git a/synapse/events/builder.py b/synapse/events/builder.py index 742bb82da9..56dc219586 100644 --- a/synapse/events/builder.py +++ b/synapse/events/builder.py @@ -24,7 +24,13 @@ from typing import TYPE_CHECKING, Any import attr from signedjson.types import SigningKey -from synapse.api.constants import MAX_DEPTH, EventTypes, StickyEvent, StickyEventField +from synapse.api.constants import ( + MAX_DEPTH, + EventTypes, + StateDag, + StickyEvent, + StickyEventField, +) from synapse.api.room_versions import ( KNOWN_EVENT_FORMAT_VERSIONS, EventFormatVersions, @@ -148,7 +154,7 @@ class EventBuilder: Only applicable on MSC4242 state DAG rooms. If this is supplied, auth_event_ids must not be specified unless this event is part of a batch such that the builder will be unable to compute the auth_event_ids due to the events not being persisted - yet. + yet. The event IDs will be trimmed to StateDag.MAX_PREV_STATE_EVENTS. Returns: The signed and hashed event. """ @@ -157,6 +163,10 @@ class EventBuilder: assert self.room_version.msc4242_state_dags if self.room_version.msc4242_state_dags: assert prev_state_events is not None + # FIXME: choose more intelligently + prev_state_events = sorted(prev_state_events)[ + : StateDag.MAX_PREV_STATE_EVENTS + ] if self.room_id: state_ids = await self._state.compute_state_after_events( self.room_id, diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 95aff08ddf..06b93dccc1 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -233,12 +233,23 @@ class FederationServer(FederationBase): # We pause a bit so that we don't start handling all rooms at once. await self._clock.sleep(Duration(seconds=random.uniform(0, 0.1))) + async def _assert_not_state_dag_room(self, room_id: str, endpoint: str) -> None: + room_version = await self.store.get_room_version(room_id) + if room_version.msc4242_state_dags: + # Not 5xx else the server will back off from us + raise SynapseError( + 400, + f"{endpoint} is not implemented for MSC4242 state DAG rooms", + errcode=Codes.UNRECOGNIZED, + ) + async def on_backfill_request( self, origin: str, room_id: str, versions: list[str], limit: int ) -> tuple[int, dict[str, Any]]: async with self._server_linearizer.queue((origin, room_id)): origin_host, _ = parse_server_name(origin) await self.check_server_matches_acl(origin_host, room_id) + await self._assert_not_state_dag_room(room_id, "/backfill") pdus = await self.handler.on_backfill_request( origin, room_id, versions, limit @@ -275,6 +286,14 @@ class FederationServer(FederationBase): origin_host, _ = parse_server_name(origin) await self.check_server_matches_acl(origin_host, room_id) + if (await self.store.get_room_version(room_id)).msc4242_state_dags: + # TODO: we currently don't support /timestamp_to_event in MSC4242 rooms. + raise SynapseError( + 404, + "This endpoint is not yet implemented in MSC4242 rooms", + errcode=Codes.NOT_FOUND, + ) + # We only try to fetch data from the local database event_id = await self.store.get_event_id_for_timestamp( room_id, timestamp, direction @@ -654,6 +673,7 @@ class FederationServer(FederationBase): await self._event_auth_handler.assert_host_in_room(room_id, origin) origin_host, _ = parse_server_name(origin) await self.check_server_matches_acl(origin_host, room_id) + await self._assert_not_state_dag_room(room_id, "/state") # we grab the linearizer to protect ourselves from servers which hammer # us. In theory we might already have the response to this query @@ -694,6 +714,7 @@ class FederationServer(FederationBase): await self._event_auth_handler.assert_host_in_room(room_id, origin) origin_host, _ = parse_server_name(origin) await self.check_server_matches_acl(origin_host, room_id) + await self._assert_not_state_dag_room(room_id, "/state_ids") resp = await self._state_ids_resp_cache.wrap( (room_id, event_id), @@ -1154,6 +1175,14 @@ class FederationServer(FederationBase): # Note that get_room_version throws if the room does not exist here. room_version = await self.store.get_room_version(room_id) + if room_version.msc4242_state_dags and membership_type != Membership.JOIN: + # 4xx rather than 5xx, so the caller does not back off from us + raise SynapseError( + 400, + f"/send_{membership_type} is not implemented for MSC4242 state DAG rooms", + errcode=Codes.UNRECOGNIZED, + ) + if await self.store.is_partial_state_room(room_id): # If our server is still only partially joined, we can't give a complete # response to /send_join, /send_knock or /send_leave. @@ -1247,6 +1276,7 @@ class FederationServer(FederationBase): await self._event_auth_handler.assert_host_in_room(room_id, origin) origin_host, _ = parse_server_name(origin) await self.check_server_matches_acl(origin_host, room_id) + await self._assert_not_state_dag_room(room_id, "/event_auth") time_now = self._clock.time_msec() auth_pdus = await self.handler.on_event_auth(event_id, room_id) @@ -1332,6 +1362,8 @@ class FederationServer(FederationBase): origin, room_id, earliest_events, latest_events, limit ) else: + # TODO: currently we only handle /send_join and connected events in /send for state DAG rooms + await self._assert_not_state_dag_room(room_id, "/get_missing_events") missing_events = await self.handler.on_get_missing_events( origin, room_id, earliest_events, latest_events, limit ) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 52efbe92cc..fab47968fa 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -64,7 +64,6 @@ from synapse.api.errors import ( PartialStateConflictError, RequestSendFailed, SynapseError, - UnsupportedRoomVersionError, ) from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion from synapse.crypto.event_signing import compute_event_signature @@ -255,6 +254,10 @@ class FederationHandler: Returns: True if we actually tried to backfill something, otherwise False. """ + if (await self.store.get_room_version(room_id)).msc4242_state_dags: + # TODO: we currently don't support backfilling in MSC4242 rooms. + return False + # Starting the processing time here so we can include the room backfill # linearizer lock queue in the timing processing_start_time = self.clock.time_msec() if record_time else 0 @@ -712,12 +715,6 @@ class FederationHandler: room_id ) - # See related restriction in /createRoom requests in handlers/room.py - if room_version_obj.msc4242_state_dags: - raise UnsupportedRoomVersionError( - "Homeserver does not support this room version over federation" - ) - ret = await self.federation_client.send_join( host_list, event, @@ -741,6 +738,11 @@ class FederationHandler: auth_chain = ret.auth_chain auth_chain.sort(key=lambda e: e.depth) + if ret.state_dag is not None: + # MSC4242 State DAG rooms have no separate state and auth chain: the + # state DAG is both, and everything is derived from it. + state = ret.state_dag + logger.debug("do_invite_join auth_chain: %s", auth_chain) logger.debug("do_invite_join state: %s", state) @@ -1461,6 +1463,14 @@ class FederationHandler: if event.state_key is None: raise SynapseError(400, "The invite event did not have a state key") + if room_version.msc4242_state_dags: + # 4xx rather than 5xx, so the caller does not back off from us + raise SynapseError( + 400, + "/invite is not implemented for MSC4242 state DAG rooms", + errcode=Codes.UNRECOGNIZED, + ) + is_blocked = await self.store.is_room_blocked(event.room_id) if is_blocked: raise SynapseError(403, "This room has been blocked on this server") diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py index 7cbc52d350..1ea0e3524d 100644 --- a/synapse/handlers/federation_event.py +++ b/synapse/handlers/federation_event.py @@ -41,6 +41,7 @@ from synapse.api.constants import ( Membership, RejectedReason, RoomEncryptionAlgorithms, + StateDag, ) from synapse.api.errors import ( AuthError, @@ -61,6 +62,7 @@ from synapse.event_auth import ( validate_event_for_room_version, ) from synapse.events import EventBase +from synapse.events.py_protocol import MSC4242Event, supports_msc4242_state_dag from synapse.events.snapshot import ( EventContext, EventPersistencePair, @@ -522,8 +524,10 @@ class FederationEventHandler: Args: origin: Where the events came from room_id: - auth_events - state + auth_events: The auth chain from the send_join response. Empty for MSC4242 + State DAG rooms, which do not have one. + state: The current state of the room. For MSC4242 State DAG rooms this is the + state DAG, which includes the auth chains for the state. event room_version: The room version we expect this room to have, and will raise if it doesn't match the version in the create event. @@ -555,6 +559,25 @@ class FederationEventHandler: if room_version.identifier != room_version_id: raise SynapseError(400, "Room version mismatch") + # Additional validation is performed on the returned events prior to persisting in MSC4242 rooms. + if room_version.msc4242_state_dags: + # The join event was built by the remote server, not us. We need to make sure it is valid. + # TODO: we almost certainly want to apply this to all room versions. + self._sanity_check_event(event) + if not is_state_dag_connected( + [ + ev + for ev in itertools.chain(state, [event]) + if supports_msc4242_state_dag(ev) + ] + ): + # We should have been given a connected state DAG. If there is a gap in it we + # cannot calculate the state, so the response is invalid and we refuse the join. + raise SynapseError( + 502, + "Unable to join because the remote server passed back a state DAG that is not connected (invalid)", + ) + # persist the auth chain and state events. # # any invalid events here will be marked as rejected, and we'll carry on. @@ -566,9 +589,26 @@ class FederationEventHandler: # signatures right now doesn't mean that we will *never* be able to, so it # is premature to reject them. # - await self._auth_and_persist_outliers( - room_id, itertools.chain(auth_events, state) + has_rejected_events = await self._auth_and_persist_outliers( + room_id, + itertools.chain(auth_events, state), + from_send_join=True, ) + if room_version.msc4242_state_dags and has_rejected_events: + # A conformant server excludes rejected events from the state DAG it sends to + # new joiners, so their presence means the remote server is faulty. We log + # that rather than refusing the join: rejected events are excluded from the + # state DAG forward extremities, so they cannot contribute to the room state + # we calculate, and any event referencing one is rejected in turn. Refusing + # would cost availability without buying any safety. + logger.warning( + "Remote server %s sent a state DAG containing rejected events when we " + "joined %s. This is a bug in that server: rejected events must be " + "excluded from /send_join responses. Joining anyway; the rejected events " + "form no part of the room state.", + origin, + room_id, + ) # and now persist the join event itself. logger.info( @@ -620,6 +660,18 @@ class FederationEventHandler: ), ) ) + elif supports_msc4242_state_dag(event): + # We don't blindly trust the state the remote server claims is current, + # and instead calculate the state before the join ourselves, which we can + # do now that the whole state DAG has been persisted. + state_before_event = ( + await self._state_handler.resolve_state_groups_for_events( + event.room_id, event.prev_state_events + ) + ) + state_ids_before_event = await state_before_event.get_state( + self._state_storage_controller + ) else: state_ids_before_event = { (e.type, e.state_key): e.event_id for e in state @@ -866,6 +918,14 @@ class FederationEventHandler: backfilled: True if this is part of a historical batch of events (inhibits notification to clients, and validation of device keys.) """ + if not events: + return + + # Our callers only ever pass events for a single room. + first_event = next(iter(events)) + room_id = first_event.room_id + is_state_dag_room = first_event.room_version.msc4242_state_dags + set_tag( SynapseTags.FUNC_ARG_PREFIX + "event_ids", str([event.event_id for event in events]), @@ -900,6 +960,7 @@ class FederationEventHandler: new_events: list[EventBase] = [] for event in events: + assert event.room_id == room_id event_id = event.event_id # If we've already seen this event ID... @@ -935,10 +996,50 @@ class FederationEventHandler: # ascending) because one backfill event is likely to be the `prev_event` of # the next event we're going to process. sorted_events = sorted(new_events, key=lambda x: x.depth) + + # Backfilled events are persisted without recalculating the current state, + # which is right for the room DAG but not always for the state DAG. Backfill + # can teach us about state events we have never seen, which then belong on + # the state DAG frontier and so do change the current state. That happens + # when an event's `prev_events` reach state events which its + # `prev_state_events` do not, because the state DAG walk in + # /get_missing_events only ever follows `prev_state_events`. + # This is defence-in-depth: prev_events on honest servers is a superset of the state + # events in that event's prev_state_events, and only when they disagree (e.g. refer to + # an event on a fork) will this occur, but we have to handle the adversarial case too. + # + # Remember the frontier so that we can recalculate the current state if it moves. + state_dag_extremities_before: frozenset[str] | None = None + if ( + backfilled # Non-backfilled pulled events will automatically update the current state, this check is only for /backfill + and sorted_events + and is_state_dag_room + ): + state_dag_extremities_before = ( + await self._store.get_state_dag_extremities(room_id) + ) + for ev in sorted_events: with nested_logging_context(ev.event_id): await self._process_pulled_event(origin, ev, backfilled=backfilled) + if state_dag_extremities_before is not None: + state_dag_extremities_after = ( + await self._store.get_state_dag_extremities(room_id) + ) + if state_dag_extremities_after != state_dag_extremities_before: + # This is unusual because it implies the graph induced by prev_events vs prev_state_events + # has diverged, so warn about it. + logger.warning( + "Backfill from %s moved the state DAG forward extremities of %s from %s " + "to %s, recalculating the current state", + origin, + room_id, + shortstr(state_dag_extremities_before), + shortstr(state_dag_extremities_after), + ) + await self._state_handler.update_current_state(room_id) + # Check if we've already tried to process these events at some point in the # past. We aren't concerned with the expontntial backoff here, just whether it # has failed to be processed before. @@ -1165,6 +1266,15 @@ class FederationEventHandler: if not missing_prevs: return await self._state_handler.compute_event_context(event) + if event.room_version.msc4242_state_dags: + raise FederationError( + "ERROR", + 500, + "Fetching the state at missing prev_events is not implemented for " + "MSC4242 state DAG rooms", + affected=event_id, + ) + logger.info( "_compute_event_context_with_maybe_missing_prevs(event_id=%s): Event in room %s is missing prev_events %s: " "calculating state for a backwards extremity", @@ -1686,12 +1796,16 @@ class FederationEventHandler: @trace async def _auth_and_persist_outliers( - self, room_id: str, events: Iterable[EventBase] - ) -> None: + self, + room_id: str, + events: Iterable[EventBase], + from_send_join: bool = False, + ) -> bool: """Persist a batch of outlier events fetched from remote servers. We first sort the events to make sure that we process each event's auth_events - before the event itself. + before the event itself. For MSC4242 State DAG events this means sorting by + `prev_state_events`, which is also where auth events are calculated from. We then mark the events as outliers, persist them to the database, and, where appropriate (eg, an invite), awake the notifier. @@ -1700,8 +1814,20 @@ class FederationEventHandler: room_id: the room that the events are meant to be in (though this has not yet been checked) events: the events that have been fetched + from_send_join: If True, `events` is the complete state DAG from a /send_join + response, which allows the events to be processed as a batch rather than + one at a time. Only used for MSC4242 State DAG rooms. + Returns: + True if any of the events were rejected. """ event_map = {event.event_id: event for event in events} + if not event_map: + return False + + # There must be an event in events else we would have returned by now + is_state_dag_room = next( + iter(event_map.values()) + ).room_version.msc4242_state_dags event_ids = event_map.keys() set_tag( @@ -1719,34 +1845,87 @@ class FederationEventHandler: # # This is just an optimisation, so it doesn't need to be watertight - the event # persister does another round of deduplication. + has_rejected_events = False seen_remotes = await self._store.have_seen_events(room_id, event_map.keys()) - for s in seen_remotes: - event_map.pop(s, None) + unseen_event_map = dict(event_map) + if seen_remotes: + for s in seen_remotes: + unseen_event_map.pop(s, None) + if is_state_dag_room: + rejected_event_ids = await self._store.has_rejected_event_ids( + seen_remotes + ) + if len(rejected_event_ids) > 0: + has_rejected_events = True + # The copies in `event_map` came off the wire, so they claim to be + # unrejected even for events we rejected when we first saw them, e.g + # during an earlier join of this room. Swap in the persisted copies, + # which carry `rejected_reason`, so that rejection cascades into the + # events in this batch which reference them via `prev_state_events`. + # + # Events rejected during *this* batch are not covered by this: they + # have not been persisted yet, so they are tracked separately and + # passed to the auth rules as `batched_rejections`. + persisted_rejected_events = await self._store.get_events( + rejected_event_ids, + allow_rejected=True, + redact_behaviour=EventRedactBehaviour.as_is, + ) + missing = rejected_event_ids - set(persisted_rejected_events) + if missing: + # We have seen these events and they have rejection rows, so we + # should be able to load them. If we cannot, rejection would + # silently stop cascading, so complain loudly instead. + raise AssertionError( + f"Could not load rejected events {missing} " + + f"out of {rejected_event_ids}" + ) + event_map.update(persisted_rejected_events) # XXX: it might be possible to kick this process off in parallel with fetching # the events. # We need to persist an event's auth events before the event. auth_graph = { - ev.event_id: [e_id for e_id in ev.auth_event_ids() if e_id in event_map] - for ev in event_map.values() + ev.event_id: [ + e_id + for e_id in ( + ev.prev_state_events + if supports_msc4242_state_dag(ev) + else ev.auth_event_ids() + ) + if e_id in unseen_event_map + ] + for ev in unseen_event_map.values() } - sorted_auth_event_ids = sorted_topologically(event_map.keys(), auth_graph) - sorted_auth_events = [event_map[e_id] for e_id in sorted_auth_event_ids] + unseen_sorted_event_ids = sorted_topologically( + unseen_event_map.keys(), auth_graph + ) + unseen_sorted_events = [ + unseen_event_map[e_id] for e_id in unseen_sorted_event_ids + ] logger.info( "Persisting %i remaining outliers: %s", - len(sorted_auth_events), - shortstr(e.event_id for e in sorted_auth_events), + len(unseen_sorted_events), + shortstr(e.event_id for e in unseen_sorted_events), ) + if is_state_dag_room: + has_new_rejected_events = await self._auth_and_persist_state_dag_outliers( + room_id, event_map, unseen_sorted_events, from_send_join + ) + return has_rejected_events or has_new_rejected_events + + events_and_contexts_to_persist: list[EventPersistencePair] = [] + # get all the auth events for all the events in this batch. By now, they should # have been persisted. auth_event_ids = { - aid for event in sorted_auth_events for aid in event.auth_event_ids() + aid for event in unseen_sorted_events for aid in event.auth_event_ids() } auth_map = { ev.event_id: ev - for ev in sorted_auth_events + for ev in unseen_sorted_events if ev.event_id in auth_event_ids } @@ -1759,8 +1938,6 @@ class FederationEventHandler: ) auth_map.update(persisted_events) - events_and_contexts_to_persist: list[EventPersistencePair] = [] - async def prep(event: EventBase) -> None: with nested_logging_context(suffix=event.event_id): auth = [] @@ -1806,7 +1983,7 @@ class FederationEventHandler: events_and_contexts_to_persist.append((event, context)) - for i, event in enumerate(sorted_auth_events): + for i, event in enumerate(unseen_sorted_events): await prep(event) # The above function is typically not async, and so won't yield to @@ -1815,6 +1992,13 @@ class FederationEventHandler: if (i + 1) % 1000 == 0: await self._clock.sleep(Duration(seconds=0)) + if not has_rejected_events: + # Check if any of the persisted events were rejected + has_rejected_events = any( + context.rejected is not None + for _, context in events_and_contexts_to_persist + ) + # Also persist the new event in batches for similar reasons as above. for batch in batch_iter(events_and_contexts_to_persist, 1000): await self.persist_events_and_notify( @@ -1826,6 +2010,351 @@ class FederationEventHandler: backfilled=True, ) + return has_rejected_events + + async def _auth_and_persist_state_dag_outliers( + self, + room_id: str, + event_map: dict[str, EventBase], + sorted_events: list[EventBase], + from_send_join: bool, + ) -> bool: + """Persist a batch of MSC4242 State DAG outlier events. + + Other room versions check that an event is allowed according to the auth events + the event itself lists. MSC4242 rooms instead calculate the auth state before the + event and then determine if the event is allowed. + + This means we normally cannot prep the whole batch and then persist the whole + batch, as the state at an event's `prev_state_events` has to have been persisted + before we can work out that event's auth events. The exception is a /send_join + response, where the entire state DAG is in `event_map` and so the state at each + event can be remembered as we go. + + Params: + room_id: the room that the events are meant to be in (though this has + not yet been checked) + event_map: every event we were given, by event ID. This includes events we + have already persisted and so are not persisting again, because we may + still need them to work out the auth events of an event we have not seen + before. Any of those which were rejected are the persisted copies, + so they carry `rejected_reason`. On /send_join this is the complete + state DAG. + sorted_events: the events being persisted, sorted so that an event's + `prev_state_events` come before it + from_send_join: True if `event_map` is the complete state DAG from a + /send_join response + Returns: + True if any of the events were rejected. + """ + for event in sorted_events: + self._sanity_check_event(event) + + has_rejected_events = False + events_and_contexts_to_persist: list[EventPersistencePair] = [] + event_id_to_state_group: dict[str, int] = {} + state_group_to_state_map: dict[int, StateMap[str]] = {} + # Tracks which events have been processed in causal order (prev_state_events first) + processed_event_map: dict[str, EventBase] = {} + # The events in this batch which we have rejected, and why. Rejection cascades + # through `prev_state_events`, but nothing in the batch is persisted until we have + # processed all of it, so an event we rejected a moment ago still looks accepted + # both in the database and on the in-memory event. We therefore have to remember + # our own decisions and tell the auth rules about them. + rejected_event_id_to_reason_map: dict[str, str] = {} + + async def prev_state_maps(event: MSC4242Event) -> dict[int, StateMap[str]]: + """The state at each of an event's `prev_state_events`, by state group. + + A `prev_state_event` processed earlier in this batch has not been persisted + yet, so the state at it is only in memory. A `prev_state_event` persisted in the past + e.g. from a previous room join are stored in the database. A single event can reference + both, so fill in from the database on a miss and remember it for the rest of + the batch. + """ + missing = [ + prev_state_event_id + for prev_state_event_id in event.prev_state_events + if prev_state_event_id not in event_id_to_state_group + ] + if missing: + # Anything we have not remembered was persisted before this batch, so + # the database knows the state at it. + groups = ( + await self._state_storage_controller.get_state_group_for_events( + missing, + await_full_state=False, + ) + ) + unknown = [ + prev_state_event_id + for prev_state_event_id in missing + if groups.get(prev_state_event_id) is None + ] + if unknown: + # `get_state_group_for_events` claims to raise when it can't + # find a state group, but that only covers lookups which reached the + # database :/ cached entries come back as None. + # We can't calculate the state at `event` without knowing the state groups + # for ALL prev_state_events so bail out if we're missing them. + raise RuntimeError( + f"No state group for prev_state_events {unknown} of " + f"{event.event_id}, so the state before it cannot be calculated" + ) + + group_to_state = ( + await self._state_storage_controller.get_state_for_groups( + set(groups.values()) + ) + ) + # Now populate the in-memory map + for prev_state_event_id, state_group in groups.items(): + event_id_to_state_group[prev_state_event_id] = state_group + state_group_to_state_map[state_group] = group_to_state[state_group] + + return { + event_id_to_state_group[prev_state_event_id]: state_group_to_state_map[ + event_id_to_state_group[prev_state_event_id] + ] + for prev_state_event_id in event.prev_state_events + } + + async def process(event: MSC4242Event) -> EventPersistencePair: + assert supports_msc4242_state_dag(event) + with nested_logging_context(suffix=event.event_id): + # On a /send_join the events in this batch are not persisted until we + # have processed all of them, so the state at an event's + # `prev_state_events` can be spread across memory and the database. + known_prev_state_maps = None + if from_send_join: + known_prev_state_maps = await prev_state_maps(event) + + ( + context, + calculated_auth_event_ids, + ) = await self._calculate_state_dag_context( + event, + known_prev_state_maps=known_prev_state_maps, + event_map=processed_event_map, + ) + event.internal_metadata.calculated_auth_event_ids = ( + calculated_auth_event_ids + ) + event.internal_metadata.outlier = True + + batched_auth_events = None + if from_send_join: + # event_map is the complete state DAG + + # The events being persisted aren't in the database yet, so pull the + # auth events out of `event_map`, which holds the whole state DAG, + # including the events we had already persisted before this join. + missing_auth_event_ids = set(calculated_auth_event_ids).difference( + event_map + ) + if missing_auth_event_ids: + raise SynapseError( + 502, + f"state DAG is missing the calculated auth events {missing_auth_event_ids} " + f"for event {event.event_id}", + ) + calculated_auth_events = { + event_id: event_map[event_id] + for event_id in calculated_auth_event_ids + } + # In order to run auth rule checks we need the events referenced in the event + # (prev_state_events) and the events we calculated (auth_events) so load them now. + batched_auth_events = { + event_id: event_map[event_id] + for event_id in itertools.chain( + calculated_auth_event_ids, event.prev_state_events + ) + } + # If we have calculated state before this event, remember it so we don't need + # to keep asking the database. The state we are remembering is StateMap[event_id] + # along with the state group integer, which is exactly the data needed by + # _calculate_state_dag_context + if context.state_group is not None: + event_id_to_state_group[event.event_id] = context.state_group + state_group_to_state_map[ + context.state_group + ] = await context.get_current_state_ids() + else: + calculated_auth_events = await self._store.get_events( + calculated_auth_event_ids, + # Allow rejected events so events which depend on them fail with coherent + # error messages. If we filtered them out here, we'd instead fail with + # a missing events error which is misleading. + allow_rejected=True, + redact_behaviour=EventRedactBehaviour.as_is, + ) + + try: + validate_event_for_room_version(event) + await check_state_independent_auth_rules( + self._store, + event, + batched_auth_events, + batched_rejections=rejected_event_id_to_reason_map, + ) + check_state_dependent_auth_rules( + event, calculated_auth_events.values() + ) + except AuthError as e: + logger.warning( + "Rejecting %r while persisting state DAG because %s", event, e + ) + context.rejected = RejectedReason.AUTH_ERROR + except EventSizeError as e: + if e.unpersistable: + # This event is completely unpersistable. + raise e + # Otherwise, we are somewhat lenient and just persist the event + # as rejected, for moderate compatibility with older Synapse + # versions. + logger.warning("While validating received event %r: %s", event, e) + context.rejected = RejectedReason.OVERSIZED_EVENT + + if context.rejected is not None: + rejected_event_id_to_reason_map[event.event_id] = context.rejected + processed_event_map[event.event_id] = event + return event, context + + if from_send_join: + for i, event in enumerate(sorted_events): + assert supports_msc4242_state_dag( + event + ) # to convert from EventBase to MSC4242Event + event_and_context = await process(event) + if event_and_context[1].rejected is not None: + has_rejected_events = True + events_and_contexts_to_persist.append(event_and_context) + + # The above function is typically not async, and so won't yield to + # the reactor. For large rooms let's yield to the reactor + # occasionally to ensure we don't block other work. + if (i + 1) % 1000 == 0: + await self._clock.sleep(Duration(seconds=0)) + + for batch in batch_iter(events_and_contexts_to_persist, 1000): + await self.persist_events_and_notify( + room_id, + batch, + # Mark these events as backfilled as they're historic events that + # will eventually be backfilled. For example, missing events we + # fetch during backfill should be marked as backfilled as well. + backfilled=True, + ) + else: + for event in sorted_events: + assert supports_msc4242_state_dag( + event + ) # to convert from EventBase to MSC4242Event + event_and_context = await process(event) + if event_and_context[1].rejected is not None: + has_rejected_events = True + await self.persist_events_and_notify( + room_id, + [event_and_context], + # Mark these events as backfilled as they're historic events that + # will eventually be backfilled. For example, missing events we + # fetch during backfill should be marked as backfilled as well. + backfilled=True, + ) + + return has_rejected_events + + async def _calculate_state_dag_context( + self, + event: MSC4242Event, + known_prev_state_maps: dict[int, StateMap[str]] | None = None, + event_map: dict[str, EventBase] | None = None, + ) -> tuple[EventContext, list[str]]: + """Calculate the context and auth events for an MSC4242 State DAG event. + + The state before the event is resolved from the event's `prev_state_events`, and + the auth events are then calculated from that state, in the same way + `EventBuilder.build` does when creating events locally. This is the expensive + bit, but it's a one-time cost as the calculated auth events are remembered on + the event. + + Args: + event: The event to calculate the context for. + known_prev_state_maps: The state maps at all of the event's + `prev_state_events`, keyed by state group, if known. This allows events + which have not been persisted yet to be processed. + event_map: Events which may be needed to resolve the state but which have not + been persisted yet. + Returns: + The persisted EventContext and the calculated auth event IDs. + + Raises: + SynapseError if the event has no `prev_state_events` and is not a create + event. + """ + is_create_event = len(event.prev_state_events) == 0 and ( + event.type, + event.get_state_key(), + ) == (EventTypes.Create, "") + if len(event.prev_state_events) == 0 and not is_create_event: + raise SynapseError(502, f"event {event.event_id} has no prev_state_events") + + if known_prev_state_maps: + if len(known_prev_state_maps) == 1: + # There's only 1 state map so we don't need to do state resolution at all. + # resolve_state_groups should not be called for a single state group as per its docstring + state_ids = list(known_prev_state_maps.values())[0] + else: + res = await self._state_resolution_handler.resolve_state_groups( + event.room_id, + event.room_version.identifier, + known_prev_state_maps, + event_map=event_map, + state_res_store=StateResolutionStore( + self._store, self._state_deletion_store + ), + ) + state_ids = await res.get_state(self._state_storage_controller) + else: + state_ids = await self._state_handler.compute_state_after_events( + event.room_id, + event.prev_state_events, + # There is no filter to apply here and we want all the events anyway since we need + # to persist the entire calculate state. + state_filter=None, + # By this point, we expect to have already finished processing the + # events *causally prior* to `event` in the state DAG because this + # function is called unilaterally WHEN we are processing events in the + # state DAG all the way down the chain and the caller ensures we process + # events in topological order. + # + # If this assumption is ever violated, `compute_state_after_events` will + # yell loudly about missing state groups. + # + # In the context of partially stated rooms (faster remote room joins) + # this would be part of background step verifying the entire state DAG, + # so we know that we don't have the full state processed yet. To wait + # here would be circular, we'd deadlock. + await_full_state=False, + ) + + # We should always have some resolved state after the prev_state_events, except + # for the create event which is the start of the state DAG. + + if not is_create_event: + # We must have calculated room state for every event by now. + # Only the create event can have no room state (since it's the first event in the room) + assert len(state_ids) > 0, "State at prev_state_events was not calculated" + + context = await self._state_handler.compute_event_context( + event, + state_ids, + # `compute_event_context` asserts this is None if there is no state, which + # is the case for the create event. + partial_state=None if is_create_event else False, + ) + return context, self._event_auth_handler.compute_auth_events(event, state_ids) + @trace async def _check_event_auth( self, origin: str | None, event: EventBase, context: EventContext @@ -1873,7 +2402,7 @@ class FederationEventHandler: # caller rather than swallow with `context.rejected` (since we cannot be # certain that there is a permanent problem with the event). claimed_auth_events = await self._load_or_fetch_auth_events_for_event( - origin, event + origin, event, context ) set_tag( SynapseTags.RESULT_PREFIX + "claimed_auth_events", @@ -1898,6 +2427,13 @@ class FederationEventHandler: context.rejected = RejectedReason.AUTH_ERROR return + if supports_msc4242_state_dag(event): + # For MSC4242 State DAG rooms, the auth events are calculated from the state before + # the event rather than taken from the event, so checking the event against its auth + # events (step 4) is the same as checking it against the state before it (step 5), + # and only step 4 is performed. + return + # now check the auth rules pass against the room state before the event # https://spec.matrix.org/v1.3/server-server-api/#checks-performed-on-receipt-of-a-pdu: # 5. Passes authorization rules based on the state before the event, @@ -2111,7 +2647,7 @@ class FederationEventHandler: event.internal_metadata.soft_failed = True async def _load_or_fetch_auth_events_for_event( - self, destination: str | None, event: EventBase + self, destination: str | None, event: EventBase, context: EventContext ) -> Collection[EventBase]: """Fetch this event's auth_events, from database or remote @@ -2124,6 +2660,10 @@ class FederationEventHandler: to `destination`, or we couldn't validate the signature on the event (which in turn has multiple potential causes). + MSC4242 State DAG events do not list their auth events, so instead we make sure we + have all of the event's `prev_state_events` and then calculate the auth events from + the state before the event. + Args: destination: where to send the /event_auth request. Typically the server that sent us `event` in the first place. @@ -2133,8 +2673,11 @@ class FederationEventHandler: event: the event whose auth_events we want + context: the state before the event. + Returns: - all of the events listed in `event.auth_events_ids`, after deduplication + all of the events listed in `event.auth_events_ids`, after deduplication. For + MSC4242 State DAG events, the calculated auth events. Raises: AssertionError if some auth events were missing and no `destination` was @@ -2142,7 +2685,11 @@ class FederationEventHandler: AuthError if we were unable to fetch the auth_events for any reason. """ - event_auth_event_ids = set(event.auth_event_ids()) + event_auth_event_ids = set( + event.prev_state_events + if supports_msc4242_state_dag(event) + else event.auth_event_ids() + ) event_auth_events = await self._store.get_events( event_auth_event_ids, allow_rejected=True ) @@ -2150,7 +2697,15 @@ class FederationEventHandler: event_auth_events.keys() ) if not missing_auth_event_ids: + if supports_msc4242_state_dag(event): + return await self._load_calculated_auth_events(event, context) return event_auth_events.values() + if supports_msc4242_state_dag(event): + raise AuthError( + code=HTTPStatus.FORBIDDEN, + msg=f"Unknown prev_state_events {missing_auth_event_ids}; fetching them " + "is not implemented for MSC4242 state DAG rooms", + ) if destination is None: # this shouldn't happen: destination must be set unless we know we have already # persisted the auth events. @@ -2180,6 +2735,8 @@ class FederationEventHandler: missing_auth_event_ids.difference_update(extra_auth_events.keys()) event_auth_events.update(extra_auth_events) if not missing_auth_event_ids: + if supports_msc4242_state_dag(event): + return await self._load_calculated_auth_events(event, context) return event_auth_events.values() # we still don't have all the auth events. @@ -2193,6 +2750,36 @@ class FederationEventHandler: # instead we raise an AuthError, which will make the caller ignore it. raise AuthError(code=HTTPStatus.FORBIDDEN, msg="Auth events could not be found") + async def _load_calculated_auth_events( + self, event: MSC4242Event, context: EventContext + ) -> Collection[EventBase]: + """Calculate the auth events for an MSC4242 State DAG event and load them. + + MSC4242 events do not list their auth events, so they are calculated from the + state before the event, in the same way `EventBuilder.build` does when creating + events locally. The calculated auth event IDs are remembered on the event so we + don't need to calculate them again when the event is persisted. + + Args: + event: the event whose auth events we want. + context: the state before the event. + + Returns: + The calculated auth events. + """ + state_ids = await context.get_prev_state_ids() + calculated_auth_event_ids = self._event_auth_handler.compute_auth_events( + event, state_ids + ) + event.internal_metadata.calculated_auth_event_ids = calculated_auth_event_ids + calculated_auth_events = await self._store.get_events( + calculated_auth_event_ids, + # Rejected events never form part of the room state, so the calculated auth events + # cannot be rejected and are loaded without allowing rejected events. + allow_rejected=False, + ) + return calculated_auth_events.values() + @trace @tag_args async def _get_remote_auth_chain_for_event( @@ -2421,10 +3008,38 @@ class FederationEventHandler: ) raise SynapseError(HTTPStatus.BAD_REQUEST, "Too many prev_events") - if len(ev.auth_event_ids()) > 10: + if ( + supports_msc4242_state_dag(ev) + and len(ev.prev_state_events) > StateDag.MAX_PREV_STATE_EVENTS + ): + logger.warning( + "Rejecting event %s which has %i prev_state_events", + ev.event_id, + len(ev.prev_state_events), + ) + raise SynapseError(HTTPStatus.BAD_REQUEST, "Too many prev_state_events") + + # MSC4242 State DAG events don't list their auth events (they're calculated from + # the state DAG), so there's nothing to bound here for them. + if not supports_msc4242_state_dag(ev) and len(ev.auth_event_ids()) > 10: logger.warning( "Rejecting event %s which has %i auth_events", ev.event_id, len(ev.auth_event_ids()), ) raise SynapseError(HTTPStatus.BAD_REQUEST, "Too many auth_events") + + +def is_state_dag_connected(state_dag: Collection[MSC4242Event]) -> bool: + """Returns True if the given events form a connected state DAG (MSC4242). + + The DAG is connected if every event referenced in a `prev_state_events` field is + itself present in `state_dag`, i.e. there are no gaps in the DAG. + """ + have_event_ids = {ev.event_id for ev in state_dag} + all_prev_state_events = { + prev_state_event_id + for ev in state_dag + for prev_state_event_id in ev.prev_state_events + } + return all_prev_state_events.issubset(have_event_ids) diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 53cf5300bf..22cc946a64 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -653,7 +653,54 @@ class EventsPersistenceStorageController: state_delta_for_room = None new_state_dag_extrems = None - if not backfilled: + if all_supports_msc4242_state_dag(chunk) and backfilled: + # Backfilled events do not update the room DAG extremities or the + # current state, but a state event which is new to us still has to move + # the state DAG forward extremities. + # + # In practice the only thing which gets here are state events from /send_join. + # Those events are persisted as backfilled so that they do not appear down + # /sync, but the state DAG forward extremities must be updated otherwise when + # we hit the join event we won't calculate the current state correctly due to having + # stale extremities. + # + # Catching up on a room we are already in does not come through here: the + # state DAG walk in /get_missing_events fetches `prev_state_events` at + # backfilled=False, and we refuse to persist any event whose + # `prev_state_events` we do not already hold, so an event cannot reach us + # ahead of the prev_state_events it references. + # + # There is one edge case here: when prev_events disagrees with prev_state_events. E.g: + # + # prev_events graph We know S1, S2, S3 + # S1 <- S2 <- S3 Sx = state event, M = message + # ^-- S4 <- S5 <- M Backfilling M returns S4, S5 alongside it + # + # Suppose M incorrectly has prev_state_events=[S3] but prev_events=[S5] + # then /get_missing_events will NOT fetch S4, S5 and only when you /backfill + # the prev_events graph will you suddenly become aware of [S4, S5], + # which will hit this code path. In this case, we should be recalculating the current + # state to take S4,S5 into account but this code path does not. + # We rely on the caller (_process_pulled_events) to handle this case by checking if + # _backfilled_ events changed the state DAG forward extremities, and in the extremely + # unlikely case it did, warn in the logs and recalculate the room state. + seen_event_ids = await self.main_store.have_seen_events( + room_id, [event.event_id for event, _ in chunk] + ) + unseen_chunk = [ + (event, context) + for event, context in chunk + if event.event_id not in seen_event_ids + ] + if unseen_chunk: + new_state_dag_extrems = ( + await self._calculate_new_state_dag_extremities( + room_id, + await self.main_store.get_state_dag_extremities(room_id), + unseen_chunk, + ) + ) + elif not backfilled: if all_supports_msc4242_state_dag(chunk): with Measure( self._clock, diff --git a/synapse/storage/controllers/state.py b/synapse/storage/controllers/state.py index b2e1d61a33..a6226c9dbd 100644 --- a/synapse/storage/controllers/state.py +++ b/synapse/storage/controllers/state.py @@ -533,7 +533,14 @@ class StateStorageController: Raises: RuntimeError if we don't have a state group for one or more of the events - (ie. they are outliers or unknown) + (ie. they are outliers or unknown) AND the lookup reached the database. + + XXX: `_get_state_group_for_events` raises `RuntimeError` for unknown events, but + the `@cachedList` wrapper swallows it and the caller just gets partial results. + For the time-being, Callers MUST NOT rely on this to determine missing events, + as cached responses return None for the state group instead of raising. + + TODO: audit every caller, as some of them may be relying on the current behaviour. """ if await_full_state: await self._partial_state_events_tracker.await_full_state(event_ids) diff --git a/synapse/storage/databases/main/rejections.py b/synapse/storage/databases/main/rejections.py index c73c3d761d..1e0e87ae28 100644 --- a/synapse/storage/databases/main/rejections.py +++ b/synapse/storage/databases/main/rejections.py @@ -35,3 +35,17 @@ class RejectionsStore(SQLBaseStore): allow_none=True, desc="get_rejection_reason", ) + + async def has_rejected_event_ids(self, event_ids: set[str]) -> set[str]: + """Filter the provided event IDs to only return rejected events. + This does not check if the event IDs are persisted (missing events would appear as non-rejected). + """ + rows = await self.db_pool.simple_select_many_batch( + table="rejections", + column="event_id", + iterable=event_ids, + retcols=("event_id",), + keyvalues={}, + desc="has_rejected_event_ids", + ) + return {r[0] for r in rows} diff --git a/synapse/storage/databases/main/state.py b/synapse/storage/databases/main/state.py index 3e42d46084..bc8a902846 100644 --- a/synapse/storage/databases/main/state.py +++ b/synapse/storage/databases/main/state.py @@ -702,6 +702,12 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore): ) -> Mapping[str, int]: """Returns mapping event_id -> state_group. + XXX: the `@cachedList` wrapper swallows RuntimeError and the caller just gets partial results. + For the time-being, Callers MUST NOT rely on this to determine missing events, + as cached responses return None for the state group instead of raising. + + TODO: audit every caller, as some of them may be relying on the current behaviour. + Raises: RuntimeError if the state is unknown at any of the given events """ diff --git a/tests/federation/test_federation_server.py b/tests/federation/test_federation_server.py index 61dcb7446b..bba12f5860 100644 --- a/tests/federation/test_federation_server.py +++ b/tests/federation/test_federation_server.py @@ -20,7 +20,6 @@ # import logging from http import HTTPStatus -from unittest import skip as skip_test from unittest.mock import Mock from parameterized import parameterized @@ -44,6 +43,7 @@ from synapse.types import JsonDict, UserID from synapse.util.clock import Clock from tests import unittest +from tests.server import FakeChannel from tests.test_utils.event_builders import make_test_event, make_test_pdu_event from tests.unittest import override_config @@ -440,13 +440,13 @@ class GetMissingEventsStateDagTests(unittest.FederatingHomeserverTestCase): self.room_id, "m.room.topic", {"topic": "two"}, tok=self.local_user_token )["event_id"] - def _get_missing_events( + def _get_missing_events_channel( self, earliest_events: list[str], latest_events: list[str], walk_state_dag: bool, limit: int = 10, - ) -> list[JsonDict]: + ) -> FakeChannel: content: JsonDict = { "earliest_events": earliest_events, "latest_events": latest_events, @@ -455,11 +455,25 @@ class GetMissingEventsStateDagTests(unittest.FederatingHomeserverTestCase): if walk_state_dag: content[StateDag.GET_MISSING_EVENTS_FIELD] = True - channel = self.make_signed_federation_request( + return self.make_signed_federation_request( "POST", f"/_matrix/federation/v1/get_missing_events/{self.room_id}", content=content, ) + + def _get_missing_events( + self, + earliest_events: list[str], + latest_events: list[str], + walk_state_dag: bool, + limit: int = 10, + ) -> list[JsonDict]: + channel = self._get_missing_events_channel( + earliest_events=earliest_events, + latest_events=latest_events, + walk_state_dag=walk_state_dag, + limit=limit, + ) self.assertEqual(HTTPStatus.OK, channel.code, channel.json_body) return channel.json_body["events"] @@ -511,25 +525,27 @@ class GetMissingEventsStateDagTests(unittest.FederatingHomeserverTestCase): self.assertEqual(len(events), 2) def test_is_opt_in(self) -> None: + # the state DAG is only walked when the request asks for it... state_dag_events = self._get_missing_events( earliest_events=[], latest_events=[self.second_topic_id], limit=20, walk_state_dag=True, ) - timeline_events = self._get_missing_events( + self.assertEqual( + [ev for ev in state_dag_events if ev["type"] == "m.room.message"], [] + ) + + # ...and when it doesn't, we refuse to serve the request rather than walking + # the timeline DAG, which isn't implemented for state DAG rooms. + channel = self._get_missing_events_channel( earliest_events=[], latest_events=[self.second_topic_id], limit=20, walk_state_dag=False, ) - - self.assertEqual( - [ev for ev in state_dag_events if ev["type"] == "m.room.message"], [] - ) - self.assertNotEqual( - [ev for ev in timeline_events if ev["type"] == "m.room.message"], [] - ) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.json_body) + self.assertEqual(Codes.UNRECOGNIZED, channel.json_body["errcode"]) def test_walks_from_a_non_state_event(self) -> None: store = self.hs.get_datastores().main @@ -1234,7 +1250,6 @@ class SendJoinFederationTests(unittest.FederatingHomeserverTestCase): """Test send_join with USE_FROZEN_DICTS=False""" self._test_send_join_common(room_version) - @skip_test("requires MSC4242 inbound event auth") @override_config({"experimental_features": {"msc4242_enabled": True}}) def test_send_join_state_dag(self) -> None: """ @@ -1246,7 +1261,6 @@ class SendJoinFederationTests(unittest.FederatingHomeserverTestCase): """ self._test_send_join_common(RoomVersions.MSC4242v12.identifier) - @skip_test("requires MSC4242 inbound event auth") @override_config({"experimental_features": {"msc4242_enabled": True}}) def test_send_join_state_dag_ignores_partial_state(self) -> None: # FIXME: when we support partial joins this test can be deleted diff --git a/tests/handlers/test_federation_event.py b/tests/handlers/test_federation_event.py index b423948096..8c952352c1 100644 --- a/tests/handlers/test_federation_event.py +++ b/tests/handlers/test_federation_event.py @@ -23,13 +23,15 @@ from unittest import mock from twisted.internet.testing import MemoryReactor from synapse.api.errors import AuthError, StoreError -from synapse.api.room_versions import RoomVersion +from synapse.api.room_versions import RoomVersion, RoomVersions from synapse.event_auth import ( check_state_dependent_auth_rules, check_state_independent_auth_rules, ) +from synapse.events.py_protocol import MSC4242Event, supports_msc4242_state_dag from synapse.events.snapshot import EventContext from synapse.federation.transport.client import StateRequestResponse +from synapse.handlers.federation_event import is_state_dag_connected from synapse.logging.context import LoggingContext from synapse.rest import admin from synapse.rest.client import login, room @@ -1182,3 +1184,193 @@ class FederationEventHandlerTests(unittest.FederatingHomeserverTestCase): bert_member_event.event_id, "Rejected kick event unexpectedly became part of room state.", ) + + +class IsStateDagConnectedTests(unittest.TestCase): + def setUp(self) -> None: + super().setUp() + self.origin_server_ts = 0 + + def _make_event(self, prev_state_events: list[str]) -> MSC4242Event: + """Build an MSC4242 event with the given `prev_state_events`.""" + # Event IDs are hashes of the event, so vary a field to keep them distinct. + self.origin_server_ts += 1 + event = make_test_event( + { + "type": "m.room.topic", + "state_key": "", + "content": {}, + "sender": "@alice:test", + "origin_server_ts": self.origin_server_ts, + "prev_state_events": prev_state_events, + }, + room_version=RoomVersions.MSC4242v12, + ) + assert supports_msc4242_state_dag(event) + return event + + def test_linear(self) -> None: + create = self._make_event([]) + second = self._make_event([create.event_id]) + third = self._make_event([second.event_id]) + + self.assertTrue(is_state_dag_connected([third, create, second])) + + def test_fork_and_merge(self) -> None: + create = self._make_event([]) + fork_one = self._make_event([create.event_id]) + fork_two = self._make_event([create.event_id]) + fork_three = self._make_event([create.event_id]) + merge = self._make_event([fork_one.event_id, fork_three.event_id]) + + self.assertTrue( + is_state_dag_connected([create, fork_one, fork_two, fork_three, merge]) + ) + + def test_missing_prev_state_event(self) -> None: + create = self._make_event([]) + second = self._make_event([create.event_id]) + unconnected = self._make_event(["$unknown"]) + + self.assertFalse(is_state_dag_connected([create, second, unconnected])) + + def test_missing_intermediate_event(self) -> None: + create = self._make_event([]) + second = self._make_event([create.event_id]) + third = self._make_event([second.event_id]) + + self.assertFalse(is_state_dag_connected([create, third])) + + +class BackfilledStateDagEventTests(unittest.FederatingHomeserverTestCase): + servlets = [ + admin.register_servlets, + login.register_servlets, + room.register_servlets, + ] + + @unittest.override_config({"experimental_features": {"msc4242_enabled": True}}) + def test_backfilled_state_events_update_current_state(self) -> None: + """State events we only learn about through backfill must reach the current state. + + An event's `prev_events` can reach state events which its `prev_state_events` do + not. This should not normally happen but the protocol allows it, so we have to handle it. + The state DAG walk in /get_missing_events only follows `prev_state_events`, + so those state events are never fetched that way, and we first see them when the + `prev_events` graph is backfilled. They are still state events which nothing in + our state DAG references, so they belong on the state DAG frontier and therefore + change the current state. + + prev_state_events graph We know MEMBER and TOPIC + Sx = state event + MEMBER <- TOPIC Backfill hands us S4 and S5, forking at MEMBER + `-- S4 <- S5 + + The frontier becomes {TOPIC, S5}, so S5 has to be merged into the current state. + """ + OTHER_USER = f"@user:{self.OTHER_SERVER_NAME}" + main_store = self.hs.get_datastores().main + federation_event_handler = self.hs.get_federation_event_handler() + + user_id = self.register_user("kermit", "test") + tok = self.login("kermit", "test") + room_id = self.helper.create_room_as( + room_creator=user_id, + tok=tok, + room_version=RoomVersions.MSC4242v12.identifier, + ) + room_version = self.get_success(main_store.get_room_version(room_id)) + self.assertTrue(room_version.msc4242_state_dags) + + # allow the remote user to send state events + self.helper.send_state( + room_id, + "m.room.power_levels", + {"events_default": 0, "state_default": 0}, + tok=tok, + ) + + # add the remote user to the room. The backfilled branch forks here. + member_event = self.get_success( + event_injection.inject_member_event(self.hs, room_id, OTHER_USER, "join") + ) + + # Move the state DAG locally, so that the backfill below is a genuine fork + # rather than an extension of the only branch we have. + topic_event_id = self.helper.send_state( + room_id, "m.room.topic", {"topic": "local"}, tok=tok + )["event_id"] + + extremities_before = self.get_success( + main_store.get_state_dag_extremities(room_id) + ) + self.assertIncludes({topic_event_id}, extremities_before, exact=True) + + # S4 and S5 are a branch of the state DAG which we have never been told about: + # nothing we hold references them via `prev_state_events`, so the state DAG walk + # would never have fetched them. + s4 = make_test_event( + self.add_hashes_and_signatures_from_other_server( + { + "type": "test_state_type", + "state_key": "", + "room_id": room_id, + "sender": OTHER_USER, + "prev_events": [member_event.event_id], + "prev_state_events": [member_event.event_id], + "origin_server_ts": 1, + "depth": 10, + "content": {"body": "s4"}, + }, + room_version, + ), + room_version, + ) + s5 = make_test_event( + self.add_hashes_and_signatures_from_other_server( + { + "type": "test_state_type", + "state_key": "", + "room_id": room_id, + "sender": OTHER_USER, + "prev_events": [s4.event_id], + "prev_state_events": [s4.event_id], + "origin_server_ts": 2, + "depth": 11, + "content": {"body": "s5"}, + }, + room_version, + ), + room_version, + ) + + self.get_success( + federation_event_handler._process_pulled_events( + self.OTHER_SERVER_NAME, [s4, s5], backfilled=True + ) + ) + + # The tip of the backfilled branch is now a state DAG forward extremity, and the + # branch we already had still is too: the fork is unmerged in the state DAG. + extremities_after = self.get_success( + main_store.get_state_dag_extremities(room_id) + ) + self.assertIncludes( + {topic_event_id, s5.event_id}, + extremities_after, + exact=True, + message="the backfilled state event did not become a state DAG forward extremity", + ) + + # ...and resolving that frontier is what the current state must reflect. + state_map = self.get_success(main_store.get_partial_current_state_ids(room_id)) + self.assertEqual( + state_map.get(("test_state_type", "")), + s5.event_id, + "the state event learned through backfill is not part of the current state", + ) + self.assertEqual( + state_map.get(("m.room.topic", "")), + topic_event_id, + "merging in the backfilled branch lost the state on the branch we had", + )