diff --git a/changelog.d/20127.feature b/changelog.d/20127.feature new file mode 100644 index 00000000000..5e6738e0359 --- /dev/null +++ b/changelog.d/20127.feature @@ -0,0 +1 @@ +Add experimental federation client support for [MSC4242](https://github.com/matrix-org/matrix-spec-proposals/pull/4242): State DAGs. diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 5b334543256..e46392944be 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -125,6 +125,9 @@ class SendJoinResult: # Always contains the server we joined off. servers_in_room: AbstractSet[str] + # Only valid for state DAG rooms (MSC4242) + state_dag: list[EventBase] | None + class FederationClient(FederationBase): def __init__(self, hs: "HomeServer"): @@ -1107,11 +1110,12 @@ async def send_join( SynapseError: if the chosen remote server returns a 300/400 code, or no servers successfully handle the request. """ - # See related restriction in /createRoom requests in handlers/room.py - if room_version.msc4242_state_dags: - raise UnsupportedRoomVersionError( - "Homeserver does not support this room version over federation" - ) + + def find_create_event(events: list[EventBase]) -> EventBase | None: + for e in events: + if (e.type, e.state_key) == (EventTypes.Create, ""): + return e + return None async def send_request(destination: str) -> SendJoinResult: response = await self._do_send_join( @@ -1141,13 +1145,16 @@ async def send_request(destination: str) -> SendJoinResult: state = response.state auth_chain = response.auth_events - - create_event = None - for e in state: - if (e.type, e.state_key) == (EventTypes.Create, ""): - create_event = e - break - + state_dag: list[EventBase] = [] + if room_version.msc4242_state_dags: + if not response.state_dag: + raise InvalidResponseError("No state_dag returned") + state_dag = response.state_dag + + # Validate the create event and room version are what we expect to see. + create_event = find_create_event( + state_dag if room_version.msc4242_state_dags else state + ) 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. @@ -1165,8 +1172,31 @@ async def send_request(destination: str) -> SendJoinResult: % (create_room_version,) ) + # Validate and set faster room joins fields + servers_in_room = None + if response.servers_in_room is not None: + servers_in_room = set(response.servers_in_room) + + if response.members_omitted: + if not servers_in_room: + raise InvalidResponseError( + "members_omitted was set, but no servers were listed in the room" + ) + + if not partial_state: + raise InvalidResponseError( + "members_omitted was set, but we asked for full state" + ) + + # `servers_in_room` is supposed to be a complete list. + # Fix things up in case the remote homeserver is badly behaved. + servers_in_room.add(destination) + logger.info( - "Processing from send_join %d events", len(state) + len(auth_chain) + "Processing from send_join %d events", + len(state_dag) + if room_version.msc4242_state_dags + else (len(state) + len(auth_chain)), ) # We now go and check the signatures and hashes for the event. Note @@ -1184,65 +1214,84 @@ async def _execute(pdu: EventBase) -> None: if valid_pdu: valid_pdus_map[valid_pdu.event_id] = valid_pdu - await concurrently_execute( - _execute, itertools.chain(state, auth_chain), 10000 - ) - - # NB: We *need* to copy to ensure that we don't have multiple - # references being passed on, as that causes... issues. - signed_state = [ - valid_pdus_map[p.event_id].deep_copy() - for p in state - if p.event_id in valid_pdus_map - ] - - signed_auth = [ - valid_pdus_map[p.event_id] - for p in auth_chain - if p.event_id in valid_pdus_map - ] - - # double-check that the auth chain doesn't include a different create event - auth_chain_create_events = [ - e.event_id - for e in signed_auth - if (e.type, e.state_key) == (EventTypes.Create, "") - ] - if auth_chain_create_events and auth_chain_create_events != [ - create_event.event_id - ]: - raise InvalidResponseError( - "Unexpected create event(s) in auth chain: %s" - % (auth_chain_create_events,) - ) - - servers_in_room = None - if response.servers_in_room is not None: - servers_in_room = set(response.servers_in_room) - - if response.members_omitted: - if not servers_in_room: + # Verify signatures/hashes on events, and make sure they all refer to the same room. + if room_version.msc4242_state_dags: + if state or auth_chain or servers_in_room: raise InvalidResponseError( - "members_omitted was set, but no servers were listed in the room" + "State DAG rooms must not set servers_in_room, state or auth_chain fields" ) + await concurrently_execute(_execute, itertools.chain(state_dag), 10000) + # NB: We *need* to copy to ensure that we don't have multiple + # references being passed on, as that causes... issues. + signed_state_dag = [ + valid_pdus_map[p.event_id].deep_copy() + for p in state_dag + if p.event_id in valid_pdus_map + ] - if not partial_state: + # Verify each event is for this room (and thus has the same create event as it is v12+) + for state_event in signed_state_dag: + if state_event.room_id != pdu.room_id: + raise InvalidResponseError( + "%s in state_dag belongs to room %s, not %s which we are joining" + % (state_event.event_id, state_event.room_id, pdu.room_id) + ) + return SendJoinResult( + event=event, + state=[], + auth_chain=[], + state_dag=signed_state_dag, + origin=destination, + # The current Synapse implementation of MSC4242 does not support + # faster remote room joins, so always set partial_state=False. + partial_state=False, + servers_in_room=frozenset(), + ) + else: + if state_dag: raise InvalidResponseError( - "members_omitted was set, but we asked for full state" + "Room does not support state DAGs but set state_dag field" ) + await concurrently_execute( + _execute, itertools.chain(state, auth_chain), 10000 + ) - # `servers_in_room` is supposed to be a complete list. - # Fix things up in case the remote homeserver is badly behaved. - servers_in_room.add(destination) + # NB: We *need* to copy to ensure that we don't have multiple + # references being passed on, as that causes... issues. + signed_state = [ + valid_pdus_map[p.event_id].deep_copy() + for p in state + if p.event_id in valid_pdus_map + ] - return SendJoinResult( - event=event, - state=signed_state, - auth_chain=signed_auth, - origin=destination, - partial_state=response.members_omitted, - servers_in_room=servers_in_room or frozenset(), - ) + signed_auth = [ + valid_pdus_map[p.event_id] + for p in auth_chain + if p.event_id in valid_pdus_map + ] + + # double-check that the auth chain doesn't include a different create event + auth_chain_create_events = [ + e.event_id + for e in signed_auth + if (e.type, e.state_key) == (EventTypes.Create, "") + ] + if auth_chain_create_events and auth_chain_create_events != [ + create_event.event_id + ]: + raise InvalidResponseError( + "Unexpected create event(s) in auth chain: %s" + % (auth_chain_create_events,) + ) + return SendJoinResult( + event=event, + state=signed_state, + auth_chain=signed_auth, + origin=destination, + partial_state=response.members_omitted, + servers_in_room=servers_in_room or frozenset(), + state_dag=None, + ) # MSC3083 defines additional error codes for room joins. failover_errcodes = None @@ -1542,6 +1591,7 @@ async def get_missing_events( limit: int, min_depth: int, timeout: int, + state_dag: bool = False, ) -> list[EventBase]: """Tries to fetch events we are missing. This is called when we receive an event without having received all of its ancestors. @@ -1557,6 +1607,7 @@ async def get_missing_events( limit: Maximum number of events to return. min_depth: Minimum depth of events to return. timeout: Max time to wait in ms + state_dag: True to walk the state DAG (MSC4242 rooms) """ try: content = await self.transport_layer.get_missing_events( @@ -1567,6 +1618,7 @@ async def get_missing_events( limit=limit, min_depth=min_depth, timeout=timeout, + state_dag=state_dag, ) received_time = self._clock.time_msec() diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 5d5212ef96c..e4747016c40 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -776,18 +776,21 @@ async def get_missing_events( limit: int, min_depth: int, timeout: int, + state_dag: bool, ) -> JsonDict: path = _create_v1_path("/get_missing_events/%s", room_id) - + request_body = { + "limit": int(limit), + "min_depth": int(min_depth), + "earliest_events": earliest_events, + "latest_events": latest_events, + } + if state_dag: + request_body["org.matrix.msc4242.state_dag"] = True return await self.client.post_json( destination=destination, path=path, - data={ - "limit": int(limit), - "min_depth": int(min_depth), - "earliest_events": earliest_events, - "latest_events": latest_events, - }, + data=request_body, timeout=timeout, ) @@ -986,6 +989,10 @@ class SendJoinResponse: # "event" is not included in the response. event: EventBase | None = None + # MSC4242: State DAGs. Always included for state dag rooms, else None. + # Replaces auth_events. + state_dag: list[EventBase] | None = None + # The room state is incomplete members_omitted: bool = False @@ -1068,7 +1075,7 @@ class SendJoinParser(ByteParser[SendJoinResponse]): MAX_RESPONSE_SIZE = 500 * 1024 * 1024 def __init__(self, room_version: RoomVersion, v1_api: bool): - self._response = SendJoinResponse([], [], event_dict={}) + self._response = SendJoinResponse([], [], event_dict={}, state_dag=[]) self._room_version = room_version self._coros: list[Generator[None, bytes, None]] = [] @@ -1112,6 +1119,15 @@ def __init__(self, room_version: RoomVersion, v1_api: bool): ) ) + if room_version.msc4242_state_dags: + self._coros.append( + ijson.items_coro( + _event_list_parser(room_version, self._response.state_dag), + prefix + "state_dag.item", + use_float=True, + ) + ) + def write(self, data: bytes) -> int: for c in self._coros: c.send(data) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 4816c79c7b7..7c5534f403a 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -53,6 +53,7 @@ PartialStateConflictError, RequestSendFailed, SynapseError, + UnsupportedRoomVersionError, ) from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion from synapse.crypto.event_signing import compute_event_signature @@ -669,6 +670,12 @@ async def do_invite_join( 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, diff --git a/tests/handlers/test_device.py b/tests/handlers/test_device.py index cb047d118a7..ce1bd06b6d8 100644 --- a/tests/handlers/test_device.py +++ b/tests/handlers/test_device.py @@ -844,6 +844,7 @@ def test_local_device_changes_sent_to_new_servers_on_un_partial_state( partial_state=True, # Only REMOTE1_SERVER_NAME is known at join time. servers_in_room={self.REMOTE1_SERVER_NAME}, + state_dag=None, ) ) diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index 1f42490d96a..71f29be2ea1 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -574,6 +574,7 @@ def test_failed_partial_join_is_clean(self) -> None: ], partial_state=True, servers_in_room={"example.com"}, + state_dag=None, ) ) diff --git a/tests/handlers/test_room_member.py b/tests/handlers/test_room_member.py index 0a7475856a8..a4a57b64d97 100644 --- a/tests/handlers/test_room_member.py +++ b/tests/handlers/test_room_member.py @@ -169,6 +169,7 @@ def test_remote_joins_contribute_to_rate_limit(self) -> None: auth_chain=[create_event], partial_state=False, servers_in_room=frozenset(), + state_dag=None, ) ) diff --git a/tests/storage/test_stream.py b/tests/storage/test_stream.py index de127e3971d..4ece16e9a0c 100644 --- a/tests/storage/test_stream.py +++ b/tests/storage/test_stream.py @@ -1455,6 +1455,7 @@ def test_remote_join(self) -> None: auth_chain=[create_event, creator_join_event], partial_state=False, servers_in_room=frozenset(), + state_dag=None, ) )