diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs index fa40fdcfad..750c70814e 100644 --- a/rust/src/events/internal_metadata.rs +++ b/rust/src/events/internal_metadata.rs @@ -59,6 +59,7 @@ enum EventInternalMetadataData { TxnId(Box), TokenId(i64), DeviceId(Box), + CalculatedAuthEventIDs(Vec), } impl EventInternalMetadataData { @@ -123,6 +124,10 @@ impl EventInternalMetadataData { pyo3::intern!(py, "device_id"), o.into_pyobject(py).unwrap_infallible().into_any(), ), + EventInternalMetadataData::CalculatedAuthEventIDs(o) => ( + pyo3::intern!(py, "calculated_auth_event_ids"), + o.into_pyobject(py).unwrap().into_any(), + ), } } @@ -190,6 +195,11 @@ impl EventInternalMetadataData { .map(String::into_boxed_str) .with_context(|| format!("'{key_str}' has invalid type"))?, ), + "calculated_auth_event_ids" => EventInternalMetadataData::CalculatedAuthEventIDs( + value + .extract() + .with_context(|| format!("'{key_str}' has invalid type"))?, + ), _ => return Ok(None), }; @@ -472,6 +482,17 @@ impl EventInternalMetadata { set_property!(self, TxnId, obj.into_boxed_str()); } + /// The calculated auth event IDs, if it was set when the event was created. + #[getter] + fn get_calculated_auth_event_ids(&self) -> PyResult<&Vec> { + let s = get_property!(self, CalculatedAuthEventIDs)?; + Ok(s) + } + #[setter] + fn set_calculated_auth_event_ids(&mut self, obj: Vec) { + set_property!(self, CalculatedAuthEventIDs, obj); + } + /// The access token ID of the user who sent this event, if any. #[getter] fn get_token_id(&self) -> PyResult { diff --git a/synapse/api/room_versions.py b/synapse/api/room_versions.py index 97dac661a3..9aad68629a 100644 --- a/synapse/api/room_versions.py +++ b/synapse/api/room_versions.py @@ -37,6 +37,7 @@ class EventFormatVersions: ROOM_V3 = 2 # MSC1659-style $hash event id format: used for room v3 ROOM_V4_PLUS = 3 # MSC1884-style $hash format: introduced for room v4 ROOM_V11_HYDRA_PLUS = 4 # MSC4291 room IDs as hashes: introduced for room HydraV11 + ROOM_VMSC4242 = 5 KNOWN_EVENT_FORMAT_VERSIONS = { @@ -44,6 +45,7 @@ KNOWN_EVENT_FORMAT_VERSIONS = { EventFormatVersions.ROOM_V3, EventFormatVersions.ROOM_V4_PLUS, EventFormatVersions.ROOM_V11_HYDRA_PLUS, + EventFormatVersions.ROOM_VMSC4242, } @@ -112,6 +114,8 @@ class RoomVersion: msc3931_push_features: tuple[str, ...] # values from PushRuleRoomFlag # MSC3757: Restricting who can overwrite a state event msc3757_enabled: bool + # MSC4242: Creates events with prev_state_events instead of auth_events and derives state from it. + msc4242_state_dags: bool # MSC4289: Creator power enabled msc4289_creator_power_enabled: bool # MSC4291: Room IDs as hashes of the create event @@ -138,6 +142,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -160,6 +165,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -182,6 +188,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -204,6 +211,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -226,6 +234,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -248,6 +257,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -270,6 +280,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -292,6 +303,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -314,6 +326,7 @@ class RoomVersions: enforce_int_power_levels=False, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -336,6 +349,7 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -359,6 +373,7 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(PushRuleRoomFlag.EXTENSIBLE_EVENTS,), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -382,6 +397,7 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(), msc3757_enabled=True, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -404,6 +420,7 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -427,6 +444,7 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(), msc3757_enabled=True, + msc4242_state_dags=False, msc4289_creator_power_enabled=False, msc4291_room_ids_as_hashes=False, ) @@ -439,8 +457,8 @@ class RoomVersions: special_case_aliases_auth=False, strict_canonicaljson=True, limit_notifications_power_levels=True, - implicit_room_creator=True, # Used by MSC3820 - updated_redaction_rules=True, # Used by MSC3820 + implicit_room_creator=True, + updated_redaction_rules=True, restricted_join_rule=True, restricted_join_rule_fix=True, knock_join_rule=True, @@ -449,6 +467,7 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=True, # Changed from v11 msc4291_room_ids_as_hashes=True, # Changed from v11 ) @@ -471,9 +490,33 @@ class RoomVersions: enforce_int_power_levels=True, msc3931_push_features=(), msc3757_enabled=False, + msc4242_state_dags=False, msc4289_creator_power_enabled=True, # Changed from v11 msc4291_room_ids_as_hashes=True, # Changed from v11 ) + MSC4242v12 = RoomVersion( + "org.matrix.msc4242.12", # MSC4242 (State DAGs) based on room version "12" + RoomDisposition.UNSTABLE, + EventFormatVersions.ROOM_VMSC4242, + StateResolutionVersions.V2_1, + enforce_key_validity=True, + special_case_aliases_auth=False, + strict_canonicaljson=True, + limit_notifications_power_levels=True, + implicit_room_creator=True, + updated_redaction_rules=True, + restricted_join_rule=True, + restricted_join_rule_fix=True, + knock_join_rule=True, + msc3389_relation_redactions=False, + knock_restricted_join_rule=True, + enforce_int_power_levels=True, + msc3931_push_features=(), + msc3757_enabled=False, + msc4289_creator_power_enabled=True, + msc4291_room_ids_as_hashes=True, + msc4242_state_dags=True, + ) KNOWN_ROOM_VERSIONS: dict[str, RoomVersion] = { @@ -493,6 +536,7 @@ KNOWN_ROOM_VERSIONS: dict[str, RoomVersion] = { RoomVersions.V12, RoomVersions.MSC3757v10, RoomVersions.MSC3757v11, + RoomVersions.MSC4242v12, RoomVersions.HydraV11, ) } diff --git a/synapse/event_auth.py b/synapse/event_auth.py index 4b354dd6fb..5f5ea18cab 100644 --- a/synapse/event_auth.py +++ b/synapse/event_auth.py @@ -61,7 +61,7 @@ from synapse.api.room_versions import ( RoomVersion, RoomVersions, ) -from synapse.events import is_creator +from synapse.events import FrozenEventVMSC4242, is_creator from synapse.state import CREATE_KEY from synapse.storage.databases.main.events_worker import EventRedactBehaviour from synapse.types import ( @@ -186,6 +186,70 @@ async def check_state_independent_auth_rules( # 1.5 Otherwise, allow return + # State DAGs 2. Considering the event's prev_state_events: + if event.room_version.msc4242_state_dags: + prev_state_events_ids = set(cast(FrozenEventVMSC4242, event).prev_state_events) + needed_prev_state_event_ids = { + event_id + for event_id in prev_state_events_ids + if not batched_auth_events or event_id not in batched_auth_events + } + prev_state_events = ( + {} + if not batched_auth_events + else { + e: batched_auth_events[e] + for e in prev_state_events_ids - needed_prev_state_event_ids + } + ) + if len(needed_prev_state_event_ids) > 0: + needed_prev_state_events = await store.get_events( + needed_prev_state_event_ids, + redact_behaviour=EventRedactBehaviour.as_is, + allow_rejected=True, + ) + prev_state_events.update(needed_prev_state_events) + if len(prev_state_events) != len(prev_state_events_ids): + # we should have all the prev state events by now, so if we do not, that suggests + # a synapse programming error + known_prev_state_event_ids = set(prev_state_events) + raise RuntimeError( + f"Event {event.event_id} has unknown prev state events " + + f"{len(prev_state_events)} != {len(prev_state_events_ids)} " + + f"{prev_state_events_ids - known_prev_state_event_ids} missing " + + f"out of {prev_state_events_ids}" + ) + for prev_state_event in prev_state_events.values(): + # 2.1 If there are entries which do not belong in the same room, reject. + if prev_state_event.room_id != event.room_id: + raise AuthError( + 403, + "During auth for event %s in room %s, found event %s in prev state events " + "which is in room %s" + % ( + event.event_id, + event.room_id, + prev_state_event.event_id, + prev_state_event.room_id, + ), + ) + # 2.2 If there are entries which do not have a state_key, reject. + if not prev_state_event.is_state(): + 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 not state: {prev_state_event.event_id}", + ) + # 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: + 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.event_id}", + ) + # 2. Reject if event has auth_events that: ... auth_events: ChainMap[str, EventBase] = ChainMap() if batched_auth_events: @@ -471,6 +535,13 @@ def _check_create(event: "EventBase") -> None: if event.prev_event_ids(): raise AuthError(403, "Create event has prev events") + # State DAGs 1.2 If it has any prev_state_events, reject. + if ( + event.room_version.msc4242_state_dags + and cast(FrozenEventVMSC4242, event).prev_state_events + ): + raise AuthError(403, "Create event has prev state events") + if event.room_version.msc4291_room_ids_as_hashes: # 1.2 If the create event has a room_id, reject if "room_id" in event: diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index c7eaf7eda2..adec922a75 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -28,6 +28,7 @@ from typing import ( Generic, Iterable, Literal, + Optional, TypeVar, Union, overload, @@ -39,10 +40,7 @@ from unpaddedbase64 import encode_base64 from synapse.api.constants import EventContentFields, EventTypes, RelationTypes from synapse.api.room_versions import EventFormatVersions, RoomVersion, RoomVersions from synapse.synapse_rust.events import EventInternalMetadata -from synapse.types import ( - JsonDict, - StrCollection, -) +from synapse.types import JsonDict, StateKey, StrCollection from synapse.util.caches import intern_dict from synapse.util.frozenutils import freeze @@ -501,8 +499,8 @@ class FrozenEventV4(FrozenEventV3): self, event_dict: JsonDict, room_version: RoomVersion, - internal_metadata_dict: JsonDict | None = None, - rejected_reason: str | None = None, + internal_metadata_dict: Optional[JsonDict] = None, + rejected_reason: Optional[str] = None, ): super().__init__( event_dict=event_dict, @@ -551,9 +549,59 @@ class FrozenEventV4(FrozenEventV3): return [*self._dict["auth_events"], create_event_id] +class FrozenEventVMSC4242(FrozenEventV4): + """FrozenEventVMSC4242, which differs from FrozenEventV4 only in the addition of prev_state_events""" + + format_version = EventFormatVersions.ROOM_VMSC4242 + prev_state_events: DictProperty[list[str]] = DictProperty("prev_state_events") + + def __init__( + self, + event_dict: JsonDict, + room_version: RoomVersion, + internal_metadata_dict: Optional[JsonDict] = None, + rejected_reason: Optional[str] = None, + ): + # Similar to how we assert event_id isn't in V2+ events, we do the same with auth_events. + assert "auth_events" not in event_dict + super().__init__( + event_dict=event_dict, + room_version=room_version, + internal_metadata_dict=internal_metadata_dict, + rejected_reason=rejected_reason, + ) + + def auth_event_ids(self) -> StrCollection: + """Returns the list of _calculated_ auth event IDs. + + Returns: + The list of event IDs of this event's auth events + """ + # Catches cases where we accidentally call auth_event_ids() prior to calculating what they + # actually are. The exception being the m.room.create event which has no auth events. + if self.type != EventTypes.Create: + assert len(self.internal_metadata.calculated_auth_event_ids) > 0 + return self.internal_metadata.calculated_auth_event_ids + + def __repr__(self) -> str: + rejection = f"REJECTED={self.rejected_reason}, " if self.rejected_reason else "" + + return ( + f"<{self.__class__.__name__} " + f"{rejection}" + f"event_id={self.event_id}, " + f"type={self.get('type')}, " + f"state_key={self.get('state_key')}, " + f"prev_events={self.get('prev_events')}, " + f"prev_state_events={self.get('prev_state_events')}, " + f"outlier={self.internal_metadata.is_outlier()}" + ">" + ) + + def _event_type_from_format_version( format_version: int, -) -> type[FrozenEvent | FrozenEventV2 | FrozenEventV3]: +) -> type[FrozenEvent | FrozenEventV2 | FrozenEventV3 | FrozenEventVMSC4242]: """Returns the python type to use to construct an Event object for the given event format version. @@ -570,6 +618,8 @@ def _event_type_from_format_version( return FrozenEventV2 elif format_version == EventFormatVersions.ROOM_V4_PLUS: return FrozenEventV3 + elif format_version == EventFormatVersions.ROOM_VMSC4242: + return FrozenEventVMSC4242 elif format_version == EventFormatVersions.ROOM_V11_HYDRA_PLUS: return FrozenEventV4 else: @@ -631,6 +681,24 @@ def relation_from_event(event: EventBase) -> _EventRelation | None: return _EventRelation(parent_id, rel_type, aggregation_key) +def event_exists_in_state_dag( + event: Union["EventBase", "EventBuilder", "EventMetadata", "StateKey"], +) -> bool: + """Given an event, returns true if this event should form part of the state DAG. + Only valid for room versions which use a state DAG.""" + state_key = None + if isinstance(event, EventMetadata): + state_key = event.state_key + elif isinstance(event, tuple): # StateKey + # can't use StateKey else you get: + # "Subscripted generics cannot be used with class and instance checks" + state_key = event[1] + else: + state_key = event.state_key if event.is_state() else None + + return state_key is not None + + def is_creator(create: EventBase, user_id: str) -> bool: """ Return true if the provided user ID is the room creator. @@ -665,3 +733,13 @@ class StrippedStateEvent: state_key: str sender: str content: dict[str, Any] + + +@attr.s(slots=True, frozen=True, auto_attribs=True) +class EventMetadata: + """Returned by `get_metadata_for_events`""" + + room_id: str + event_type: str + state_key: Optional[str] + rejection_reason: Optional[str] diff --git a/synapse/events/builder.py b/synapse/events/builder.py index 6a2812109d..695c2c44dd 100644 --- a/synapse/events/builder.py +++ b/synapse/events/builder.py @@ -128,6 +128,7 @@ class EventBuilder: prev_event_ids: list[str], auth_event_ids: list[str] | None, depth: int | None = None, + prev_state_events: list[str] | None = None, ) -> EventBase: """Transform into a fully signed and hashed event @@ -139,10 +140,46 @@ class EventBuilder: depth: Override the depth used to order the event in the DAG. Should normally be set to None, which will cause the depth to be calculated based on the prev_events. - + prev_state_events: The event IDs to use as prev_state_events. + Only applicable on MSC4242v11 rooms. If this is supplied, auth_event_ids + is ignored. Returns: The signed and hashed event. """ + if self.room_version.msc4242_state_dags: + assert prev_state_events is not None + if self.room_id: + state_ids = await self._state.compute_state_after_events( + self.room_id, + prev_state_events, + state_filter=StateFilter.from_types( + auth_types_for_event(self.room_version, self) + ), + await_full_state=False, + ) + # When we create rooms we only insert the create+member events, and batch the rest. + # Therefore, we may not have state_ids from compute_state_after_events as the + # prev_state_events are unknown. If this happens, the caller provides the auth events + # to use instead. + calculated_auth_event_ids: list[ + str + ] = [] # assume it's the create event which has [] + if len(state_ids) == 0 and len(prev_state_events) > 0: + # it's a batched event, so we should have been provided the auth_events + assert auth_event_ids and len(auth_event_ids) > 0 + calculated_auth_event_ids = auth_event_ids + else: + calculated_auth_event_ids = ( + self._event_auth_handler.compute_auth_events(self, state_ids) + ) + else: + # event is a state DAG event and is the create event (room_id is not provided), + # therefore there are no auth_events. + calculated_auth_event_ids = [] + self.internal_metadata.calculated_auth_event_ids = calculated_auth_event_ids + auth_event_ids = calculated_auth_event_ids + + # this block must not be hit for MSC4242 rooms as it resolves state with prev_events # Create events always have empty auth_events. if self.type == EventTypes.Create and self.is_state() and self.state_key == "": auth_event_ids = [] @@ -227,7 +264,6 @@ class EventBuilder: # rejected by other servers (and so that they can be persisted in # the db) depth = min(depth, MAX_DEPTH) - event_dict: dict[str, Any] = { "auth_events": auth_events, "prev_events": prev_events, @@ -237,8 +273,6 @@ class EventBuilder: "unsigned": self.unsigned, "depth": depth, } - if self.room_id is not None: - event_dict["room_id"] = self.room_id if self.room_version.msc4291_room_ids_as_hashes: # In MSC4291: the create event has no room ID as the create event ID /is/ the room ID. @@ -258,6 +292,13 @@ class EventBuilder: auth_event_ids.remove(create_event_id) event_dict["auth_events"] = auth_event_ids + if self.room_version.msc4242_state_dags and prev_state_events is not None: + # Auth events are removed entirely on state DAG rooms + event_dict.pop("auth_events") + event_dict["prev_state_events"] = prev_state_events + if self.room_id is not None: + event_dict["room_id"] = self.room_id + if self.is_state(): event_dict["state_key"] = self._state_key diff --git a/synapse/events/utils.py b/synapse/events/utils.py index b79a68f589..fedf7ee72d 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -145,6 +145,10 @@ def prune_event_dict(room_version: RoomVersion, event_dict: JsonDict) -> JsonDic # Earlier room versions from had additional allowed keys. if not room_version.updated_redaction_rules: allowed_keys.extend(["prev_state", "membership", "origin"]) + # Custom room versions add new allowed keys and remove others + if room_version.msc4242_state_dags: + allowed_keys.extend(["prev_state_events"]) + allowed_keys.remove("auth_events") event_type = event_dict["type"] diff --git a/synapse/events/validator.py b/synapse/events/validator.py index b27f8a942a..ff22b2287f 100644 --- a/synapse/events/validator.py +++ b/synapse/events/validator.py @@ -63,14 +63,17 @@ class EventValidator: if event.format_version == EventFormatVersions.ROOM_V1_V2: EventID.from_string(event.event_id) - required = [ + required = { "auth_events", "content", "hashes", "prev_events", "sender", "type", - ] + } + if event.room_version.msc4242_state_dags: + required.remove("auth_events") + required.add("prev_state_events") for k in required: if k not in event: diff --git a/synapse/handlers/admin.py b/synapse/handlers/admin.py index 2fb0e5814f..2e9bf33176 100644 --- a/synapse/handlers/admin.py +++ b/synapse/handlers/admin.py @@ -485,9 +485,21 @@ class AdminHandler: event_dict["redacts"] = event.event_id try: + prev_state_events = None + if room_version.msc4242_state_dags: + # TODO(kegan): better way to typecast and get at this field? + prev_state_events = event.get_dict().get( + "prev_state_events", None + ) + if prev_state_events is None: + # should be impossible as prev_state_events is a required field. + raise Exception( + "cannot set prev_state_events for redaction event as parent has " + + "no prev_state_events", + ) # set the prev event to the offending message to allow for redactions # to be processed in the case where the user has been kicked/banned before - # redactions are requested + # redactions are requested. ( redaction, _, @@ -496,6 +508,7 @@ class AdminHandler: event_dict, prev_event_ids=[event.event_id], ratelimit=False, + prev_state_events=prev_state_events, ) except Exception as ex: logger.info( diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 99ce120736..d5cb46090f 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -53,7 +53,7 @@ from synapse.api.errors import ( from synapse.api.room_versions import KNOWN_ROOM_VERSIONS from synapse.api.urls import ConsentURIBuilder from synapse.event_auth import validate_event_for_room_version -from synapse.events import EventBase, relation_from_event +from synapse.events import EventBase, FrozenEventVMSC4242, relation_from_event from synapse.events.builder import EventBuilder from synapse.events.snapshot import ( EventContext, @@ -585,6 +585,7 @@ class EventCreationHandler: state_map: StateMap[str] | None = None, for_batch: bool = False, current_state_group: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[EventBase, UnpersistedEventContextBase]: """ Given a dict from a client, create a new event. If bool for_batch is true, will @@ -639,6 +640,10 @@ class EventCreationHandler: current_state_group: the current state group, used only for creating events for batch persisting + prev_state_events: + The state event IDs which represent the current forward extremities of the state DAG. + Only applicable on room versions which use a state DAG. + Raises: ResourceLimitError if server is blocked to some resource being exceeded @@ -738,6 +743,7 @@ class EventCreationHandler: state_map=state_map, for_batch=for_batch, current_state_group=current_state_group, + prev_state_events=prev_state_events, ) # In an ideal world we wouldn't need the second part of this condition. However, @@ -966,6 +972,7 @@ class EventCreationHandler: ignore_shadow_ban: bool = False, outlier: bool = False, depth: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[EventBase, int]: """ Creates an event, then sends it. @@ -994,7 +1001,9 @@ class EventCreationHandler: depth: Override the depth used to order the event in the DAG. Should normally be set to None, which will cause the depth to be calculated based on the prev_events. - + prev_state_events: + The state event IDs which represent the current forward extremities of the state DAG. + Only applicable on room versions which use a state DAG. Returns: The event, and its stream ordering (if deduplication happened, the previous, duplicate event). @@ -1090,6 +1099,7 @@ class EventCreationHandler: ignore_shadow_ban=ignore_shadow_ban, outlier=outlier, depth=depth, + prev_state_events=prev_state_events, ) async def _create_and_send_nonmember_event_locked( @@ -1103,6 +1113,7 @@ class EventCreationHandler: ignore_shadow_ban: bool = False, outlier: bool = False, depth: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[EventBase, int]: room_id = event_dict["room_id"] @@ -1131,6 +1142,7 @@ class EventCreationHandler: state_event_ids=state_event_ids, outlier=outlier, depth=depth, + prev_state_events=prev_state_events, ) context = await unpersisted_context.persist(event) @@ -1225,6 +1237,7 @@ class EventCreationHandler: state_map: StateMap[str] | None = None, for_batch: bool = False, current_state_group: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[EventBase, UnpersistedEventContextBase]: """Create a new event for a local client. If bool for_batch is true, will create an event using the prev_event_ids, and will create an event context for @@ -1266,9 +1279,29 @@ class EventCreationHandler: current_state_group: the current state group, used only for creating events for batch persisting + prev_state_events: + The state event IDs which represent the current forward extremities of the state DAG. + Only applicable on room versions which use a state DAG. If unset, populates them + from the current state dag forward extremities. + Returns: Tuple of created event, UnpersistedEventContext """ + if builder.room_version.msc4242_state_dags: + assert auth_event_ids is None + # (kegan) I can't find any call-site which uses this. We can't risk letting in + # untrusted input, so for now assert that we aren't told about any state. + assert state_event_ids is None + + if builder.room_id: + if prev_state_events is None: + prev_state_events = list( + await self.store.get_state_dag_extremities(builder.room_id) + ) + else: + # create event doesn't need prev_state_events to be fetched, but it must be non-None. + prev_state_events = [] + # Strip down the state_event_ids to only what we need to auth the event. # For example, we don't need extra m.room.member that don't match event.sender if state_event_ids is not None: @@ -1342,7 +1375,10 @@ class EventCreationHandler: assert state_map is not None auth_ids = self._event_auth_handler.compute_auth_events(builder, state_map) event = await builder.build( - prev_event_ids=prev_event_ids, auth_event_ids=auth_ids, depth=depth + prev_event_ids=prev_event_ids, + auth_event_ids=auth_ids, + depth=depth, + prev_state_events=prev_state_events, ) context: UnpersistedEventContextBase = ( @@ -1359,6 +1395,7 @@ class EventCreationHandler: prev_event_ids=prev_event_ids, auth_event_ids=auth_event_ids, depth=depth, + prev_state_events=prev_state_events, ) # Pass on the outlier property from the builder to the event @@ -1548,6 +1585,17 @@ class EventCreationHandler: auth_event = event_id_to_event.get(event_id) if auth_event: batched_auth_events[event_id] = auth_event + if event.room_version.msc4242_state_dags and isinstance( + event, FrozenEventVMSC4242 + ): + # State DAG rooms will check that the prev_state_events are not rejected. + # To do that, we need to make sure we pass in the prev_state_events as + # batched_auth_events, else we will fail the event due to the + # prev_state_events not existing in the database. + for event_id in event.prev_state_events: + pse = event_id_to_event.get(event_id) + if pse: + batched_auth_events[event_id] = pse await self._event_auth_handler.check_auth_rules_from_context( event, batched_auth_events ) @@ -1802,7 +1850,10 @@ class EventCreationHandler: # set for a while, so that the expiry time is reset. state_entry = await self.state.resolve_state_groups_for_events( - event.room_id, event_ids=event.prev_event_ids() + event.room_id, + event_ids=event.prev_state_events + if isinstance(event, FrozenEventVMSC4242) + else event.prev_event_ids(), ) if state_entry.state_group: @@ -2345,9 +2396,16 @@ class EventCreationHandler: # case. prev_event_ids = await self.store.get_prev_events_for_room(builder.room_id) + prev_state_events = None + if original_event.room_version.msc4242_state_dags: + prev_state_events = list( + await self.store.get_state_dag_extremities(builder.room_id) + ) + event = await builder.build( prev_event_ids=prev_event_ids, auth_event_ids=None, + prev_state_events=prev_state_events, ) # we rebuild the event context, to be on the safe side. If nothing else, diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index e03a912319..d614535b40 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -65,7 +65,7 @@ from synapse.api.filtering import Filter from synapse.api.ratelimiting import Ratelimiter from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion 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.snapshot import UnpersistedEventContext from synapse.events.utils import copy_and_fixup_power_levels_contents from synapse.handlers.relations import BundledAggregations @@ -1486,6 +1486,9 @@ class RoomCreationHandler: # the most recently created event prev_event: list[str] = [] + # the most recently created state event + prev_state_event: list[str] = [] + # a map of event types, state keys -> event_ids. We collect these mappings this as events are # created (but not persisted to the db) to determine state for future created events # (as this info can't be pulled from the db) @@ -1512,6 +1515,7 @@ class RoomCreationHandler: """ nonlocal depth nonlocal prev_event + nonlocal prev_state_event # Create the event dictionary. event_dict = {"type": etype, "content": content} @@ -1525,6 +1529,7 @@ class RoomCreationHandler: creator, event_dict, prev_event_ids=prev_event, + prev_state_events=prev_state_event, depth=depth, # Take a copy to ensure each event gets a unique copy of # state_map since it is modified below. @@ -1535,7 +1540,8 @@ class RoomCreationHandler: depth += 1 prev_event = [new_event.event_id] state_map[(new_event.type, new_event.state_key)] = new_event.event_id - + if room_version.msc4242_state_dags and event_exists_in_state_dag(new_event): + prev_state_event = [new_event.event_id] return new_event, new_unpersisted_context preset_config, config = self._room_preset_config(room_config) @@ -1568,6 +1574,7 @@ class RoomCreationHandler: ignore_shadow_ban=True, ) last_sent_event_id = ev.event_id + prev_state_event = [ev.event_id] member_event_id, _ = await self.room_member_handler.update_membership( creator, @@ -1579,8 +1586,10 @@ class RoomCreationHandler: new_room=True, prev_event_ids=[last_sent_event_id], depth=depth, + prev_state_events=prev_state_event, ) prev_event = [member_event_id] + prev_state_event = [member_event_id] # update the depth and state map here as the membership event has been created # through a different code path diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py index a8935fded6..bb80d09c34 100644 --- a/synapse/handlers/room_member.py +++ b/synapse/handlers/room_member.py @@ -36,9 +36,11 @@ from synapse.api.constants import ( from synapse.api.errors import ( AuthError, Codes, + NotFoundError, PartialStateConflictError, ShadowBanError, SynapseError, + UnsupportedRoomVersionError, ) from synapse.api.ratelimiting import Ratelimiter from synapse.event_auth import get_named_level, get_power_level_event @@ -406,6 +408,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): require_consent: bool = True, outlier: bool = False, origin_server_ts: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[str, int]: """ Internal membership update function to get an existing event or create @@ -490,6 +493,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): depth=depth, require_consent=require_consent, outlier=outlier, + prev_state_events=prev_state_events, ) context = await unpersisted_context.persist(event) prev_state_ids = await context.get_prev_state_ids( @@ -585,6 +589,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): state_event_ids: list[str] | None = None, depth: int | None = None, origin_server_ts: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[str, int]: """Update a user's membership in a room. @@ -610,6 +615,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): by the MSC2716 /batch_send endpoint. This should normally be left as None, which will cause the auth_event_ids to be calculated based on the room state at the prev_events. + TODO(kegan): I don't see this being used anywhere. depth: Override the depth used to order the event in the DAG. Should normally be set to None, which will cause the depth to be calculated based on the prev_events. @@ -677,6 +683,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): state_event_ids=state_event_ids, depth=depth, origin_server_ts=origin_server_ts, + prev_state_events=prev_state_events, ) return result @@ -699,6 +706,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): state_event_ids: list[str] | None = None, depth: int | None = None, origin_server_ts: int | None = None, + prev_state_events: list[str] | None = None, ) -> tuple[str, int]: """Helper for update_membership. @@ -941,9 +949,20 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): require_consent=require_consent, outlier=outlier, origin_server_ts=origin_server_ts, + prev_state_events=prev_state_events, ) - latest_event_ids = await self.store.get_prev_events_for_room(room_id) + is_state_dags = False + try: + room_version = await self.store.get_room_version(room_id) + is_state_dags = room_version.msc4242_state_dags + except (NotFoundError, UnsupportedRoomVersionError): + pass + + if is_state_dags: + latest_event_ids = list(await self.store.get_state_dag_extremities(room_id)) + else: + latest_event_ids = await self.store.get_prev_events_for_room(room_id) is_partial_state_room = await self.store.is_partial_state_room(room_id) partial_state_before_join = await self.state_handler.compute_state_after_events( @@ -1154,6 +1173,8 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): # see: https://github.com/matrix-org/synapse/issues/7139 if len(latest_event_ids) == 0: latest_event_ids = [invite.event_id] + # TODO(kegan): room version check? + prev_state_events = [invite.event_id] # or perhaps this is a remote room that a local user has knocked on elif current_membership_type == Membership.KNOCK: @@ -1199,6 +1220,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): require_consent=require_consent, outlier=outlier, origin_server_ts=origin_server_ts, + prev_state_events=prev_state_events, ) async def check_for_any_membership_in_room( @@ -2084,6 +2106,12 @@ class RoomMemberMasterHandler(RoomMemberHandler): auth_event_ids = ( list(previous_membership_event.auth_event_ids()) + prev_event_ids ) + # Authorise the leave by referencing the previous membership + prev_state_events = None + if previous_membership_event.room_version.msc4242_state_dags: + prev_state_events = [ + previous_membership_event.event_id, + ] # Try several times, it could fail with PartialStateConflictError # in handle_new_client_event, cf comment in except block. @@ -2100,6 +2128,7 @@ class RoomMemberMasterHandler(RoomMemberHandler): prev_event_ids=prev_event_ids, auth_event_ids=auth_event_ids, outlier=True, + prev_state_events=prev_state_events, ) context = await unpersisted_context.persist(event) event.internal_metadata.out_of_band_membership = True diff --git a/synapse/state/__init__.py b/synapse/state/__init__.py index a92233c863..0f671e3100 100644 --- a/synapse/state/__init__.py +++ b/synapse/state/__init__.py @@ -37,7 +37,7 @@ from prometheus_client import Counter, Histogram from synapse.api.constants import EventTypes from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, StateResolutionVersions -from synapse.events import EventBase +from synapse.events import EventBase, FrozenEventVMSC4242 from synapse.events.snapshot import ( EventContext, UnpersistedEventContext, @@ -239,6 +239,7 @@ class StateHandler: ) return await ret.get_state(self._state_storage_controller, state_filter) + # TODO(kegan): this is unused? async def get_current_user_ids_in_room( self, room_id: str, latest_event_ids: StrCollection ) -> set[str]: @@ -303,7 +304,8 @@ class StateHandler: membership events. `False` if `state_ids_before_event` is the full state. `None` when `state_ids_before_event` is not provided. In this case, the - flag will be calculated based on `event`'s prev events. + flag will be calculated based on `event`'s prev events or prev_state_events + for state DAG rooms. state_group_before_event: the current state group at the time of event, if known Returns: @@ -337,7 +339,11 @@ class StateHandler: # (This is slightly racy - the prev-events might get fixed up before we use # their states - but I don't think that really matters; it just means we # might redundantly recalculate the state for this event later.) - prev_event_ids = event.prev_event_ids() + prev_event_ids = frozenset( + event.prev_state_events + if isinstance(event, FrozenEventVMSC4242) + else event.prev_event_ids() + ) incomplete_prev_events = await self.store.get_partial_state_events( prev_event_ids ) @@ -355,7 +361,7 @@ class StateHandler: entry = await self.resolve_state_groups_for_events( event.room_id, - event.prev_event_ids(), + prev_event_ids, await_full_state=False, ) diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 2948227807..b3d9095d44 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -35,6 +35,7 @@ from typing import ( Generic, Iterable, TypeVar, + cast, ) import attr @@ -43,7 +44,9 @@ from prometheus_client import Counter, Histogram from twisted.internet import defer from synapse.api.constants import EventTypes, Membership -from synapse.events import EventBase +from synapse.api.errors import SynapseError +from synapse.api.room_versions import KNOWN_ROOM_VERSIONS +from synapse.events import EventBase, FrozenEventVMSC4242, event_exists_in_state_dag from synapse.events.snapshot import EventContext, EventPersistencePair from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable @@ -111,6 +114,14 @@ stale_forward_extremities_counter = Histogram( buckets=(0, 1, 2, 3, 5, 7, 10, 15, 20, 50, 100, 200, 500, "+Inf"), ) +# The number of forward extremities for each new event. +msc4242_state_dag_forward_extremities_counter = Histogram( + "synapse_storage_msc4242_state_dag_forward_extremities_persisted", + "Number of forward extremities for each new event in the state DAG", + labelnames=[SERVER_NAME_LABEL], + buckets=(1, 2, 3, 5, 7, 10, 15, 20, 50, 100, 200, 500, "+Inf"), +) + state_resolutions_during_persistence = Counter( "synapse_storage_events_state_resolutions_during_persistence", "Number of times we had to do state res to calculate new current state", @@ -529,7 +540,15 @@ class EventsPersistenceStorageController: Returns: map from (type, state_key) to event id for the current state in the room """ - latest_event_ids = await self.main_store.get_latest_event_ids_in_room(room_id) + room_version = await self.main_store.get_room_version_id(room_id) + room_version_obj = KNOWN_ROOM_VERSIONS[room_version] + if room_version_obj.msc4242_state_dags: + latest_event_ids = await self.main_store.get_state_dag_extremities(room_id) + else: + latest_event_ids = await self.main_store.get_latest_event_ids_in_room( + room_id + ) + state_groups = set( ( await self.main_store._get_state_group_for_events(latest_event_ids) @@ -551,7 +570,6 @@ class EventsPersistenceStorageController: # Avoid a circular import. from synapse.state import StateResolutionStore - room_version = await self.main_store.get_room_version_id(room_id) res = await self._state_resolution_handler.resolve_state_groups( room_id, room_version, @@ -615,28 +633,56 @@ class EventsPersistenceStorageController: for x in range(0, len(events_and_contexts), 100) ] + # Get the room version for the first event. This room version is the same for all events + # as events_and_contexts is all for one room. + room_version = ( + events_and_contexts[0][0].room_version + if len(events_and_contexts) > 0 + else None + ) + for chunk in chunks: # We can't easily parallelize these since different chunks # might contain the same event. :( new_forward_extremities = None state_delta_for_room = None + new_state_dag_extrems = None if not backfilled: - with Measure( - self._clock, - name="_calculate_state_and_extrem", - server_name=self.server_name, - ): - # Work out the new "current state" for the room. - # We do this by working out what the new extremities are and then - # calculating the state from that. - ( - new_forward_extremities, - state_delta_for_room, - ) = await self._calculate_new_forward_extremities_and_state_delta( - room_id, chunk - ) + if room_version and room_version.msc4242_state_dags: + with Measure( + self._clock, + name="_process_state_dag_forward_extremities_and_state_delta", + server_name=self.server_name, + ): + ( + new_forward_extremities, # for prev_events + state_delta_for_room, # for state groups + new_state_dag_extrems, # for prev_state_events + ) = await self._process_state_dag_forward_extremities_and_state_delta( + room_id, + [ + (ev, ctx) + for ev, ctx in chunk + if isinstance(ev, FrozenEventVMSC4242) + ], + ) + else: + with Measure( + self._clock, + name="_calculate_state_and_extrem", + server_name=self.server_name, + ): + # Work out the new "current state" for the room. + # We do this by working out what the new extremities are and then + # calculating the state from that. + ( + new_forward_extremities, + state_delta_for_room, + ) = await self._calculate_new_forward_extremities_and_state_delta( + room_id, chunk + ) with Measure( self._clock, @@ -666,6 +712,7 @@ class EventsPersistenceStorageController: use_negative_stream_ordering=backfilled, inhibit_local_membership_updates=backfilled, new_event_links=new_event_links, + new_state_dag_forward_extremities=new_state_dag_extrems, ) return replaced_events @@ -793,6 +840,207 @@ class EventsPersistenceStorageController: return (new_forward_extremities, delta) + async def _process_state_dag_forward_extremities_and_state_delta( + self, + room_id: str, + event_contexts: list[tuple[FrozenEventVMSC4242, EventContext]], + ) -> tuple[set[str] | None, DeltaState | None, set[str] | None]: + """Process the forwards extremities for state DAG rooms. + Returns: + - the new room dag extremities which should be written when these events are persisted. + - the state delta for the room, if applicable. + - the new state dag extremities which should be written when these events are persisted. + + NB: this does not write them because if it did, new events may see them _before_ the events + get persisted, causing failures in retrieving state groups. + """ + # Update forward extremities + # ...for the state DAG + existing_state_dag_fwd_extrems = ( + await self.main_store.get_state_dag_extremities(room_id) + ) + new_state_dag_fwd_extrems = await self._calculate_new_state_dag_extremities( + room_id, + existing_state_dag_fwd_extrems, + event_contexts, + ) + assert new_state_dag_fwd_extrems, ( + f"No state dag forward extremities left in room {room_id}!" + ) + # ...and the room DAG + existing_room_dag_fwd_extrems = ( + await self.main_store.get_latest_event_ids_in_room(room_id) + ) + new_room_dag_fwd_extrems = await self._calculate_new_extremities( + room_id, + cast(list[EventPersistencePair], event_contexts), + existing_room_dag_fwd_extrems, + ) + assert new_room_dag_fwd_extrems, ( + f"No room dag forward extremities left in room {room_id}!" + ) + + # See if we need to calculate a state delta + if new_state_dag_fwd_extrems == existing_state_dag_fwd_extrems: + # No change in state extremities, so no new state to calculate + return new_room_dag_fwd_extrems, None, new_state_dag_fwd_extrems + + with Measure( + self._clock, + name="persist_events.state_dag.get_new_state_after_events", + server_name=self.server_name, + ): + (current_state, delta_ids, _) = await self._get_new_state_after_events( + room_id, + cast(list[EventPersistencePair], event_contexts), + existing_state_dag_fwd_extrems, + new_state_dag_fwd_extrems, + should_prune=False, + ) + + # Following logic cargoculted from _calculate_new_forward_extremities_and_state_delta + # If either are not None then there has been a change, + # and we need to work out the delta (or use that + # given) + delta = None + if delta_ids is not None: + # If there is a delta we know that we've + # only added or replaced state, never + # removed keys entirely. + delta = DeltaState([], delta_ids) + elif current_state is not None: + with Measure( + self._clock, + name="persist_events.calculate_state_delta", + server_name=self.server_name, + ): + delta = await self._calculate_state_delta(room_id, current_state) + + if delta: + # If we have a change of state then lets check + # whether we're actually still a member of the room, + # or if our last user left. If we're no longer in + # the room then we delete the current state and + # extremities. + is_still_joined = await self._is_server_still_joined( + room_id, + cast(list[EventPersistencePair], event_contexts), + delta, + ) + if not is_still_joined: + logger.info("Server no longer in room %s", room_id) + delta.no_longer_in_room = True + + return new_room_dag_fwd_extrems, delta, new_state_dag_fwd_extrems + + async def _calculate_new_state_dag_extremities( + self, + room_id: str, + existing_fwd_extrems: frozenset[str], + event_contexts: list[tuple[FrozenEventVMSC4242, EventContext]], + ) -> set[str]: + """Calculate the new state dag forward extremities. Modifies existing_fwd_extrems. + + Assumes that new_state_events are only state events which should be in the state DAG. + """ + # filter out events which don't belong in the state dag. + new_state_events_contexts = [ + (e, ctx) + for e, ctx in event_contexts + if event_exists_in_state_dag(e) and isinstance(e, FrozenEventVMSC4242) + ] + if len(new_state_events_contexts) == 0: + # if there are no state events being persisted, then the fwd extremities of the state dag + # do not change. + return set(existing_fwd_extrems) + + # This logic is very similar to _calculate_new_extremities with a few key differences: + # - We do not "Remove any events which are prev_events of any existing events." because the + # state DAG mandates that events are processed in causal order, so there MUST NOT be any + # existing, processed events which have the to-be-persisted events as prev_state_events. + # - We don't care if they are an "outlier" in the main room dag, so long as they AREN'T + # an outlier on the state dag, which this function checks, so we don't check outlier-ness. + # - We allow soft-failed events to become forward extremities, as per the MSC. We do not + # allow rejected events to become forward extremities though. + + rejected_events = [ev for ev, ctx in new_state_events_contexts if ctx.rejected] + new_state_events = [ + ev for ev, ctx in new_state_events_contexts if not ctx.rejected + ] + # We include rejected_event_ids in this check + # otherwise if we persist a chunk of (R <- O) then O would fail the check since R was not in + # the DB of the to-be-persisted chunk yet (remember all the events in the chunk aren't + # persisted yet!) + all_events = set(rejected_events + new_state_events) + + # First, verify that we know all prev_state_events. If we fail this check then we don't have + # a complete DAG and that is bad, so bail out. + + # Start with them all missing. + missing_prev_state_events = { + e_id for event in all_events for e_id in event.prev_state_events + } + + # remove prev events which appear in all_events + missing_prev_state_events.difference_update( + event.event_id for event in all_events + ) + # the rest of these events should be present in the DB. Some of them may be forward extremities, + # some may not be, that's ok. + seen_events = await self.main_store.have_seen_events( + room_id, + missing_prev_state_events, + ) + missing_prev_state_events.difference_update(seen_events) + + if len(missing_prev_state_events) > 0: + logger.error( + "_calculate_new_state_dag_extremities: missing the following prev_state_events in room %s : %s", + room_id, + missing_prev_state_events, + ) + msg = (ev.event_id for ev in all_events) + logger.error( + "_calculate_new_state_dag_extremities: was handling %s", + msg, + ) + raise SynapseError( + code=500, + msg=f"missing {len(missing_prev_state_events)} prev_state_events in room {room_id}", + ) + + # Now calculate the forward extremities. + + # start with the existing forward extremities + result = set(existing_fwd_extrems) + + # add all the new events to the list + result.update(event.event_id for event in new_state_events) + + # Now remove all events which are prev_state_events of any of the new events + result.difference_update( + e_id for event in new_state_events for e_id in event.prev_state_events + ) + + # Finally handle the case where the new events have rejected prev + # events. If they do we need to remove them and their prev events, + # otherwise we end up with dangling extremities. + # Specifically, this handles the case where (F=fwd extrem, SF=soft-failed, N=new event) + # F <-- SF <-- SF <-- N + # where we want to remove F as a forward extremity and replace with N. + existing_prevs = await self.persist_events_store._get_prevs_before_rejected( + (e_id for event in new_state_events for e_id in event.prev_state_events), + include_soft_failed=False, + ) + result.difference_update(existing_prevs) + + # We only update metrics for events that change forward extremities + if result != existing_fwd_extrems: + msc4242_state_dag_forward_extremities_counter.labels( + **{SERVER_NAME_LABEL: self.server_name} + ).observe(len(result)) + return result + async def _calculate_new_extremities( self, room_id: str, @@ -859,6 +1107,7 @@ class EventsPersistenceStorageController: events_context: list[EventPersistencePair], old_latest_event_ids: AbstractSet[str], new_latest_event_ids: set[str], + should_prune: bool = True, ) -> tuple[StateMap[str] | None, StateMap[str] | None, set[str]]: """Calculate the current state dict after adding some new events to a room @@ -876,6 +1125,9 @@ class EventsPersistenceStorageController: new_latest_event_ids : the new forward extremities for the room. + should_prune : + if true, attempt to prune the forward extremities. + Returns: Returns a tuple of two state maps and a set of new forward extremities. @@ -1015,7 +1267,7 @@ class EventsPersistenceStorageController: # If the returned state matches the state group of one of the new # forward extremities then we check if we are able to prune some state # extremities. - if res.state_group and res.state_group in new_state_groups: + if should_prune and res.state_group and res.state_group in new_state_groups: new_latest_event_ids = await self._prune_extremities( room_id, new_latest_event_ids, diff --git a/synapse/storage/databases/main/event_federation.py b/synapse/storage/databases/main/event_federation.py index cc7083b605..415926eb0a 100644 --- a/synapse/storage/databases/main/event_federation.py +++ b/synapse/storage/databases/main/event_federation.py @@ -1493,6 +1493,15 @@ class EventFederationWorkerStore( ) return frozenset(event_ids) + async def get_state_dag_extremities(self, room_id: str) -> frozenset[str]: + event_ids = await self.db_pool.simple_select_onecol( + table="msc4242_state_dag_forward_extremities", + keyvalues={"room_id": room_id}, + retcol="event_id", + desc="get_state_dag_extremities", + ) + return frozenset(event_ids) + async def get_min_depth(self, room_id: str) -> int | None: """For the given room, get the minimum depth we have seen for it.""" return await self.db_pool.runInteraction( diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 60fc884c3a..79c94fb29f 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -48,7 +48,9 @@ from synapse.api.errors import PartialStateConflictError from synapse.api.room_versions import RoomVersions from synapse.events import ( EventBase, + FrozenEventVMSC4242, StrippedStateEvent, + event_exists_in_state_dag, is_creator, relation_from_event, ) @@ -294,6 +296,7 @@ class PersistEventsStore: new_event_links: dict[str, NewEventChainLinks], use_negative_stream_ordering: bool = False, inhibit_local_membership_updates: bool = False, + new_state_dag_forward_extremities: set[str] | None = None, ) -> None: """Persist a set of events alongside updates to the current state and forward extremities tables. @@ -314,6 +317,8 @@ class PersistEventsStore: from being updated by these events. This should be set to True for backfilled events because backfilled events in the past do not affect the current local state. + new_state_dag_forward_extremities: A set of event IDs that are the new forward + extremities for the state DAG for this room. MSC4242 only. Returns: Resolves when the events have been persisted @@ -378,6 +383,7 @@ class PersistEventsStore: new_forward_extremities=new_forward_extremities, new_event_links=new_event_links, sliding_sync_table_changes=sliding_sync_table_changes, + new_state_dag_forward_extremities=new_state_dag_forward_extremities, ) persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc( len(events_and_contexts) @@ -961,8 +967,10 @@ class PersistEventsStore: return results - async def _get_prevs_before_rejected(self, event_ids: Iterable[str]) -> set[str]: - """Get soft-failed ancestors to remove from the extremities. + async def _get_prevs_before_rejected( + self, event_ids: Iterable[str], include_soft_failed: bool = True + ) -> set[str]: + """Get soft-failed/rejected ancestors to remove from the extremities. Given a set of events, find all those that have been soft-failed or rejected. Returns those soft failed/rejected events and their prev @@ -975,7 +983,8 @@ class PersistEventsStore: Args: event_ids: Events to find prev events for. Note that these must have already been persisted. - + include_soft_failed: Soft-failed events are included in the search. If false, only + rejected events are included. Returns: The previous events. """ @@ -1015,7 +1024,7 @@ class PersistEventsStore: continue soft_failed = db_to_json(metadata).get("soft_failed") - if soft_failed or rejected: + if (include_soft_failed and soft_failed) or rejected: to_recursively_check.append(prev_event_id) existing_prevs.add(prev_event_id) @@ -1037,6 +1046,7 @@ class PersistEventsStore: new_forward_extremities: set[str] | None, new_event_links: dict[str, NewEventChainLinks], sliding_sync_table_changes: SlidingSyncTableChanges | None, + new_state_dag_forward_extremities: set[str] | None = None, ) -> None: """Insert some number of room events into the necessary database tables. @@ -1145,6 +1155,11 @@ class PersistEventsStore: max_stream_order=max_stream_order, ) + if new_state_dag_forward_extremities: + self._set_state_dag_extremities_txn( + txn, room_id, new_state_dag_forward_extremities + ) + self._persist_transaction_ids_txn(txn, events_and_contexts) # Insert into event_to_state_groups. @@ -2469,6 +2484,29 @@ class PersistEventsStore: ], ) + def _set_state_dag_extremities_txn( + self, txn: LoggingTransaction, room_id: str, new_extrems: Collection[str] + ) -> None: + self.db_pool.simple_delete_txn( + txn, + table="msc4242_state_dag_forward_extremities", + keyvalues={ + "room_id": room_id, + }, + ) + self.db_pool.simple_insert_many_txn( + txn, + table="msc4242_state_dag_forward_extremities", + keys=("room_id", "event_id"), + values=[ + ( + room_id, + event_id, + ) + for event_id in new_extrems + ], + ) + @classmethod def _filter_events_and_contexts_for_duplicates( cls, events_and_contexts: list[EventPersistencePair] @@ -2848,6 +2886,12 @@ class PersistEventsStore: self._handle_event_relations(txn, event) + if event.room_version.msc4242_state_dags and event_exists_in_state_dag( + event + ): + assert isinstance(event, FrozenEventVMSC4242) + self._store_state_dag_edges(txn, event) + # Store the labels for this event. labels = event.content.get(EventContentFields.LABELS) if labels: @@ -2924,6 +2968,34 @@ class PersistEventsStore: txn.async_call_after(external_prefill) txn.call_after(local_prefill) + def _store_state_dag_edges( + self, txn: LoggingTransaction, event: FrozenEventVMSC4242 + ) -> None: + # the create event has no edge but we still need to persist it as get_state_dag just + # yanks all rows in this table. It's a bit gross to store NULL as the prev_state_event_id + # though. + if len(event.prev_state_events) == 0 and event.type == EventTypes.Create: + self.db_pool.simple_insert_txn( + txn, + table="msc4242_state_dag_edges", + values={ + "room_id": event.room_id, + "event_id": event.event_id, + "prev_state_event_id": None, + }, + ) + return + self.db_pool.simple_upsert_many_txn( + txn, + table="msc4242_state_dag_edges", + key_names=["room_id", "event_id", "prev_state_event_id"], + key_values=[ + (event.room_id, event.event_id, pae) for pae in event.prev_state_events + ], + value_names=(), + value_values=(), + ) + def _store_redaction(self, txn: LoggingTransaction, event: EventBase) -> None: assert event.redacts is not None self.db_pool.simple_upsert_txn( @@ -3444,7 +3516,13 @@ class PersistEventsStore: """ state_groups = {} for event, context in events_and_contexts: - if event.internal_metadata.is_outlier(): + # state dag rooms allow outliers to have state, as /gme'd state dag events are nominally + # outliers (not present in the timeline) but do need state persisted so we can calculate + # what the auth_events are for the event. + if ( + not event.room_version.msc4242_state_dags + and event.internal_metadata.is_outlier() + ): # double-check that we don't have any events that claim to be outliers # *and* have partial state (which is meaningless: we should have no # state at all for an outlier) diff --git a/synapse/storage/databases/main/state.py b/synapse/storage/databases/main/state.py index cfde107b48..87523e6f18 100644 --- a/synapse/storage/databases/main/state.py +++ b/synapse/storage/databases/main/state.py @@ -38,7 +38,7 @@ import attr from synapse.api.constants import EventContentFields, EventTypes, Membership from synapse.api.errors import NotFoundError, UnsupportedRoomVersionError from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion -from synapse.events import EventBase +from synapse.events import EventBase, EventMetadata from synapse.events.snapshot import EventContext from synapse.logging.opentracing import trace from synapse.replication.tcp.streams import UnPartialStatedEventStream @@ -78,16 +78,6 @@ class Sentinel: ROOM_UNKNOWN_SENTINEL = Sentinel() -@attr.s(slots=True, frozen=True, auto_attribs=True) -class EventMetadata: - """Returned by `get_metadata_for_events`""" - - room_id: str - event_type: str - state_key: str | None - rejection_reason: str | None - - def _retrieve_and_check_room_version(room_id: str, room_version_id: str) -> RoomVersion: v = KNOWN_ROOM_VERSIONS.get(room_version_id) if not v: diff --git a/synapse/storage/schema/__init__.py b/synapse/storage/schema/__init__.py index c4c4d7bcc4..43a7c84be9 100644 --- a/synapse/storage/schema/__init__.py +++ b/synapse/storage/schema/__init__.py @@ -171,6 +171,7 @@ Changes in SCHEMA_VERSION = 92 Changes in SCHEMA_VERSION = 93 - MSC4140: Set delayed events to be uniquely identifiable by their delay ID. + - MSC4242: Add state DAG tables. """ diff --git a/synapse/storage/schema/main/delta/93/04_state_dag_fwd_extrems.sql b/synapse/storage/schema/main/delta/93/04_state_dag_fwd_extrems.sql new file mode 100644 index 0000000000..51e90c960c --- /dev/null +++ b/synapse/storage/schema/main/delta/93/04_state_dag_fwd_extrems.sql @@ -0,0 +1,42 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations, Ltd +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + +CREATE TABLE IF NOT EXISTS msc4242_state_dag_forward_extremities( + -- we always expect the room to exist. If it gets removed, delete fwd extremities. + room_id TEXT NOT NULL REFERENCES rooms(room_id) ON DELETE CASCADE, + event_id TEXT NOT NULL REFERENCES events(event_id) ON DELETE CASCADE, + -- it doesn't make sense to reference the same event multiple times, and this uniqueness + -- index is also used to delete events once they are no longer forward extremities. + UNIQUE (event_id, room_id) +); +-- When creating events, we want to select all forward extremities for a room which this index helps with. +CREATE INDEX msc4242_state_dag_room ON msc4242_state_dag_forward_extremities(room_id); + + +CREATE TABLE IF NOT EXISTS msc4242_state_dag_edges( + -- Deleting the room deletes the state DAG. + room_id TEXT NOT NULL REFERENCES rooms(room_id) ON DELETE CASCADE, + -- the event IDs being referenced must exist (hence REFERENCES) and we do not want to accidentally delete + -- the event and create a hole in the state DAG. It is not possible for a state + -- DAG room to function with an holey DAG, so these events _cannot_ be purged. To purge them, the + -- entire room would need to be deleted. + event_id TEXT NOT NULL REFERENCES events(event_id), + -- one of the prev state events for this event ID. We must have it since we must have the entire state DAG. + -- can be NULL for the create event. + prev_state_event_id TEXT REFERENCES events(event_id) + -- calculated depth for this event ID. There is some denormalisation here as we're storing the depth + -- multiple times for each prev_state_event_id. + -- depth BIGINT NOT NULL +); +CREATE INDEX msc4242_state_dag_edges_by_room ON msc4242_state_dag_edges(room_id); +CREATE UNIQUE INDEX msc4242_state_dag_edges_key ON msc4242_state_dag_edges(room_id, event_id, prev_state_event_id); diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi index 0add391c65..ebe9184b07 100644 --- a/synapse/synapse_rust/events.pyi +++ b/synapse/synapse_rust/events.pyi @@ -43,6 +43,9 @@ class EventInternalMetadata: device_id: str """The device ID of the user who sent this event, if any.""" + # MSC4242 state dags + calculated_auth_event_ids: list[str] + def get_dict(self) -> JsonDict: ... def is_outlier(self) -> bool: ... def copy(self) -> "EventInternalMetadata": ... diff --git a/tests/federation/test_federation_server.py b/tests/federation/test_federation_server.py index c4491d5b3c..5cd4e154cf 100644 --- a/tests/federation/test_federation_server.py +++ b/tests/federation/test_federation_server.py @@ -438,12 +438,18 @@ class SendJoinFederationTests(unittest.FederatingHomeserverTestCase): @override_config({"use_frozen_dicts": True}) def test_send_join_with_frozen_dicts(self, room_version: str) -> None: """Test send_join with USE_FROZEN_DICTS=True""" + if room_version == RoomVersions.MSC4242v12.identifier: + # TODO: This room version doesn't work over federation in this PR. + return self._test_send_join_common(room_version) @parameterized.expand([(k,) for k in KNOWN_ROOM_VERSIONS.keys()]) @override_config({"use_frozen_dicts": False}) def test_send_join_without_frozen_dicts(self, room_version: str) -> None: """Test send_join with USE_FROZEN_DICTS=False""" + if room_version == RoomVersions.MSC4242v12.identifier: + # TODO: This room version doesn't work over federation in this PR. + return self._test_send_join_common(room_version) def test_send_join_partial_state(self) -> None: diff --git a/tests/handlers/test_federation_event.py b/tests/handlers/test_federation_event.py index 3d856b9346..1aaa86e2e8 100644 --- a/tests/handlers/test_federation_event.py +++ b/tests/handlers/test_federation_event.py @@ -1121,7 +1121,9 @@ class FederationEventHandlerTests(unittest.FederatingHomeserverTestCase): return {"pdus": [missing_event.get_pdu_json()]} async def get_room_state_ids( - destination: str, room_id: str, event_id: str + destination: str, + room_id: str, + event_id: str, ) -> JsonDict: self.assertEqual(destination, self.OTHER_SERVER_NAME) self.assertEqual(event_id, missing_event.event_id) @@ -1131,7 +1133,10 @@ class FederationEventHandlerTests(unittest.FederatingHomeserverTestCase): } async def get_room_state( - room_version: RoomVersion, destination: str, room_id: str, event_id: str + room_version: RoomVersion, + destination: str, + room_id: str, + event_id: str, ) -> StateRequestResponse: self.assertEqual(destination, self.OTHER_SERVER_NAME) self.assertEqual(event_id, missing_event.event_id) diff --git a/tests/storage/test_msc4242_state_dag.py b/tests/storage/test_msc4242_state_dag.py new file mode 100644 index 0000000000..c746bf4e10 --- /dev/null +++ b/tests/storage/test_msc4242_state_dag.py @@ -0,0 +1,337 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations, Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . + +from typing import Iterable +from unittest.mock import Mock + +from twisted.test.proto_helpers import MemoryReactor + +from synapse.api.constants import EventTypes +from synapse.api.errors import SynapseError +from synapse.api.room_versions import RoomVersions +from synapse.events import FrozenEventVMSC4242, make_event_from_dict +from synapse.events.snapshot import EventContext +from synapse.rest.client import room +from synapse.server import HomeServer +from synapse.util.clock import Clock + +from tests.unittest import HomeserverTestCase + + +class MSC4242StateDagsTests(HomeserverTestCase): + user_id = "@red:server" + servlets = [room.register_servlets] + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + hs = self.setup_test_homeserver("server") + return hs + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.room_id = self.helper.create_room_as( + self.user_id, + room_version=RoomVersions.MSC4242v12.identifier, + ) + + self.store = hs.get_datastores().main + self._storage_controllers = self.hs.get_storage_controllers() + + def _get_prev_state_events(self, event_id: str) -> list[str]: + ev = self.helper.get_event(self.room_id, event_id) + prev_state_events: list[str] | None = ev.get("prev_state_events", None) + assert prev_state_events is not None + return prev_state_events + + def test_forward_extremities_are_calculated(self) -> None: + """ + Check that forward extremities are set as prev_state_events and that they don't change + for non-state events. + """ + # they don't change for messages + first_id = self.helper.send(self.room_id, body="test1")["event_id"] + first_pae = self._get_prev_state_events(first_id) + assert len(first_pae) == 1 + second_id = self.helper.send(self.room_id, body="test2")["event_id"] + second_pae = self._get_prev_state_events(second_id) + assert len(second_pae) == 1 + self.assertEquals(first_pae, second_pae) + + # send another auth event, which should change the prev_state_events on subsequent events + jr_id = self.helper.send_state( + self.room_id, + EventTypes.JoinRules, + { + "join_rule": "knock", + }, + tok="nope", + )["event_id"] + jr_pae = self._get_prev_state_events(jr_id) + self.assertEquals(second_pae, jr_pae) + + # prev_state_events should always point to the join rule now + third_id = self.helper.send(self.room_id, body="test3")["event_id"] + third_pae = self._get_prev_state_events(third_id) + self.assertEquals(third_pae, [jr_id]) + # ..including for non-auth state + # TODO FIXME KEGAN + # name_id = self.helper.send_state( + # self.room_id, + # EventTypes.Name, + # { + # "name": "State DAGs!", + # }, + # tok="nope", + # )["event_id"] + # name_pae = self._get_prev_state_events(name_id) + # self.assertEquals(name_pae, [jr_id]) + + +class MSC4242EventPersistenceAuthDagsStoreTestCase(HomeserverTestCase): + servlets = [ + room.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + persistence = hs.get_storage_controllers().persistence + assert persistence is not None + self.persistence = persistence + self.room_id = "!foo:bar" + self.seen_event_ids: set[str] = set() + self.persistence.main_store = Mock(spec=["have_seen_events"]) + self.persistence.main_store.have_seen_events.side_effect = ( + self._have_seen_events + ) + self.rejected_event_ids_and_their_prevs: set[str] = set() + self.persistence.persist_events_store = Mock( + spec=["_get_prevs_before_rejected"] + ) + self.persistence.persist_events_store._get_prevs_before_rejected.side_effect = ( + self._get_prevs_before_rejected + ) + + async def _have_seen_events( + self, room_id: str, event_ids: Iterable[str] + ) -> set[str]: + unknown_events = set(event_ids) + return self.seen_event_ids.intersection(unknown_events) + + async def _get_prevs_before_rejected( + self, event_ids: Iterable[str], include_soft_failed: bool = True + ) -> set[str]: + return self.rejected_event_ids_and_their_prevs + + def _make_event( + self, + id: str, + prev_state_events: list[str], + rejected: bool = False, + ) -> tuple[FrozenEventVMSC4242, EventContext]: + ev = make_event_from_dict( + { + "prev_state_events": prev_state_events, + "content": { + "membership": "join", + }, + "sender": "@unimportant:info", + "state_key": "@unimportant:info", + "type": "m.room.member", + "room_id": self.room_id, + }, + room_version=RoomVersions.MSC4242v12, + ) + assert isinstance(ev, FrozenEventVMSC4242) + ev._event_id = id + ctx = Mock() + ctx.rejected = rejected + return ev, ctx + + def _test( + self, + current_fwds: list[str], + new_events: list[tuple[FrozenEventVMSC4242, EventContext]], + want_new_extrems: set[str], + want_raises: bool = False, + ) -> None: + promise = self.persistence._calculate_new_state_dag_extremities( + self.room_id, + frozenset(current_fwds), + new_events, + ) + if want_raises: + f = self.get_failure(promise, SynapseError) + assert f is not None + return + + new_extrems = set(self.get_success(promise)) + self.assertEqual( + new_extrems, + want_new_extrems, + f"want_new_extrems={want_new_extrems} got={new_extrems}", + ) + + def test_calculate_new_state_dag_extremities_simple(self) -> None: + # Simple linear chain + self._test( + current_fwds=[], + new_events=[ + self._make_event("$1", []), + self._make_event("$2", ["$1"]), + self._make_event("$3", ["$2"]), + self._make_event("$4", ["$3"]), + ], + want_new_extrems={"$4"}, + ) + + def test_calculate_new_state_dag_extremities_fork(self) -> None: + # Simple fork so we end up with two forward extrems + self._test( + current_fwds=[], + new_events=[ + self._make_event("$1", []), + self._make_event("$2", ["$1"]), + self._make_event("$3", ["$2"]), + self._make_event("$4", ["$2"]), + ], + want_new_extrems={"$3", "$4"}, + ) + + def test_calculate_new_state_dag_extremities_merge(self) -> None: + # Simple fork so we end up with two forward extrems + self._test( + current_fwds=[], + new_events=[ + self._make_event("$1", []), + self._make_event("$2", ["$1"]), + self._make_event("$3", ["$1"]), + self._make_event("$4", ["$2", "$3"]), + ], + want_new_extrems={"$4"}, + ) + + def test_calculate_new_state_dag_extremities_fork_on_existing(self) -> None: + # Fork where we are adding to older events + self.seen_event_ids = {"$1", "$2", "$3"} + self._test( + current_fwds=["$3"], + new_events=[ + self._make_event("$4", ["$3"]), # append to the forward extrem + self._make_event("$5", ["$1"]), # append to the root + ], + want_new_extrems={"$4", "$5"}, + ) + + def test_calculate_new_state_dag_extremities_merge_on_existing(self) -> None: + # Merge where we are merging to older events + self.seen_event_ids = {"$1", "$2", "$3"} + self._test( + current_fwds=["$3"], + new_events=[ + self._make_event("$4", ["$3", "$2"]), + ], + want_new_extrems={"$4"}, + ) + + def test_calculate_new_state_dag_extremities_merge_on_not_current(self) -> None: + # Merge where we are merging to older events + self.seen_event_ids = {"$1", "$2", "$3"} + self._test( + current_fwds=["$3"], + new_events=[ + self._make_event("$4", ["$1", "$2"]), + ], + want_new_extrems={"$3", "$4"}, + ) + + def test_calculate_new_state_dag_extremities_append_with_rejected(self) -> None: + # rejected events cannot be forward extremities + self.seen_event_ids = {"$1", "$2", "$3"} + self._test( + current_fwds=["$3"], + new_events=[ + self._make_event("$4", ["$3"], rejected=True), + ], + want_new_extrems={"$3"}, + ) + + self._test( + current_fwds=["$3"], + new_events=[ + self._make_event("$4", ["$3"], rejected=True), + self._make_event("$5", ["$4"], rejected=True), + ], + want_new_extrems={"$3"}, + ) + + def test_calculate_new_state_dag_extremities_append_with_rejected_in_chain( + self, + ) -> None: + # rejected events cannot be forward extremities, but events that come after them can. + # this shouldn't cause multiple forward extremities. + self.seen_event_ids = {"$1", "$2", "$3"} + self.rejected_event_ids_and_their_prevs = {"$4", "$3"} + self._test( + current_fwds=["$3"], + new_events=[ + self._make_event("$4", ["$3"], rejected=True), + self._make_event("$5", ["$4"]), + ], + want_new_extrems={"$5"}, + ) + + def test_calculate_new_state_dag_extremities_missing_prevs_raises(self) -> None: + self._test( + current_fwds=[], + new_events=[ + self._make_event("$1", []), + self._make_event("$2", ["$1"]), + self._make_event("$3", ["$unknown"]), + self._make_event("$4", ["$3"]), + ], + want_new_extrems={"$4"}, + want_raises=True, + ) + + def test_calculate_new_state_dag_extremities_complex(self) -> None: + """ + 1 + | \ + 2 4 + | + 3 + + Exists already, then becomes... + + 1______ + | \\ | + 2 4 5R + | | | + 3--7 6R + | \\ / \ + 10R 8 9 + + """ + # Merge where we are merging to older events + self.seen_event_ids = {"$1", "$2", "$3", "$4"} + self.rejected_event_ids_and_their_prevs = {"$1", "$5", "$6", "$3", "$10"} + self._test( + current_fwds=["$3", "$4"], + new_events=[ + self._make_event("$5", ["$1"], rejected=True), + self._make_event("$6", ["$5"], rejected=True), + self._make_event("$7", ["$4", "$3"]), + self._make_event("$8", ["$6", "$7"]), + self._make_event("$9", ["$6"]), + self._make_event("$10", ["$3"], rejected=True), + ], + want_new_extrems={"$8", "$9"}, + ) diff --git a/tests/storage/test_redaction.py b/tests/storage/test_redaction.py index 92eb99f1d5..ca78ad85ae 100644 --- a/tests/storage/test_redaction.py +++ b/tests/storage/test_redaction.py @@ -250,6 +250,7 @@ class RedactionTestCase(unittest.HomeserverTestCase): prev_event_ids: list[str], auth_event_ids: list[str] | None, depth: int | None = None, + prev_state_events: list[str] | None = None, ) -> EventBase: built_event = await self._base_builder.build( prev_event_ids=prev_event_ids, auth_event_ids=auth_event_ids diff --git a/tests/test_event_auth.py b/tests/test_event_auth.py index 934a2fd307..340b270f43 100644 --- a/tests/test_event_auth.py +++ b/tests/test_event_auth.py @@ -20,15 +20,16 @@ # import unittest +from collections import namedtuple from typing import Any, Collection, Iterable from parameterized import parameterized from synapse import event_auth -from synapse.api.constants import EventContentFields +from synapse.api.constants import EventContentFields, RejectedReason from synapse.api.errors import AuthError, SynapseError from synapse.api.room_versions import EventFormatVersions, RoomVersion, RoomVersions -from synapse.events import EventBase, make_event_from_dict +from synapse.events import EventBase, event_exists_in_state_dag, make_event_from_dict from synapse.storage.databases.main.events_worker import EventRedactBehaviour from synapse.types import JsonDict, get_domain_from_id @@ -374,6 +375,189 @@ class EventAuthTestCase(unittest.TestCase): auth_events, ) + def test_msc4242_state_dag_rules(self) -> None: + """Tests additional rules in place for state DAG rooms. + + 1. m.room.create => if it has any prev_state_events, reject. + 2. Considering the event's prev_state_events: + i. If there are entries which do not belong in the same room, reject. + ii. If there are entries which do not have a state_key, reject. + iii. If there are entries which were themselves rejected under the checks performed on receipt of a PDU, reject. + """ + creator = "@creator:example.com" + room_version = RoomVersions.MSC4242v12 + + create_event = make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.create", + "sender": creator, + "state_key": "", + "content": {"creator": creator}, + "prev_events": [], + "prev_state_events": [], + }, + room_version, + ) + join_event = make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.member", + "sender": creator, + "state_key": creator, + "content": {"membership": "join"}, + "prev_events": [create_event.event_id], + "prev_state_events": [create_event.event_id], + }, + room_version, + {"calculated_auth_event_ids": [create_event.event_id]}, + ) + event_in_another_room = make_event_from_dict( + { + "room_id": f"!another_room:{TEST_DOMAIN}", + "type": "m.room.join_rules", + "sender": creator, + "state_key": "", + "content": {"join_rule": "public"}, + "prev_events": [join_event.event_id], + "prev_state_events": [join_event.event_id], + }, + room_version, + {"calculated_auth_event_ids": [create_event.event_id, join_event.event_id]}, + ) + msg_event = make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.message", + "sender": creator, + "content": {"msgtype": "m.text", "body": "I am a message"}, + "prev_events": [join_event.event_id], + "prev_state_events": [join_event.event_id], + }, + room_version, + {"calculated_auth_event_ids": [create_event.event_id, join_event.event_id]}, + ) + rejected_event = make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.name", + "sender": creator, + "state_key": "", + "content": {"name": "REJECTED"}, + "prev_events": [join_event.event_id], + "prev_state_events": [join_event.event_id], + }, + room_version, + {"calculated_auth_event_ids": [create_event.event_id, join_event.event_id]}, + rejected_reason=RejectedReason.AUTH_ERROR, + ) + RejectingTestCase = namedtuple( + "RejectingTestCase", "name events_in_store test_event" + ) + rejecting_test_cases = [ + RejectingTestCase( + name="create event has prev_state_events", + events_in_store=[create_event, join_event], + test_event=make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.create", + "sender": creator, + "state_key": "", + "content": {"creator": creator}, + "prev_events": [], + "prev_state_events": [join_event.event_id], + }, + room_version, + { + "calculated_auth_event_ids": [ + create_event.event_id, + join_event.event_id, + ] + }, + ), + ), + RejectingTestCase( + name="prev_state_event belongs in a different room", + events_in_store=[create_event, join_event, event_in_another_room], + test_event=make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.name", + "sender": creator, + "state_key": "", + "content": {"name": "prev state event is in another room"}, + "prev_events": [join_event.event_id], + "prev_state_events": [event_in_another_room.event_id], + }, + room_version, + { + "calculated_auth_event_ids": [ + create_event.event_id, + join_event.event_id, + ] + }, + ), + ), + RejectingTestCase( + name="prev_state_event is a message event", + events_in_store=[create_event, join_event, msg_event], + test_event=make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.name", + "sender": creator, + "state_key": "", + "content": {"name": "prev state event is a message"}, + "prev_events": [msg_event.event_id], + "prev_state_events": [msg_event.event_id], + }, + room_version, + { + "calculated_auth_event_ids": [ + create_event.event_id, + join_event.event_id, + ] + }, + ), + ), + RejectingTestCase( + name="prev_state_event was rejected", + events_in_store=[create_event, join_event, rejected_event], + test_event=make_event_from_dict( + { + "room_id": TEST_ROOM_ID, + "type": "m.room.name", + "sender": creator, + "state_key": "", + "content": {"name": "prev state event was rejected"}, + "prev_events": [join_event.event_id], + "prev_state_events": [rejected_event.event_id], + }, + room_version, + { + "calculated_auth_event_ids": [ + create_event.event_id, + join_event.event_id, + ] + }, + ), + ), + ] + + for test_case in rejecting_test_cases: + event_store = _StubEventSourceStore() + event_store.add_events(test_case.events_in_store) + + with self.assertRaises( + AuthError, msg=f"test case {test_case.name} was not rejected" + ): + get_awaitable_result( + event_auth.check_state_independent_auth_rules( + event_store, test_case.test_event + ) + ) + @parameterized.expand([(RoomVersions.V1, True), (RoomVersions.V6, False)]) def test_notifications( self, room_version: RoomVersion, allow_modification: bool @@ -769,6 +953,105 @@ class EventAuthTestCase(unittest.TestCase): with self.assertRaises(SynapseError): event_auth._check_power_levels(event.room_version, event, {}) + def test_event_exists_in_state_dag(self) -> None: + events_that_exist_in_state_dag = [ + { + "type": "m.room.create", + "state_key": "", + "content": {}, + }, + { + "type": "m.room.join_rules", + "state_key": "", + "content": {}, + }, + { + "type": "m.room.power_levels", + "state_key": "", + "content": {}, + }, + { + "type": "m.room.server_acl", + "state_key": "", + "content": {}, + }, + { + "type": "m.room.member", + "state_key": "@alice:somewhere", + "content": {}, + }, + { + "type": "m.room.third_party_invite", + "state_key": "flibble", + "content": {}, + }, + { + "type": "m.room.create", + "state_key": " ", + "content": {}, + }, + { + "type": "m.room.join_rules", + "state_key": " ", + "content": {}, + }, + { + "type": "m.room.power_levels", + "state_key": " ", + "content": {}, + }, + { + "type": "m.room.name", + "state_key": "", + "content": {}, + }, + { + "type": "m.room.member", + "state_key": "", + "content": {}, + }, + { + "type": "m.room.member", + "state_key": "hello_world", + "content": {}, + }, + ] + events_that_dont_exist_in_state_dag = [ + { + "type": "m.room.message", + "content": {}, + }, + { + "type": "m.room.create", + "content": {}, + }, + { + "type": "m.room.join_rules", + "content": {}, + }, + { + "type": "m.room.power_levels", + "content": {}, + }, + ] + + def check_events(events: list[dict], should_exist: bool) -> None: + for ev in events: + base = { + "room_id": TEST_ROOM_ID, + "sender": "@test:test.com", + "signatures": {"test.com": {"ed25519:0": "some9signature"}}, + } + base.update(ev) + event = make_event_from_dict(base, RoomVersions.V10) + got = event_exists_in_state_dag(event) + self.assertEquals( + got, should_exist, f"{ev} should_exist={should_exist} but got {got}" + ) + + check_events(events_that_exist_in_state_dag, should_exist=True) + check_events(events_that_dont_exist_in_state_dag, should_exist=False) + # helpers for making events TEST_DOMAIN = "example.com"