MSC4242: State DAGs (inbound events)

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.

changelog.d/20221.misc

@@ -0,0 +1 @@
+Add support for receiving events in [MSC4242](https://github.com/matrix-org/matrix-spec-proposals/pull/4242) rooms over federation.

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

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

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.
+    """

synapse/app/generic_worker.py

@@ -93,6 +93,7 @@
 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,

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

synapse/events/builder.py

@@ -24,7 +24,13 @@
 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,
@@ -157,6 +163,10 @@ async def build(
             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,

synapse/federation/federation_server.py

@@ -233,12 +233,23 @@ async def _handle_old_staged_events(self) -> None:
             # 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 @@ async def on_timestamp_to_event_request(
             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 @@ async def on_room_state_request(
         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 @@ async def on_state_ids_request(
         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 @@ async def _on_send_membership_event(
         # 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 @@ async def on_event_auth(
             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 @@ async def on_get_missing_events(
                     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
                 )

synapse/handlers/federation.py

@@ -64,7 +64,6 @@
     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 @@ async def maybe_backfill(
         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 @@ 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,
@@ -741,6 +738,11 @@ async def do_invite_join(
                 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 @@ async def on_invite_request(
         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")