MSC4242: State DAGs (fed client)

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.

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,64 +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
-            )
+            # 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(
+                        "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
+                ]
 
-            # 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,)
+                # 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(),
                 )
-
-            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:
+            else:
+                if state_dag:
                     raise InvalidResponseError(
-                        "members_omitted was set, but no servers were listed in the room"
+                        "Room does not support state DAGs but set state_dag field"
                     )
+                await concurrently_execute(
+                    _execute, itertools.chain(state, auth_chain), 10000
+                )
 
-                if not partial_state:
+                # 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(
-                        "members_omitted was set, but we asked for full state"
+                        "Unexpected create event(s) in auth chain: %s"
+                        % (auth_chain_create_events,)
                     )
 
-                # `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)
-
             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.
@@ -1542,6 +1592,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 +1608,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 +1619,7 @@ async def get_missing_events(
                 limit=limit,
                 min_depth=min_depth,
                 timeout=timeout,
+                state_dag=state_dag,
             )
             received_time = self._clock.time_msec()
 

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)

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,

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,
             )
         )
 

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,
             )
         )
 

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,
             )
         )
 

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,
             )
         )