MSC4242: State DAGs (serving)

synapse/api/constants.py

@@ -500,3 +500,9 @@ class StickyEvent:
 
     This is the default specified in the MSC. Chosen arbitrarily.
     """
+
+
+class StateDag:
+    GET_MISSING_EVENTS_FIELD: Final = "org.matrix.msc4242.state_dag"
+
+    MAX_MISSING_EVENTS: Final = 1000

synapse/federation/federation_server.py

@@ -30,6 +30,7 @@
     Callable,
     Collection,
     Mapping,
+    Sequence,
 )
 
 from prometheus_client import Counter, Gauge, Histogram
@@ -56,6 +57,7 @@
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion
 from synapse.crypto.event_signing import compute_event_signature
 from synapse.events import EventBase
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.events.snapshot import EventPersistencePair
 from synapse.federation.federation_base import (
     FederationBase,
@@ -864,6 +866,10 @@ async def on_send_join_request(
         event, context = await self._on_send_membership_event(
             origin, content, Membership.JOIN, room_id
         )
+
+        if supports_msc4242_state_dag(event):
+            caller_supports_partial_state = False
+
         # Use the join event's own stream ordering as the upper bound when fetching
         # forward extremities (below), so we only consider extremities that existed at
         # or before the join rather than those introduced by concurrent writes that
@@ -890,28 +896,46 @@ async def on_send_join_request(
             state_event_ids = prev_state_ids.values()
             servers_in_room = None
 
-        auth_chain_event_ids = await self.store.get_auth_chain_ids(
-            room_id, state_event_ids
-        )
+        state_dag: Sequence[EventBase] = ()
+        state_events: Sequence[EventBase] = ()
+        auth_chain_events: Sequence[EventBase] = ()
 
-        # if the caller has opted in, we can omit any auth_chain events which are
-        # already in state_event_ids
-        if caller_supports_partial_state:
-            auth_chain_event_ids.difference_update(state_event_ids)
+        if supports_msc4242_state_dag(event):
+            state_dag_map = await self.store.get_state_dag(
+                room_id, set(event.prev_state_events)
+            )
+            # Sort by depth, though this is just a nicety, MSC4242 does not require it
+            state_dag = sorted(
+                state_dag_map.values(), key=lambda ev: (ev.depth, ev.event_id)
+            )
+        else:
+            auth_chain_event_ids = await self.store.get_auth_chain_ids(
+                room_id, state_event_ids
+            )
 
-        auth_chain_events = await self.store.get_events_as_list(auth_chain_event_ids)
-        state_events = await self.store.get_events_as_list(state_event_ids)
+            # if the caller has opted in, we can omit any auth_chain events which are
+            # already in state_event_ids
+            if caller_supports_partial_state:
+                auth_chain_event_ids.difference_update(state_event_ids)
+
+            auth_chain_events = await self.store.get_events_as_list(
+                auth_chain_event_ids
+            )
+            state_events = await self.store.get_events_as_list(state_event_ids)
 
         # we try to do all the async stuff before this point, so that time_now is as
         # accurate as possible.
         time_now = self._clock.time_msec()
         event_json = event.get_pdu_json(time_now)
-        resp = {
+        resp: JsonDict = {
             "event": event_json,
-            "state": serialize_and_filter_pdus(state_events, time_now),
-            "auth_chain": serialize_and_filter_pdus(auth_chain_events, time_now),
             "members_omitted": caller_supports_partial_state,
         }
+        if supports_msc4242_state_dag(event):
+            resp["state_dag"] = serialize_and_filter_pdus(state_dag, time_now)
+        else:
+            resp["state"] = serialize_and_filter_pdus(state_events, time_now)
+            resp["auth_chain"] = serialize_and_filter_pdus(auth_chain_events, time_now)
 
         # Check the forward extremities for the room here. If there is more than one, it
         # is likely that another event was created in the room during the
@@ -942,7 +966,7 @@ async def on_send_join_request(
 
         await self._federation_callbacks.notify_on_event_delivered_over_federation(
             origin,
-            [event, *state_events, *auth_chain_events],
+            [event, *state_dag, *state_events, *auth_chain_events],
             FederatedEventDeliveryMethod.SEND_JOIN,
         )
 
@@ -1246,6 +1270,7 @@ async def on_get_missing_events(
         earliest_events: list[str],
         latest_events: list[str],
         limit: int,
+        walk_state_dag: bool = False,
     ) -> dict[str, list]:
         async with self._server_linearizer.queue((origin, room_id)):
             origin_host, _ = parse_server_name(origin)
@@ -1260,7 +1285,7 @@ async def on_get_missing_events(
             )
 
             missing_events = await self.handler.on_get_missing_events(
-                origin, room_id, earliest_events, latest_events, limit
+                origin, room_id, earliest_events, latest_events, limit, walk_state_dag
             )
 
             if len(missing_events) < 5:

synapse/federation/sender/__init__.py

@@ -149,6 +149,7 @@
 from synapse.api.constants import EventTypes, Membership
 from synapse.api.presence import UserPresenceState
 from synapse.events import EventBase
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.federation.sender.per_destination_queue import (
     CATCHUP_RETRY_INTERVAL,
     PerDestinationQueue,
@@ -660,7 +661,10 @@ async def handle_event(event: EventBase) -> None:
                             # banned then it won't receive the event because it won't
                             # be in the room after the ban.
                             destinations = await self.state.get_hosts_in_room_at_events(
-                                event.room_id, event_ids=event.prev_event_ids()
+                                event.room_id,
+                                event_ids=event.prev_state_events
+                                if supports_msc4242_state_dag(event)
+                                else event.prev_event_ids(),
                             )
                         except Exception:
                             logger.exception(

synapse/federation/transport/server/federation.py

@@ -28,7 +28,7 @@
     Sequence,
 )
 
-from synapse.api.constants import Direction, EduTypes
+from synapse.api.constants import Direction, EduTypes, StateDag
 from synapse.api.errors import Codes, SynapseError
 from synapse.api.room_versions import RoomVersions
 from synapse.api.urls import FEDERATION_UNSTABLE_PREFIX, FEDERATION_V2_PREFIX
@@ -639,13 +639,15 @@ async def on_POST(
         limit = int(content.get("limit", 10))
         earliest_events = content.get("earliest_events", [])
         latest_events = content.get("latest_events", [])
+        walk_state_dag = bool(content.get(StateDag.GET_MISSING_EVENTS_FIELD, False))
 
         result = await self.handler.on_get_missing_events(
             origin,
             room_id=room_id,
             earliest_events=earliest_events,
             latest_events=latest_events,
             limit=limit,
+            walk_state_dag=walk_state_dag,
         )
 
         return 200, result

synapse/handlers/federation.py

@@ -40,7 +40,13 @@
 from unpaddedbase64 import decode_base64
 
 from synapse import event_auth
-from synapse.api.constants import MAX_DEPTH, EventContentFields, EventTypes, Membership
+from synapse.api.constants import (
+    MAX_DEPTH,
+    EventContentFields,
+    EventTypes,
+    Membership,
+    StateDag,
+)
 from synapse.api.errors import (
     AuthError,
     CodeMessageException,
@@ -57,7 +63,8 @@
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion
 from synapse.crypto.event_signing import compute_event_signature
 from synapse.event_auth import validate_event_for_room_version
-from synapse.events import EventBase
+from synapse.events import EventBase, event_exists_in_state_dag
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.events.snapshot import EventContext, UnpersistedEventContextBase
 from synapse.events.validator import EventValidator
 from synapse.federation.federation_client import InvalidResponseError
@@ -1035,6 +1042,11 @@ async def on_make_join_request(
         # Note that this requires the /send_join request to come back to the
         # same server.
         prev_event_ids = None
+        prev_state_events = None
+        if room_version.msc4242_state_dags:
+            prev_state_events = list(
+                await self.store.get_state_dag_extremities(room_id)
+            )
         if room_version.restricted_join_rule:
             # Note that the room's state can change out from under us and render our
             # nice join rules-conformant event non-conformant by the time we build the
@@ -1091,6 +1103,7 @@ async def on_make_join_request(
             ) = await self.event_creation_handler.create_new_client_event(
                 builder=builder,
                 prev_event_ids=prev_event_ids,
+                prev_state_events=prev_state_events,
             )
         except SynapseError as e:
             logger.warning("Failed to create join to %s because %s", room_id, e)
@@ -1482,11 +1495,17 @@ async def on_get_missing_events(
         earliest_events: list[str],
         latest_events: list[str],
         limit: int,
+        walk_state_dag: bool = False,
     ) -> list[EventBase]:
         # We allow partially joined rooms since in this case we are filtering out
         # non-local events in `filter_events_for_server`.
         await self._event_auth_handler.assert_host_in_room(room_id, origin, True)
 
+        if walk_state_dag:
+            return await self.on_get_missing_events_state_dag(
+                room_id, earliest_events, latest_events, limit
+            )
+
         # Only allow up to 20 events to be retrieved per request.
         limit = min(limit, 20)
 
@@ -1509,6 +1528,69 @@ async def on_get_missing_events(
 
         return missing_events
 
+    async def on_get_missing_events_state_dag(
+        self,
+        room_id: str,
+        earliest_events: list[str],
+        latest_events: list[str],
+        limit: int,
+    ) -> list[EventBase]:
+        """Processes a /get_missing_events request for the state DAG.
+
+        This is similar to processing the normal DAG with a few notable exceptions:
+          * The max 20 limit does not apply. As the entire state DAG needs to be filled
+            in, we cannot arbitrarily set a low limit. If the state DAG delta is 1000s of
+            events, we rely on the sender to set sensible limits depending on the
+            bandwidth/round trip tradeoff, capped at `StateDag.MAX_MISSING_EVENTS` so a
+            single request cannot ask for an unbounded response.
+          * We do not filter any events in the state DAG. History visibility does not
+            filter out delivery of auth chain events, so neither should this. All of the
+            returned events will be treated as outliers and as such will not be delivered
+            to clients.
+          * `latest_events` may name events which are not themselves in the state DAG,
+            because the caller seeds this request with whatever it received over /send.
+            Only state events have `msc4242_state_dag_edges` rows, so we walk from such an
+            event's `prev_state_events` instead, and return those as the first hop.
+        """
+        limit = min(limit, StateDag.MAX_MISSING_EVENTS)
+        earliest_event_set = set(earliest_events)
+
+        seed_events = await self.store.get_events(latest_events)
+
+        seed_event_ids: list[str] = []
+        first_hop_event_ids: set[str] = set()
+        for event_id, event in seed_events.items():
+            if event_exists_in_state_dag(event):
+                seed_event_ids.append(event_id)
+                continue
+            # event is a message, so its prev_state_events are the first returned hop
+            assert supports_msc4242_state_dag(
+                event
+            )  # type-assert to access .prev_state_events
+            first_hop_event_ids.update(
+                prev_state_event_id
+                for prev_state_event_id in event.prev_state_events
+                if prev_state_event_id not in earliest_event_set
+            )
+
+        first_hop_event_ids.difference_update(seed_event_ids)
+        first_hop_events: list[EventBase] = []
+        if first_hop_event_ids:
+            first_hop_events = await self.store.get_events_as_list(
+                sorted(first_hop_event_ids)
+            )
+            first_hop_events.sort(key=lambda ev: ev.event_id)
+            first_hop_events = first_hop_events[:limit]
+            seed_event_ids.extend(ev.event_id for ev in first_hop_events)
+
+        missing_events = await self.store.get_missing_events_state_dag(
+            room_id=room_id,
+            earliest_event_ids=earliest_events,
+            latest_event_ids=seed_event_ids,
+            limit=limit - len(first_hop_events),
+        )
+        return first_hop_events + missing_events
+
     async def exchange_third_party_invite(
         self, sender_user_id: str, target_user_id: str, room_id: str, signed: JsonDict
     ) -> None:

synapse/handlers/room.py

@@ -1244,10 +1244,6 @@ async def create_room(
         creation_content = config.get("creation_content", {})
         # override any attempt to set room versions via the creation_content
         creation_content["room_version"] = room_version.identifier
-        # We do not currently support federating state DAG rooms.
-        # See related restriction in /send_join requests in federation_client.py.
-        if room_version.msc4242_state_dags:
-            creation_content[EventContentFields.FEDERATE] = False
 
         # trusted private chats have the invited users marked as additional creators
         if (

synapse/storage/databases/main/event_federation.py

@@ -1195,8 +1195,6 @@ def _get_auth_chain_difference_txn(
         # Return all events where not all sets can reach them.
         return {eid for eid, n in event_to_missing_sets.items() if n}
 
-    # FIXME(2026-04-22): Remove comment when used. Unused currently, but will be used in
-    # future MSC4242 PRs.
     async def get_state_dag(
         self, room_id: str, forward_extrems: set[str]
     ) -> dict[str, MSC4242Event]:
@@ -1260,8 +1258,6 @@ def _get_state_events_txn(txn: LoggingTransaction, room_id: str) -> list[str]:
 
         return result
 
-    # FIXME(2026-04-22): Remove comment when used. Unused currently, but will be used in
-    # future MSC4242 PRs.
     async def get_missing_events_state_dag(
         self,
         *,