diff --git a/rust/src/persist/chain_cover.rs b/rust/src/persist/chain_cover.rs index fa6e17902f..f6d6de45b0 100644 --- a/rust/src/persist/chain_cover.rs +++ b/rust/src/persist/chain_cover.rs @@ -53,7 +53,7 @@ use crate::{ pool::ConnectionPool, write::{run_rolled_back_interaction, run_write_interaction, WriteTxn}, }, - events::Event, + events::{constants::event_type::M_ROOM_CREATE, Event}, }; /// The sequence `event_chain_id_gen` allocates from. @@ -273,8 +273,12 @@ pub async fn calculate_chain_cover_index_for_events_txn( state_events: &[&ChainCoverEvent], ) -> anyhow::Result, NewEventChainLinks>> { // We ignore legacy rooms that we aren't filling the chain cover index - // for. - if !store.has_auth_chain_index(room_id).await? { + // for. A room whose create event is among the events is new, or being + // joined, and has the index, though its `rooms` row may not be stored yet. + let has_create_event = state_events + .iter() + .any(|event| &*event.event_type == M_ROOM_CREATE && event.state_key.is_empty()); + if !has_create_event && !store.has_auth_chain_index(room_id).await? { return Ok(HashMap::new()); } diff --git a/rust/src/persist/context.rs b/rust/src/persist/context.rs index 8d758f1456..756cefcbce 100644 --- a/rust/src/persist/context.rs +++ b/rust/src/persist/context.rs @@ -53,6 +53,10 @@ pub struct PersistContext { /// the event's own columns, when the push rules were evaluated on this /// worker. `None` when they are in the staging table, or there are none. pub push_actions: Option>, + /// Set on the create event when the chunk creates the room: the + /// transaction inserts the `rooms` row, with this visibility, and the + /// chunk fails with [`super::txn::RoomIdCollision`] if one is there. + pub new_room_is_public: Option, } /// `(user_id, actions, notif, highlight, unread, thread_id)`: Python's @@ -94,6 +98,9 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PersistContext { Ok(PersistContext { push_actions, + new_room_is_public: context + .getattr(intern!(py, "new_room_is_public"))? + .extract()?, rejected: context.getattr(intern!(py, "rejected"))?.extract()?, state_group: context.getattr(intern!(py, "_state_group"))?.extract()?, state_group_before_event: context diff --git a/rust/src/persist/python.rs b/rust/src/persist/python.rs index b37594da47..fb97f4a98d 100644 --- a/rust/src/persist/python.rs +++ b/rust/src/persist/python.rs @@ -36,7 +36,9 @@ use super::{ reader::{self, PersistReader, PyDelta, PySlidingSync}, sliding_sync, state_delta::{self, NativeStateResolver, StateResFallback}, - txn::{self, json, streams::TxnStream, PartialStateConflict, TxnEvent, TxnInput}, + txn::{ + self, json, streams::TxnStream, PartialStateConflict, RoomIdCollision, TxnEvent, TxnInput, + }, }; use crate::{ database::postgres::{ @@ -1517,6 +1519,11 @@ pub fn persist_error_to_py(e: anyhow::Error) -> PyErr { Ok(err) | Err(err) => err, }; } + if let Some(collision) = e.downcast_ref::() { + return match py_room_id_collision_error(&collision.0) { + Ok(err) | Err(err) => err, + }; + } if let Some(missing) = e.downcast_ref::() { return SynapseError::new( StatusCode::INTERNAL_SERVER_ERROR, @@ -1540,6 +1547,17 @@ fn py_partial_state_conflict_error() -> PyResult { }) } +/// A `RoomIdCollisionError`, as the Python transaction raises. +fn py_room_id_collision_error(room_id: &str) -> PyResult { + crate::gil::with_gil("persist_room_id_collision", |py| { + let error = py + .import("synapse.api.errors")? + .getattr("RoomIdCollisionError")? + .call1((room_id,))?; + Ok(PyErr::from_value(error)) + }) +} + /// `(internal_metadata, json)`: the two `event_json` columns as Rust would /// write them for `event`, for tests to compare with Python's. #[pyfunction] diff --git a/rust/src/persist/txn/mod.rs b/rust/src/persist/txn/mod.rs index be649e91b7..8c020dcabe 100644 --- a/rust/src/persist/txn/mod.rs +++ b/rust/src/persist/txn/mod.rs @@ -177,6 +177,19 @@ impl std::fmt::Display for PartialStateConflict { impl std::error::Error for PartialStateConflict {} +/// The chunk creates a room whose ID is already taken. Python raises +/// `RoomIdCollisionError`, and the creator tries again under a fresh ID. +#[derive(Debug)] +pub struct RoomIdCollision(pub String); + +impl std::fmt::Display for RoomIdCollision { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Room ID {} is already in use", self.0) + } +} + +impl std::error::Error for RoomIdCollision {} + /// An error reading an event's fields through the Python-facing accessors. fn metadata_error(e: PyErr) -> WriteError { WriteError::Other(anyhow::Error::new(e)) @@ -239,16 +252,46 @@ async fn persist_events_txn_steps<'a>( ) -> Result<(), WriteError> { let room_id = input.room_id; - // Protect against a race with deleting the room: the share lock holds - // off a delete until we commit. - let rows = txn - .query( - "SELECT room_version FROM rooms WHERE room_id = $1 FOR SHARE", - &[&room_id], - ) - .await?; - if rows.is_empty() { - return Err(anyhow!("Room does not exist {room_id}").into()); + let new_room = input + .events + .iter() + .find_map(|event| Some((event, event.context.new_room_is_public?))); + if let Some((create, is_public)) = new_room { + // The chunk creates the room: store it, which claims the room ID. + // The row's lock holds off a delete until we commit, as the share + // lock below does. + let inserted = txn + .execute( + "INSERT INTO rooms (room_id, room_version, is_public, creator, has_auth_chain_index) + VALUES ($1, $2, $3, $4, TRUE)", + &[ + &room_id, + &create.event.room_version().identifier, + &is_public, + &create.event.sender(), + ], + ) + .await; + match inserted { + Err(WriteError::Constraint { sqlstate, .. }) if sqlstate == "23505" => { + return Err(WriteError::Other(anyhow::Error::new(RoomIdCollision( + room_id.to_owned(), + )))); + } + other => other?, + }; + } else { + // Protect against a race with deleting the room: the share lock holds + // off a delete until we commit. + let rows = txn + .query( + "SELECT room_version FROM rooms WHERE room_id = $1 FOR SHARE", + &[&room_id], + ) + .await?; + if rows.is_empty() { + return Err(anyhow!("Room does not exist {room_id}").into()); + } } let (Some(first), Some(last)) = (input.events.first(), input.events.last()) else { diff --git a/synapse/api/errors.py b/synapse/api/errors.py index bce77ffba1..cdf363290f 100644 --- a/synapse/api/errors.py +++ b/synapse/api/errors.py @@ -574,6 +574,16 @@ class StoreError(SynapseError): """An error raised when there was a problem storing some data.""" +class RoomIdCollisionError(StoreError): + """A new room's ID is already taken, found when its events are persisted. + + The creator may try again under a fresh ID. + """ + + def __init__(self, room_id: str): + super().__init__(500, f"Room ID {room_id} is already in use") + + class InvalidCaptchaError(SynapseError): def __init__( self, diff --git a/synapse/events/snapshot.py b/synapse/events/snapshot.py index ebe7eb285a..b25bea11e5 100644 --- a/synapse/events/snapshot.py +++ b/synapse/events/snapshot.py @@ -148,6 +148,11 @@ class EventContext(UnpersistedEventContextBase): with the event; before any other persister, or another worker, gets the event, ``stage_push_actions`` moves them to the staging table and clears this. Not serialized. + + new_room_is_public: Set on the create event's context when the batch + creates the room. The persister inserts the room's ``rooms`` row, + with this visibility, rather than expecting one; a row already + there fails the batch with ``RoomIdCollisionError``. """ _storage: "StorageControllers" @@ -161,6 +166,7 @@ class EventContext(UnpersistedEventContextBase): partial_state: bool = False _state_map_before_event: StateMap[str] | None = None push_actions: list[PushActionRow] | None = None + new_room_is_public: bool | None = None @staticmethod def with_state( @@ -211,6 +217,7 @@ class EventContext(UnpersistedEventContextBase): ), "app_service_id": self.app_service.id if self.app_service else None, "partial_state": self.partial_state, + "new_room_is_public": self.new_room_is_public, } @staticmethod @@ -238,6 +245,7 @@ class EventContext(UnpersistedEventContextBase): ), rejected=input["rejected"], partial_state=input.get("partial_state", False), + new_room_is_public=input.get("new_room_is_public"), ) app_service_id = input["app_service_id"] diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 2967310522..1c01597b2a 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -62,6 +62,7 @@ from synapse.api.errors import ( LimitExceededError, NotFoundError, PartialStateConflictError, + RoomIdCollisionError, StoreError, SynapseError, ) @@ -297,13 +298,15 @@ class RoomCreationHandler: creation_event_with_context = await self._generate_create_event_for_room_id( requester, creation_content, - old_room_is_public, new_version, ) (create_event, _) = creation_event_with_context new_room_id = create_event.room_id else: new_room_id = self._generate_room_id() + # The room is stored with its first events, so building them must not + # read its version back. + self.store.get_room_version_id.prefill((new_room_id,), new_version.identifier) # Try several times, it could fail with PartialStateConflictError # in _upgrade_room, cf comment in except block. @@ -408,22 +411,12 @@ class RoomCreationHandler: logger.info("Creating new room %s to replace %s", new_room_id, old_room_id) - # We've already stored the room if we have the create event - if not creation_event_with_context: - # create the new room. may raise a `StoreError` in the exceedingly unlikely - # event of a room ID collision. - await self.store.store_room( - room_id=new_room_id, - room_creator_user_id=user_id, - is_public=old_room[0], - room_version=new_version, - ) - await self.clone_existing_room( requester, old_room_id=old_room_id, new_room_id=new_room_id, new_room_version=new_version, + is_public=old_room[0], tombstone_event_id=tombstone_event.event_id, additional_creators=additional_creators, creation_event_with_context=creation_event_with_context, @@ -597,6 +590,7 @@ class RoomCreationHandler: new_room_version: RoomVersion, tombstone_event_id: str, additional_creators: list[str] | None, + is_public: bool, creation_event_with_context: tuple[ EventBase, synapse.events.snapshot.UnpersistedEventContextBase ] @@ -767,6 +761,7 @@ class RoomCreationHandler: invite_list=[], initial_state=initial_state, creation_content=creation_content, + is_public=is_public, creation_event_with_context=creation_event_with_context, ) @@ -1081,6 +1076,38 @@ class RoomCreationHandler: ratelimit: bool = True, creator_join_profile: JsonDict | None = None, ignore_forced_encryption: bool = False, + ) -> tuple[str, RoomAlias | None, int]: + """Creates a new room. See `_create_room`. + + The room's ID is claimed when its first events are persisted, so a + collision (a v12+ room with the same create event, or a random ID + already in use) surfaces then, and the creation is run again under a + fresh ID. + """ + attempt = 0 + while True: + try: + return await self._create_room( + requester, + config, + ratelimit=ratelimit and attempt == 0, + creator_join_profile=creator_join_profile, + ignore_forced_encryption=ignore_forced_encryption, + attempt=attempt, + ) + except RoomIdCollisionError: + attempt += 1 + if attempt >= 5: + raise StoreError(500, "Couldn't generate a unique room ID.") + + async def _create_room( + self, + requester: Requester, + config: JsonDict, + ratelimit: bool, + creator_join_profile: JsonDict | None, + ignore_forced_encryption: bool, + attempt: int, ) -> tuple[str, RoomAlias | None, int]: """Creates a new room. @@ -1099,6 +1126,7 @@ class RoomCreationHandler: `avatar_url` and/or `displayname`. ignore_forced_encryption: Ignore encryption forced by `encryption_enabled_by_default_for_room_type` setting. + attempt: how many earlier attempts collided on their room ID. Returns: A 3-tuple containing: @@ -1106,9 +1134,11 @@ class RoomCreationHandler: - if requested, the room alias, otherwise None; and - the `stream_id` of the last persisted event. Raises: + RoomIdCollisionError: + if the room ID is already in use. SynapseError: - if the room ID couldn't be stored, 3pid invitation config - validation failed, or something went horribly wrong. + if 3pid invitation config validation failed, or something went + horribly wrong. ResourceLimitError: if server is blocked to some resource being exceeded @@ -1296,18 +1326,14 @@ class RoomCreationHandler: creation_event_with_context = await self._generate_create_event_for_room_id( requester, creation_content, - is_public, room_version, + attempt=attempt, ) (create_event, _) = creation_event_with_context room_id = create_event.room_id else: - room_id = await self._generate_and_create_room_id( - creator_id=user_id, - is_public=is_public, - room_version=room_version, - ) - # The room is in the database now, so building its events need not + room_id = self._generate_room_id() + # The room is stored with its first events, so building them must not # read its version back. self.store.get_room_version_id.prefill((room_id,), room_version.identifier) @@ -1328,7 +1354,7 @@ class RoomCreationHandler: user_id, room_id, room_aliases ): # allow room creation to continue but do not publish room - await self.store.set_room_is_public(room_id, False) + is_public = False directory_handler = self.hs.get_directory_handler() if room_alias: @@ -1388,26 +1414,33 @@ class RoomCreationHandler: else: remaining_invites = [] - ( - last_stream_id, - last_sent_event_id, - depth, - invites_sent, - ) = await self._send_events_for_new_room( - requester, - room_id, - room_version, - room_config=config, - invite_list=invite_list, - initial_state=initial_state, - creation_content=creation_content, - room_alias=room_alias, - power_level_content_override=power_level_content_override, - creator_join_profile=creator_join_profile, - ignore_forced_encryption=ignore_forced_encryption, - creation_event_with_context=creation_event_with_context, - batched_invites=batched_invites, - ) + try: + ( + last_stream_id, + last_sent_event_id, + depth, + invites_sent, + ) = await self._send_events_for_new_room( + requester, + room_id, + room_version, + room_config=config, + invite_list=invite_list, + initial_state=initial_state, + creation_content=creation_content, + is_public=is_public, + room_alias=room_alias, + power_level_content_override=power_level_content_override, + creator_join_profile=creator_join_profile, + ignore_forced_encryption=ignore_forced_encryption, + creation_event_with_context=creation_event_with_context, + batched_invites=batched_invites, + ) + except RoomIdCollisionError: + # The alias must not point at whoever owns the room ID. + if room_alias: + await self.store.delete_room_alias(room_alias) + raise # Invites the batch could not include, e.g. as they fail the auth # rules, are sent on their own and fail as before. remaining_invites = [ @@ -1476,80 +1509,50 @@ class RoomCreationHandler: self, creator: Requester, creation_content: JsonDict, - is_public: bool, room_version: RoomVersion, + attempt: int = 0, ) -> tuple[EventBase, synapse.events.snapshot.UnpersistedEventContextBase]: - """Create the create event for a v12+ room and store the room, retrying on room ID collision. - - In v12+ rooms the room ID is derived from the create event, so this - builds the create event, stores the room under the resulting ID, and - retries with a slightly older `origin_server_ts` if that ID is already taken. + """Create the create event for a v12+ room, whose room ID it defines. Args: creator: The user creating the room. creation_content: The content for the create event. - is_public: Whether the room is published to the room directory. room_version: The version of the room being created. + attempt: how many earlier create events collided on their room ID. Returns: A tuple of the create event and its unpersisted event context, for `_send_events_for_new_room` to persist with the rest of the room. - - Raises: - StoreError: if a unique room ID could not be generated after several - attempts. """ # In v12+ rooms, the room ID is the reference hash of the create event, # so two rooms whose create events have identical content collide on the # same room ID. This happens in practice when the same user creates # several rooms at once (e.g. concurrent /createRoom requests with the # same config), as the only entropy in the create event is - # `origin_server_ts`, which has millisecond resolution. - # - # Retry a few times on collision, perturbing `origin_server_ts` - # so the create event hashes to a fresh room ID. This mirrors the - # collision handling in `_generate_and_create_room_id` used for - # older room versions. - attempts = 0 - while attempts < 5: - event_dict: JsonDict = { - "content": creation_content, - "sender": creator.user.to_string(), - "type": EventTypes.Create, - "state_key": "", - } - if attempts > 0: - # Bump the timestamp to give the create event (and hence the - # room ID it hashes to) different content from the colliding one. - event_dict["origin_server_ts"] = ( - self.clock.time_msec() + random.randint(1, 10) - ) - - ( - creation_event, - creation_context, - ) = await self.event_creation_handler.create_event( - creator, - event_dict, - prev_event_ids=[], - depth=1, - state_map={}, - for_batch=True, + # `origin_server_ts`, which has millisecond resolution. The collision + # surfaces when the room's events are persisted, and `create_room` + # tries again. + event_dict: JsonDict = { + "content": creation_content, + "sender": creator.user.to_string(), + "type": EventTypes.Create, + "state_key": "", + } + if attempt > 0: + # Bump the timestamp to give the create event (and hence the + # room ID it hashes to) different content from the colliding one. + event_dict["origin_server_ts"] = self.clock.time_msec() + random.randint( + 1, 10 ) - try: - await self.store.store_room( - room_id=creation_event.room_id, - room_creator_user_id=creator.user.to_string(), - is_public=is_public, - room_version=room_version, - ) - except StoreError: - attempts += 1 - continue - return (creation_event, creation_context) - - raise StoreError(500, "Couldn't generate a unique room ID.") + return await self.event_creation_handler.create_event( + creator, + event_dict, + prev_event_ids=[], + depth=1, + state_map={}, + for_batch=True, + ) async def _send_events_for_new_room( self, @@ -1560,6 +1563,7 @@ class RoomCreationHandler: invite_list: list[str], initial_state: MutableStateMap, creation_content: JsonMapping, + is_public: bool, room_alias: RoomAlias | None = None, power_level_content_override: JsonDict | None = None, creator_join_profile: JsonDict | None = None, @@ -1716,6 +1720,13 @@ class RoomCreationHandler: ) events_to_send.clear() + if last_persisted_event is None: + # The first batch creates the room: its persistence stores + # the room, and so claims the room ID. + (create_event, create_context) = events_and_context[0] + assert create_event.type == EventTypes.Create + create_context.new_room_is_public = is_public + last_persisted_event = ( await self.event_creation_handler.handle_new_client_event( creator, @@ -2210,9 +2221,8 @@ class RoomCreationHandler: Room IDs look like "!opaque_id:domain" and are case-sensitive as per the spec at https://spec.matrix.org/v1.2/appendices/#room-ids-and-event-ids. - Does not check for collisions with existing rooms or prevent future calls from - returning the same room ID. To ensure the uniqueness of a new room ID, use - `_generate_and_create_room_id` instead. + Does not check for collisions with existing rooms: persisting the + room's first events claims the ID, and fails if it is taken. Synapse's room IDs are 18 [a-zA-Z] characters long, which comes out to around 102 bits. @@ -2223,29 +2233,6 @@ class RoomCreationHandler: random_string = stringutils.random_string(18) return RoomIdWithDomain(random_string, self.hs.hostname).to_string() - async def _generate_and_create_room_id( - self, - creator_id: str, - is_public: bool, - room_version: RoomVersion, - ) -> str: - # autogen room IDs and try to create it. We may clash, so just - # try a few times till one goes through, giving up eventually. - attempts = 0 - while attempts < 5: - try: - gen_room_id = self._generate_room_id() - await self.store.store_room( - room_id=gen_room_id, - room_creator_user_id=creator_id, - is_public=is_public, - room_version=room_version, - ) - return gen_room_id - except StoreError: - attempts += 1 - raise StoreError(500, "Couldn't generate a room ID.") - class RoomContextHandler: def __init__(self, hs: "HomeServer"): diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index c0948a5080..6cbcfc0dbe 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -46,7 +46,7 @@ from synapse.api.constants import ( ProfileUpdateAction, RelationTypes, ) -from synapse.api.errors import PartialStateConflictError +from synapse.api.errors import PartialStateConflictError, RoomIdCollisionError from synapse.api.room_versions import RoomVersions from synapse.events import ( EventBase, @@ -1010,16 +1010,23 @@ class PersistEventsStore: # See: docs/auth_chain_difference_algorithm.md # We ignore legacy rooms that we aren't filling the chain cover index - # for. - row = self.db_pool.simple_select_one_txn( - txn, - table="rooms", - keyvalues={"room_id": room_id}, - retcols=("room_id", "has_auth_chain_index"), - allow_none=True, + # for. A room whose create event is among the events is new, or being + # joined, and has the index, though its `rooms` row may not be stored + # yet. + has_create_event = any( + event.type == EventTypes.Create and event.state_key == "" + for event in state_events ) - if row is None or row[1] is False: - return {} + if not has_create_event: + row = self.db_pool.simple_select_one_txn( + txn, + table="rooms", + keyvalues={"room_id": room_id}, + retcols=("room_id", "has_auth_chain_index"), + allow_none=True, + ) + if row is None or row[1] is False: + return {} # Filter out events that we've already calculated. rows = self.db_pool.simple_select_many_txn( @@ -1332,6 +1339,30 @@ class PersistEventsStore: max_stream_order = events_and_contexts[-1][0].internal_metadata.stream_ordering room_version = events_and_contexts[0][0].room_version + new_room = [ + (event, context.new_room_is_public) + for event, context in events_and_contexts + if context.new_room_is_public is not None + ] + if new_room: + # The batch creates the room: store it, which also claims the + # room ID. The row then serves the existence check below. + (create_event, is_public) = new_room[0] + try: + self.db_pool.simple_insert_txn( + txn, + "rooms", + { + "room_id": room_id, + "creator": create_event.sender, + "is_public": is_public, + "room_version": room_version.identifier, + "has_auth_chain_index": True, + }, + ) + except self.database_engine.module.IntegrityError: + raise RoomIdCollisionError(room_id) + # We check that the room still exists for events we're trying to # persist. This is to protect against races with deleting a room. # diff --git a/tests/rest/client/bench_createroom.py b/tests/rest/client/bench_createroom.py index 91066f4a8d..fc49bffc5f 100644 --- a/tests/rest/client/bench_createroom.py +++ b/tests/rest/client/bench_createroom.py @@ -136,6 +136,9 @@ class CreateRoomBench(HomeserverTestCase): body = dict(body) if "room_alias_name" in body: body["room_alias_name"] = f"bench-{i}" + # Rooms created in the same fake millisecond get the same v12 room ID, + # which is retried, so give each room its own. + self.reactor.advance(0.001) channel = self.make_request("POST", "/createRoom", body, access_token=self.tok) self.assertEqual(channel.code, 200, channel.result) diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py index 8f8f60d546..db805d7250 100644 --- a/tests/rest/client/test_rooms.py +++ b/tests/rest/client/test_rooms.py @@ -795,7 +795,7 @@ class RoomsCreateTestCase(RoomBase): def _expected_create_room_txn_count(self) -> int: """The Python transactions a plain room creation runs. The state groups are written in Rust on the native driver, outside them.""" - count = 8 + count = 7 if self.hs.get_datastores().state._rust_state_group_writer is not None: count -= 1 return count