fix: Only insert event ID after checking/performing signatures

src/admin/user/commands.rs

@@ -514,7 +514,12 @@ pub(super) async fn force_join_list_of_local_users(
 				.services
 				.rooms
 				.membership
-				.join_room(&user_id, &room_id, Some(String::from(BULK_JOIN_REASON)), &servers)
+				.join_room(
+					&user_id,
+					&room_id,
+					Some(String::from(BULK_JOIN_REASON)),
+					servers.clone(),
+				)
 				.await
 			{
 				| Ok(_res) => {
@@ -596,7 +601,12 @@ pub(super) async fn force_join_all_local_users(
 				.services
 				.rooms
 				.membership
-				.join_room(user_id, &room_id, Some(String::from(BULK_JOIN_REASON)), &servers)
+				.join_room(
+					user_id,
+					&room_id,
+					Some(String::from(BULK_JOIN_REASON)),
+					servers.clone(),
+				)
 				.await
 			{
 				| Ok(_res) => {
@@ -643,7 +653,7 @@ pub(super) async fn force_join_room(
 		self.services
 			.rooms
 			.membership
-			.join_room(&user_id, &room_id, None, &servers)
+			.join_room(&user_id, &room_id, None, servers)
 			.await?;
 
 		self.write_str(&format!("{user_id} has been joined to {room_id}."))

src/api/client/membership/join.rs

@@ -75,7 +75,7 @@ pub(crate) async fn join_room_by_id_route(
 	let room_id = services
 		.rooms
 		.membership
-		.join_room(sender_user, &body.room_id, body.reason.clone(), &servers)
+		.join_room(sender_user, &body.room_id, body.reason.clone(), servers)
 		.boxed()
 		.await?;
 
@@ -195,7 +195,7 @@ pub(crate) async fn join_room_by_id_or_alias_route(
 	let room_id = services
 		.rooms
 		.membership
-		.join_room(sender_user, &room_id, body.reason.clone(), &servers)
+		.join_room(sender_user, &room_id, body.reason.clone(), servers)
 		.boxed()
 		.await?;
 

src/api/client/membership/knock.rs

@@ -141,7 +141,7 @@ pub(crate) async fn knock_room_route(
 		},
 	};
 
-	knock_room_by_id_helper(&services, sender_user, &room_id, body.reason.clone(), &servers)
+	knock_room_by_id_helper(&services, sender_user, &room_id, body.reason.clone(), servers)
 		.boxed()
 		.await
 }
@@ -151,7 +151,7 @@ async fn knock_room_by_id_helper(
 	sender_user: &UserId,
 	room_id: &RoomId,
 	reason: Option<String>,
-	servers: &[OwnedServerName],
+	servers: Vec<OwnedServerName>,
 ) -> Result<knock_room::v3::Response> {
 	let state_lock = services.rooms.state.mutex.lock(room_id).await;
 
@@ -242,7 +242,7 @@ async fn knock_room_by_id_helper(
 			match services
 				.rooms
 				.membership
-				.join_room(sender_user, room_id, reason.clone(), servers)
+				.join_room(sender_user, room_id, reason.clone(), servers.clone())
 				.await
 			{
 				| Ok(_) => return Ok(knock_room::v3::Response::new(room_id.to_owned())),
@@ -326,7 +326,7 @@ async fn knock_room_helper_local(
 	sender_user: &UserId,
 	room_id: &RoomId,
 	reason: Option<String>,
-	servers: &[OwnedServerName],
+	servers: Vec<OwnedServerName>,
 	state_lock: RoomMutexGuard,
 ) -> Result {
 	debug_info!("We can knock locally");
@@ -486,7 +486,7 @@ async fn knock_room_helper_remote(
 	sender_user: &UserId,
 	room_id: &RoomId,
 	reason: Option<String>,
-	servers: &[OwnedServerName],
+	servers: Vec<OwnedServerName>,
 	state_lock: RoomMutexGuard,
 ) -> Result {
 	info!("Knocking {room_id} over federation.");
@@ -694,7 +694,7 @@ async fn make_knock_request(
 	services: &Services,
 	sender_user: &UserId,
 	room_id: &RoomId,
-	servers: &[OwnedServerName],
+	servers: Vec<OwnedServerName>,
 ) -> Result<(federation::membership::prepare_knock_event::v1::Response, OwnedServerName)> {
 	let mut make_knock_response_and_server =
 		Err!(BadServerResponse("No server available to assist in knocking."));
@@ -702,7 +702,7 @@ async fn make_knock_request(
 	let mut make_knock_counter: usize = 0;
 
 	for remote_server in servers {
-		if services.globals.server_is_ours(remote_server) {
+		if services.globals.server_is_ours(&remote_server) {
 			continue;
 		}
 
@@ -716,7 +716,7 @@ async fn make_knock_request(
 
 		let make_knock_response = services
 			.sending
-			.send_federation_request(remote_server, request)
+			.send_federation_request(&remote_server, request)
 			.await;
 
 		trace!("make_knock response: {make_knock_response:?}");

src/api/server/invite.rs

@@ -307,7 +307,7 @@ async fn validate_invite_state(
 				"PDU in invite state (index {idx}) violates the room event format"
 			)));
 		};
-		let (state_event_room_id, state_event_id, state_event_json) = services
+		let (state_event_room_id, state_event_id, mut state_event_json) = services
 			.rooms
 			.event_handler
 			.parse_incoming_pdu(&raw_pdu, Some(room_version_rules))
@@ -351,6 +351,7 @@ async fn validate_invite_state(
 					entry.key(),
 				))),
 			| Entry::Vacant(entry) => {
+				state_event_json.insert("event_id".to_owned(), state_event_id.to_string().into());
 				if entry.key().0 == StateEventType::RoomCreate {
 					// Ensure this is a legal create event.
 					let pdu_event =

src/api/server/utils.rs

@@ -91,7 +91,7 @@ pub(crate) async fn validate_any_membership_event(
 	expected_room_id: OwnedRoomId,
 	expected_event_id: OwnedEventId,
 ) -> Result<(CanonicalJsonObject, MembershipState, OwnedUserId, OwnedUserId)> {
-	let (template_room_id, template_event_id, pdu) = services
+	let (template_room_id, template_event_id, mut pdu) = services
 		.rooms
 		.event_handler
 		.parse_incoming_pdu(body, Some(room_version_rules))
@@ -164,5 +164,6 @@ pub(crate) async fn validate_any_membership_event(
 		err!(Request(InvalidParam("Membership event violates the room event format: {e}")))
 	})?;
 
+	// pdu.insert("event_id".to_owned(), template_event_id.to_string().into());
 	Ok((pdu, membership.into(), sender_user, recipient_user))
 }

src/service/rooms/event_handler/pdu_checks.rs

@@ -66,6 +66,12 @@ pub async fn signature_hash_check_2_3(
 		pdu_json: CanonicalJsonObject,
 		room_version_rules: &RoomVersionRules,
 	) -> Result<CanonicalJsonObject> {
+		#[cfg(debug_assertions)]
+		{
+			if pdu_json.contains_key("event_id") {
+				conduwuit::warn!("Checking signature of a PDU that has an `event_id` present");
+			}
+		}
 		match self
 			.services
 			.server_keys
@@ -124,7 +130,7 @@ pub async fn auth_state_check_4(
 	/// Returns a boolean indicating whether the event is authorised, and also
 	/// the resolved state before the event for later use. Returns an error if
 	/// state fetching or auth checking fails.
-	pub(super) async fn state_before_check_5(
+	pub async fn state_before_check_5(
 		&self,
 		incoming_pdu: &PduEvent,
 		room_version_rules: &RoomVersionRules,

src/service/rooms/membership/event.rs

@@ -0,0 +1,84 @@
+use conduwuit::{Result, err, utils, utils::TryFutureExtExt};
+use ruma::{
+	CanonicalJsonObject, CanonicalJsonValue, OwnedUserId, RoomId, UserId,
+	canonical_json::{redact, to_canonical_value},
+	events::{
+		StateEventType,
+		room::member::{MembershipState, RoomMemberEventContent},
+	},
+	room_version_rules::RoomVersionRules,
+};
+
+impl super::Service {
+	/// Creates a membership event locally but seeds it with data from a remote
+	/// template. This means we retain total control over the created PDU, but
+	/// still use the data from the remote server that we have to blindly trust,
+	/// like auth/prev events and depth.
+	pub(super) async fn seed_local_membership_pdu(
+		&self,
+		room_id: &RoomId,
+		user_id: &UserId,
+		membership: MembershipState,
+		reason: Option<String>,
+		template: CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<CanonicalJsonObject> {
+		// First we redact the PDU to remove any unrecognised fields.
+		let mut template = redact(template, &room_version_rules.redaction, None)
+			.map_err(|e| err!(Request(BadJson("Failed to redact membership template: {e:?}"))))?;
+
+		// Remove fields that are not necessary but not covered by redaction.
+		template.remove("event_id");
+		template.remove("hashes");
+		template.remove("prev_state");
+		template.remove("origin");
+		template.remove("membership");
+		template.remove("redacts");
+
+		// Force set some fields to known good values.
+		let join_authorized_via_users_server = {
+			if room_version_rules
+				.signatures
+				.check_join_authorised_via_users_server
+			{
+				template
+					.get("content")
+					.map(|s| {
+						s.as_object()?
+							.get("join_authorised_via_users_server")?
+							.as_str()
+					})
+					.and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok())
+			} else {
+				None
+			}
+		};
+
+		let mut content = RoomMemberEventContent::new(membership);
+		let (dn, av) = tokio::join!(
+			self.services.users.displayname(user_id).ok(),
+			self.services.users.avatar_url(user_id).ok()
+		);
+		content.displayname = dn;
+		content.avatar_url = av;
+		content.reason = reason;
+		content
+			.join_authorized_via_users_server
+			.clone_from(&join_authorized_via_users_server);
+		template.insert("content".to_owned(), to_canonical_value(content)?);
+		template.insert(
+			"origin_server_ts".to_owned(),
+			CanonicalJsonValue::Integer(
+				utils::millis_since_unix_epoch()
+					.try_into()
+					.expect("Timestamp is valid js_int value"),
+			),
+		);
+		template.insert("room_id".to_owned(), room_id.as_str().into());
+		template.insert("sender".to_owned(), user_id.to_string().into());
+		template.insert("state_key".to_owned(), user_id.to_string().into());
+		template.insert("type".to_owned(), StateEventType::RoomMember.to_string().into());
+
+		Ok(template)
+	}
+}

src/service/rooms/membership/join.rs

@@ -0,0 +1,782 @@
+use std::{
+	borrow::Borrow,
+	collections::{BTreeMap, HashMap},
+	sync::Arc,
+	time::Instant,
+};
+
+use assign::assign;
+use conduwuit::{
+	Err, Event, PduEvent, Result, debug, debug_error, debug_info, err, error, info,
+	matrix::{StateKey, event::gen_event_id},
+	state_res,
+	state_res::EventTypeExt,
+	trace,
+	utils::{
+		IterStream,
+		stream::{BroadbandExt, WidebandExt, automatic_width},
+	},
+	warn,
+};
+use futures::{StreamExt, future::ready};
+use ruma::{
+	CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId, ServerName,
+	UserId,
+	api::{
+		error::{ErrorKind, IncompatibleRoomVersionErrorData},
+		federation::membership::{create_join_event, prepare_join_event},
+	},
+	events::{StateEventType, room::member::MembershipState},
+	room_version_rules::RoomVersionRules,
+};
+use serde_json::value::to_raw_value;
+
+use crate::rooms::{
+	event_handler::{DagBuilderTree, build_local_dag},
+	membership::validate_remote_member_event_stub,
+	state::RoomMutexGuard,
+	state_compressor::{CompressedState, HashSetCompressStateEvent},
+};
+
+type StateTypeKey = (StateEventType, StateKey);
+
+pub enum MakeJoinResult {
+	/// A make_join request returned and the response was validated
+	/// successfully.
+	Success((CanonicalJsonObject, RoomVersionId)),
+	/// A make_join response indicated further attempts to join should not be
+	/// made (e.g. Forbidden).
+	Fatal(conduwuit::Error),
+	/// A make_join request failed (e.g. remote server unreachable, or unable to
+	/// fulfill restricted join requirements), but another join attempt through
+	/// another remote should be attempted.
+	Retry,
+}
+
+fn state_key_from_json(obj: &CanonicalJsonObject) -> Option<(StateEventType, StateKey)> {
+	let event_type = StateEventType::from(obj.get("type")?.as_str()?);
+	let state_key = StateKey::from_str(obj.get("state_key")?.as_str()?);
+	Some((event_type, state_key))
+}
+
+impl super::Service {
+	/// Performs `POST /_matrix/federation/v1/make_join/{room_id}/{user_id}`.
+	///
+	/// If the request is successful, the resulting template is validated to
+	/// ensure the remote server isn't returning a bad event, and then the
+	/// template and room version are returned.
+	/// If validation fails, `Ok(None)` is returned, in
+	///
+	/// If the request is unsuccessful, either an `Err` or `Ok(None)` will be
+	/// returned, depending on the reason for the failure. If the join loop
+	/// should terminate (for example, a `M_FORBIDDEN` error was returned,
+	/// indicating we cannot join), then an `Err` will be returned. If the join
+	/// loop should continue because the request failed due to conditional
+	/// problems (e.g. server offline or unable to grant join), then `Ok(None)`
+	/// will be returned.
+	pub async fn make_join(
+		&self,
+		room_id: &RoomId,
+		user_id: &UserId,
+		via: &ServerName,
+	) -> MakeJoinResult {
+		assert!(
+			self.services.globals.user_is_local(user_id),
+			"can only make_join for local users"
+		);
+
+		let request = assign!(
+			prepare_join_event::v1::Request::new(room_id.to_owned(), user_id.to_owned()),
+			{ ver: self.services.server.supported_room_versions().collect() }
+		);
+
+		info!("Asking {via} for make_join");
+		let make_join_response = self
+			.services
+			.sending
+			.send_federation_request(via, request)
+			.await
+			.inspect(|_| info!("{via} finished make_join successfully"))
+			.inspect_err(|e| debug_info!("{via} failed to make_join: {e:?}"));
+
+		let make_join_response = match make_join_response {
+			| Ok(r) => r,
+			| Err(e) => return Self::handle_make_join_error(room_id, via, e),
+		};
+
+		// Make sure the remote server hasn't sent us garbage
+		let Ok(template) = serde_json::from_str(make_join_response.event.get())
+			.inspect_err(|e| warn!("{via} returned an invalid membership template: {e}"))
+		else {
+			return MakeJoinResult::Retry;
+		};
+		if let Err(e) =
+			validate_remote_member_event_stub(&MembershipState::Join, user_id, room_id, &template)
+		{
+			warn!("{via} returned an illegal membership template event: {e:?}");
+			return MakeJoinResult::Retry;
+		}
+
+		MakeJoinResult::Success((
+			template,
+			make_join_response.room_version.unwrap_or(RoomVersionId::V1),
+		))
+	}
+
+	/// Handles an error response from make_join, determining whether it should
+	/// be fatal (and returning `MakeJoinResult::Fatal(e)`), or
+	/// temporary/situational and returning `MakeJoinResult::Retry`.
+	fn handle_make_join_error(
+		room_id: &RoomId,
+		via: &ServerName,
+		e: conduwuit::Error,
+	) -> MakeJoinResult {
+		match e.kind() {
+			| ErrorKind::UnableToAuthorizeJoin => {
+				info!(
+					"{via} was unable to verify the joining user satisfied restricted join \
+					 requirements: {e}."
+				);
+				MakeJoinResult::Retry
+			},
+			| ErrorKind::UnableToGrantJoin => {
+				info!(
+					"{via} believes the joining user satisfies restricted join rules, but is \
+					 unable to authorise a join for us."
+				);
+				MakeJoinResult::Retry
+			},
+			| ErrorKind::IncompatibleRoomVersion(IncompatibleRoomVersionErrorData {
+				room_version,
+				..
+			}) => {
+				warn!(
+					"{via} reports the room we are trying to join is v{room_version}, which we \
+					 do not support."
+				);
+				MakeJoinResult::Fatal(e)
+			},
+			| ErrorKind::Forbidden => {
+				warn!("{via} refuses to let us join: {e}.");
+				MakeJoinResult::Fatal(e)
+			},
+			| ErrorKind::NotFound => {
+				info!("{via} does not know about {room_id}: {e}.");
+				MakeJoinResult::Retry
+			},
+			| ErrorKind::Unknown if e.status_code().is_server_error() => {
+				match e.status_code() {
+					| http::StatusCode::BAD_GATEWAY | http::StatusCode::SERVICE_UNAVAILABLE =>
+						info!("{via} is unavailable: {e}"),
+					| http::StatusCode::GATEWAY_TIMEOUT =>
+						info!("{via} timed out while handling make_join: {e}"),
+					| _ => info!("{via} encountered an internal server error: {e}."),
+				}
+				MakeJoinResult::Retry
+			},
+			| _ => {
+				info!("{via} unexpectedly failed to make_join: {e}.");
+				MakeJoinResult::Retry
+			},
+		}
+	}
+
+	pub async fn send_join(
+		&self,
+		room_id: OwnedRoomId,
+		event_id: OwnedEventId,
+		event: &CanonicalJsonObject,
+		via: &ServerName,
+		room_version_rules: &RoomVersionRules,
+		state_lock: &RoomMutexGuard,
+	) -> Result<Option<()>> {
+		let send_join_request = create_join_event::v2::Request::new(
+			room_id.clone(),
+			event_id.clone(),
+			self.services
+				.sending
+				.convert_to_outgoing_federation_event(event.clone())
+				.await,
+		);
+
+		info!("Asking {via} for send_join in room {room_id}");
+		let Ok(send_join_response) = self
+			.services
+			.sending
+			.send_slow_federation_request(via, send_join_request)
+			.await
+			.inspect(|resp| {
+				info!(
+					"{via} finished send_join successfully, returning {} state events and {} \
+					 authentication events.",
+					resp.room_state.state.len(),
+					resp.room_state.auth_chain.len(),
+				);
+			})
+			.inspect_err(|e| debug_error!("{via} failed to send_join: {e:?}"))
+		else {
+			return Ok(None); // Try another server
+		};
+		self.services
+			.short
+			.get_or_create_shortroomid(&room_id)
+			.await;
+
+		info!("Acquiring server signing keys for room state events");
+		self.services
+			.server_keys
+			.acquire_events_pubkeys(
+				send_join_response
+					.room_state
+					.state
+					.iter()
+					.chain(send_join_response.room_state.auth_chain.iter()),
+			)
+			.await;
+
+		info!("Building state map");
+		let (create_event, untrusted_state_before) = match self
+			.build_state_map(&send_join_response, &room_id, room_version_rules)
+			.await
+		{
+			| Ok(result) => result,
+			| Err(e) => {
+				error!("Failed to build state map: {e}");
+				// This usually happens when the remote server sends a malformed response (e.g.
+				// forgotten m.room.create event). In this case, we might succeed with another
+				// server, so we will return Ok(None) to encourage the caller to re-attempt.
+				return Ok(None);
+			},
+		};
+
+		info!(
+			"Handling returned room state authentication events ({} events)",
+			send_join_response.room_state.auth_chain.len()
+		);
+		let auth_map = self
+			.process_send_join_auth_chain(
+				&send_join_response,
+				&room_id,
+				room_version_rules,
+				&create_event,
+			)
+			.await?;
+
+		info!("Handling returned room state ({} events)", untrusted_state_before.len());
+		let untrusted_state_before_by_id = untrusted_state_before
+			.values()
+			.map(|value| {
+				(
+					gen_event_id(value, room_version_rules)
+						.expect("must be able to generate event ID"),
+					value,
+				)
+			})
+			.collect::<HashMap<_, _>>();
+		let state_before = self
+			.process_send_join_state(
+				untrusted_state_before_by_id,
+				&auth_map,
+				room_version_rules,
+				&create_event,
+			)
+			.await?;
+		let state_before_by_id = state_before
+			.values()
+			.map(|value| (value.event_id().to_owned(), value))
+			.collect::<HashMap<_, _>>();
+
+		info!("Authorizing join event");
+		let final_membership_event =
+			if let Some(event) = send_join_response.room_state.event.as_ref() {
+				event
+			} else {
+				&to_raw_value(&event)
+					.expect("must be able to convert CanonicalJsonObject to RawJsonValue")
+			};
+
+		let (membership_pdu, membership_pdu_json) = self
+			.process_send_join_membership_event(
+				final_membership_event,
+				&event_id,
+				&state_before,
+				&state_before_by_id,
+				room_version_rules,
+			)
+			.await?;
+
+		// We've successfully joined now, and can update our local state cache.
+		info!("Compressing resolved state");
+		let compressed_state_map = state_before
+			.into_iter()
+			.stream()
+			.broad_filter_map(|(key, pdu)| async move {
+				let ssk = self
+					.services
+					.short
+					.get_or_create_shortstatekey(&key.0, key.1.as_str())
+					.await;
+				Some((ssk, pdu.event_id.clone()))
+			})
+			.collect::<Vec<_>>()
+			.await;
+		let compressed: CompressedState = self
+			.services
+			.state_compressor
+			.compress_state_events(
+				compressed_state_map
+					.iter()
+					.map(|(ssk, eid)| (ssk, eid.borrow())),
+			)
+			.collect()
+			.await;
+
+		debug!("Saving compressed state");
+		let HashSetCompressStateEvent {
+			shortstatehash: statehash_before_join,
+			added,
+			removed,
+		} = self
+			.services
+			.state_compressor
+			.save_state(&room_id, Arc::new(compressed))
+			.await?;
+
+		debug!("Updating state cache");
+		self.services
+			.state
+			.force_state(&room_id, statehash_before_join, added, removed, state_lock)
+			.await?;
+
+		let statehash_after_join = self
+			.services
+			.state
+			.append_to_state(&membership_pdu, &room_id)
+			.await?;
+
+		debug!("Promoting membership event");
+		self.services
+			.timeline
+			.append_pdu(
+				&membership_pdu,
+				membership_pdu_json,
+				std::iter::once(membership_pdu.event_id()),
+				state_lock,
+				&room_id,
+			)
+			.await
+			.inspect(|_| info!("Finished joining {room_id}"))?;
+		self.services
+			.metadata
+			.maybe_set_mindepth(&room_id, membership_pdu.depth.into())
+			.await;
+
+		info!("Setting final room state for new room");
+		// We set the room state after inserting the pdu, so that we never have a moment
+		// in time where events in the current room state do not exist
+		self.services
+			.state
+			.set_room_state(&room_id, statehash_after_join, state_lock);
+
+		Ok(Some(()))
+	}
+
+	async fn build_state_map(
+		&self,
+		response: &create_join_event::v2::Response,
+		room_id: &RoomId,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<(PduEvent, HashMap<StateTypeKey, CanonicalJsonObject>)> {
+		// NOTE: this is untrusted as we haven't performed any of the PDU checks yet.
+		// Events in this map should not be persisted blindly.
+		let untrusted_state_before = response
+			.room_state
+			.state
+			.iter()
+			.stream()
+			.wide_filter_map(async |pdu| {
+				let (event_room_id, event_id, value) = self
+					.services
+					.event_handler
+					.parse_incoming_pdu(pdu, Some(room_version_rules))
+					.await
+					.inspect_err(|e| warn!("Invalid PDU in room state (dropping): {e:?}"))
+					.ok()?;
+				if event_room_id != room_id {
+					warn!(%event_id, expected=%room_id, actual=%event_room_id, "PDU in room state belongs to a different room (dropping)");
+					return None;
+				}
+				// value.insert("event_id".to_owned(), event_id.to_string().into());
+				if self.services.pdu_metadata.is_event_rejected(&event_id).await {
+					// Rejection will be re-evaluated later
+					trace!(%event_id, "Un-rejecting event");
+					self.services.pdu_metadata.unmark_event_rejected(&event_id);
+				}
+				Some((state_key_from_json(&value)?, value))
+			})
+			.collect::<HashMap<(StateEventType, StateKey), CanonicalJsonObject>>()
+			.await;
+
+		let create_event_json = untrusted_state_before
+			.get(&StateEventType::RoomCreate.with_state_key(""))
+			.ok_or_else(|| {
+				err!("Room state returned from send_join did not contain a room create event")
+			})?;
+		let create_event_id = gen_event_id(create_event_json, room_version_rules)
+			.map_err(|e| err!("Failed to generate event ID for room create event: {e:?}"))?;
+		let create_event = PduEvent::from_id_val(&create_event_id, create_event_json.clone())?;
+
+		Ok((create_event, untrusted_state_before))
+	}
+
+	async fn process_send_join_auth_chain(
+		&self,
+		response: &create_join_event::v2::Response,
+		room_id: &RoomId,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+	) -> Result<BTreeMap<OwnedEventId, PduEvent>> {
+		let auth_chain_timing_start = Instant::now();
+		let unauthed_auth_chain = response
+			.room_state
+			.auth_chain
+			.iter()
+			.stream()
+			.broad_filter_map(|value| async move {
+				self
+					.services
+					.event_handler
+					.parse_incoming_pdu(
+						&to_raw_value(&value)
+							.expect("CanonicalJsonObject just convert to RawJsonValue"),
+						Some(room_version_rules),
+					)
+					.await
+					.inspect_err(|e| warn!("Invalid PDU in send_join auth chain (dropping): {e:?}"))
+					.ok()
+			})
+			.broad_filter_map(|(auth_room_id, event_id, value)| async move {
+				if self.services.pdu_metadata.is_event_rejected(&event_id).await {
+					// Rejection will be re-evaluated later
+					trace!(%event_id, "Un-rejecting event");
+					self.services.pdu_metadata.unmark_event_rejected(&event_id);
+				}
+				trace!(%event_id, "Validating event");
+				if auth_room_id != room_id {
+					warn!(%event_id, %room_id, %auth_room_id, "PDU in send_join auth chain belongs to a different room (dropping)");
+					return None;
+				}
+
+				crate::rooms::event_handler::Service::pdu_format_check_1(&value, room_version_rules, create_event.event_id()).inspect_err(|e| {
+					warn!(%event_id, "PDU in send_join auth chain failed format check (dropping): {e:?}");
+				}).ok()?;
+
+				let mut value = self.services.event_handler.signature_hash_check_2_3(value, room_version_rules).await.inspect_err(|e| {
+					warn!(%event_id, "PDU in send_join auth chain failed signature check (dropping): {e:?}");
+				}).ok()?;
+
+				value.insert("event_id".to_owned(), event_id.as_str().into());
+				Some((event_id, value))
+			})
+			.collect::<HashMap<_, _>>()
+			.await;
+		debug!(
+			elapsed=?auth_chain_timing_start.elapsed(),
+			"Finished validating auth chain ({}/{} events passed validation)",
+			unauthed_auth_chain.len(),
+			response.room_state.auth_chain.len()
+		);
+		let auth_chain = build_local_dag(&unauthed_auth_chain, DagBuilderTree::AuthEvents)
+			.await?
+			.into_iter()
+			.stream()
+			.wide_filter_map(async |event_id| {
+				// Perform PDU check 4 to make sure the event doesn't need
+				// rejecting
+				let pdu =
+					PduEvent::from_id_val(&event_id, unauthed_auth_chain[&event_id].clone())
+						.expect("We already validated auth chian PDUs");
+				let auth_state = pdu
+					.auth_events()
+					.filter_map(|event_id| {
+						PduEvent::from_id_val(
+							event_id,
+							unauthed_auth_chain.get(event_id)?.clone(),
+						)
+						.map(|pdu| {
+							Some((
+								pdu.kind().with_state_key(
+									pdu.state_key()
+										.expect("Auth chain events must have a state key"),
+								),
+								pdu,
+							))
+						})
+						.ok()?
+					})
+					.collect::<HashMap<_, _>>();
+				self.services
+					.event_handler
+					.auth_state_check_4(
+						&pdu,
+						room_version_rules,
+						create_event.as_pdu(),
+						&auth_state,
+					)
+					.await
+					.inspect(|result| {
+						if !*result {
+							// NOTE: usually we don't *drop* events that fail PDU check 4, but
+							// send_join handling is special. Instead of rejecting this event and
+							// consequently every other event that depends on it, we simply drop
+							// it from the chain
+							warn!(%event_id, "Auth chain event failed self-authorization (dropping)");
+						}
+					})
+					.inspect_err(
+						|e| error!(%event_id, ?e, "Failed to run auth check on auth chain event"),
+					)
+					.unwrap_or_default()
+					.then_some((event_id, pdu))
+			})
+			.collect::<BTreeMap<_, _>>()
+			.await;
+		debug!(
+			elapsed=?auth_chain_timing_start.elapsed(),
+			"Finished authentication auth chain ({}/{} events passed authentication)",
+			auth_chain.len(),
+			unauthed_auth_chain.len()
+		);
+		// Now we need to persist them
+		auth_chain
+			.keys()
+			.stream()
+			.for_each_concurrent(automatic_width(), async |event_id| {
+				if self.services.timeline.get_pdu(event_id).await.is_err() {
+					// Only add events that we don't already have
+					self.services
+						.outlier
+						.add_pdu_outlier(event_id, &unauthed_auth_chain[event_id]);
+				}
+			})
+			.await;
+		info!(elapsed=?auth_chain_timing_start.elapsed(), "Finished processing authentication events");
+
+		Ok(auth_chain)
+	}
+
+	async fn process_send_join_state(
+		&self,
+		untrusted_state_before: HashMap<OwnedEventId, &CanonicalJsonObject>,
+		auth_chain: &BTreeMap<OwnedEventId, PduEvent>,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+	) -> Result<HashMap<StateTypeKey, PduEvent>> {
+		let state_timing_start = Instant::now();
+		let state_size = untrusted_state_before.len();
+		// NOTE: unlike the auth chain, we need to run PDU checks 1 through 4.
+		let state_before = untrusted_state_before
+			.into_iter()
+			.stream()
+			.broad_filter_map(|(event_id, value)| async move {
+				crate::rooms::event_handler::Service::pdu_format_check_1(
+					value,
+					room_version_rules,
+					create_event.event_id(),
+				)
+				.inspect_err(|e| {
+					warn!(%event_id, "PDU in send_join state failed format check (dropping): {e:?}");
+				})
+				.ok()?;
+
+				let mut value = self
+					.services
+					.event_handler
+					.signature_hash_check_2_3(value.clone(), room_version_rules)
+					.await
+					.inspect_err(|e| {
+						warn!(%event_id, "PDU in send_join state failed signature check (dropping): {e:?}");
+					})
+					.ok()?;
+
+				value.insert("event_id".to_owned(), event_id.as_str().into());
+
+				let pdu = PduEvent::from_id_val(&event_id, value.clone())
+					.expect("We already validated this state event");
+				let auth_state = pdu
+					.auth_events()
+					.filter_map(|event_id| {
+						auth_chain.get(event_id).map(|p| {
+							(
+								p.kind().with_state_key(
+									p.state_key()
+										.expect("auth chain events must have a state key"),
+								),
+								p.clone(),
+							)
+						})
+					})
+					.collect::<HashMap<_, _>>();
+				self.services
+					.event_handler
+					.auth_state_check_4(
+						&pdu,
+						room_version_rules,
+						create_event.as_pdu(),
+						&auth_state,
+					)
+					.await
+					.inspect(|result| {
+						if !*result {
+							warn!(%event_id, "State event failed self-authorization (dropping)");
+						}
+					})
+					.inspect_err(
+						|e| error!(%event_id, ?e, "Failed to run auth check on state event"),
+					)
+					.unwrap_or_default()
+					.then_some((pdu, value))
+			})
+			.collect::<Vec<_>>()
+			.await;
+		debug!(
+			elapsed=?state_timing_start.elapsed(),
+			"Finished validation and authentication of room state ({}/{} events retained)",
+			state_before.len(),
+			state_size,
+		);
+
+		state_before
+			.iter()
+			.stream()
+			.for_each_concurrent(automatic_width(), async |(pdu, pdu_json)| {
+				if self
+					.services
+					.timeline
+					.get_pdu(pdu.event_id())
+					.await
+					.is_err()
+				{
+					// Only add events that we don't already have
+					self.services
+						.outlier
+						.add_pdu_outlier(pdu.event_id(), pdu_json);
+				}
+			})
+			.await;
+
+		info!(
+			elapsed=?state_timing_start.elapsed(),
+			"Finished processing send join state ({}/{} events)",
+			state_before.len(),
+			state_size,
+		);
+
+		let state_before_by_key = state_before
+			.into_iter()
+			.map(|(pdu, _)| {
+				(
+					pdu.kind().with_state_key(
+						pdu.state_key().expect("state event must have a state key"),
+					),
+					pdu,
+				)
+			})
+			.collect::<HashMap<_, _>>();
+
+		Ok(state_before_by_key)
+	}
+
+	async fn process_send_join_membership_event(
+		&self,
+		membership_event: &serde_json::value::RawValue,
+		event_id: &EventId,
+		state_before: &HashMap<StateTypeKey, PduEvent>,
+		state_before_by_id: &HashMap<OwnedEventId, &PduEvent>,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<(PduEvent, CanonicalJsonObject)> {
+		let (_, calculated_event_id, value) = self
+			.services
+			.event_handler
+			.parse_incoming_pdu(membership_event, Some(room_version_rules))
+			.await?;
+
+		if calculated_event_id != event_id {
+			return Err!(Request(InvalidParam(debug_warn!(
+				expected=%event_id,
+				actual=%calculated_event_id,
+				"Remote server returned a different membership event to the one we sent it"
+			))));
+		}
+
+		if self.services.pdu_metadata.is_event_rejected(event_id).await {
+			// Rejection will be re-evaluated later
+			trace!(%event_id, "Un-rejecting event");
+			self.services.pdu_metadata.unmark_event_rejected(event_id);
+		}
+
+		let create_event = state_before
+			.get(&StateEventType::RoomCreate.with_state_key(""))
+			.expect("create event must be in state before");
+		crate::rooms::event_handler::Service::pdu_format_check_1(
+			&value,
+			room_version_rules,
+			create_event.event_id(),
+		)?;
+
+		let mut value = self
+			.services
+			.event_handler
+			.signature_hash_check_2_3(value, room_version_rules)
+			.await?;
+
+		value.insert("event_id".to_owned(), event_id.as_str().into());
+
+		let pdu = PduEvent::from_id_val(event_id, value.clone())
+			.expect("We already validated this state event");
+
+		// Perform check 4 to make sure the PDU is self-authorised
+		let auth_state = pdu
+			.auth_events()
+			.filter_map(|event_id| {
+				state_before_by_id
+					.get(event_id)
+					.copied()
+					.map(|p| (p.kind().with_state_key(p.state_key().unwrap()), p.clone()))
+			})
+			.collect::<HashMap<_, _>>();
+		if !self
+			.services
+			.event_handler
+			.auth_state_check_4(&pdu, room_version_rules, create_event.as_pdu(), &auth_state)
+			.await?
+		{
+			return Err!(Request(Forbidden("Membership event was not self-authorised")));
+		}
+
+		// We can also perform check 5 (state before), since the state returned in the
+		// response is the state just before our join. However, state_before_check_5
+		// (part of the event_handler service) won't understand this, so we have to do
+		// the check manually.
+		let state_fetch =
+			|k: StateEventType, s: StateKey| ready(state_before.get(&(k, s)).cloned());
+
+		// PDU check: 5
+		let auth_check = state_res::event_auth::auth_check(
+			room_version_rules,
+			&pdu,
+			None, // TODO: third party invite
+			|ty, sk| state_fetch(ty.clone(), sk.into()),
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
+		if !auth_check {
+			return Err!(Request(Forbidden(
+				"Membership event was not authorised based on the state before it"
+			)));
+		}
+
+		Ok((pdu, value))
+	}
+}

src/service/rooms/membership/mod.rs

@@ -1,26 +1,17 @@
-use std::{collections::HashMap, sync::Arc};
+mod event;
+mod join;
+
+use std::{collections::VecDeque, sync::Arc};
 
 use conduwuit::{
-	Err, Event, Pdu, Result, Server, debug, debug_info, debug_warn, err, error, info, is_true,
-	matrix::{
-		StateKey,
-		event::{gen_event_id, gen_event_id_canonical_json},
-	},
-	pdu::PartialPdu,
-	state_res, trace,
-	utils::{self, IterStream, ReadyExt, to_canonical_object},
-	warn,
+	Err, Result, Server, debug, debug_info, debug_warn, info, is_true,
+	matrix::event::gen_event_id, pdu::PartialPdu, warn,
 };
 use database::Database;
-use futures::{FutureExt, StreamExt, TryFutureExt, join};
+use futures::{FutureExt, StreamExt, join};
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, OwnedRoomId, OwnedServerName, OwnedUserId, RoomId,
 	RoomVersionId, UserId,
-	api::{
-		error::{ErrorKind, IncompatibleRoomVersionErrorData},
-		federation,
-	},
-	canonical_json::to_canonical_value,
 	events::{
 		StateEventType, StaticEventContent,
 		room::{
@@ -34,11 +25,13 @@
 use crate::{
 	Dep, antispam, globals,
 	rooms::{
-		event_handler, metadata, outlier, pdu_metadata, short,
+		event_handler,
+		membership::join::MakeJoinResult,
+		metadata, outlier, pdu_metadata, short,
 		state::{self, RoomMutexGuard},
 		state_accessor, state_cache,
-		state_compressor::{self, CompressedState, HashSetCompressStateEvent},
-		timeline::{self, pdu_fits},
+		state_compressor::{self},
+		timeline,
 	},
 	sending, server_keys, sync, users,
 };
@@ -99,18 +92,19 @@ fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
 impl Service {
-	/// Join a local user to a room.
+	/// Join a local user to a room. If the room is not local, this will attempt
+	/// to join via federation. If the room cannot be joined locally, a
+	/// federation join may be attempted. Users trying to join a room they're
+	/// already joined to will short-circuit.
 	pub async fn join_room(
 		&self,
 		sender_user: &UserId,
 		room_id: &RoomId,
 		reason: Option<String>,
-		servers: &[OwnedServerName],
+		servers: Vec<OwnedServerName>,
 	) -> Result<OwnedRoomId> {
 		assert!(self.services.globals.user_is_local(sender_user), "user should be local");
 
-		let state_lock = self.services.state.mutex.lock(room_id).await;
-
 		if self
 			.services
 			.state_cache
@@ -187,12 +181,12 @@ pub async fn join_room(
 		}
 
 		if server_in_room {
-			self.join_local_room(sender_user, room_id, reason, servers, state_lock)
+			self.join_local_room(sender_user, room_id, reason, servers)
 				.boxed()
 				.await?;
 		} else {
 			// Ask a remote server if we are not participating in this room
-			self.join_remote_room(sender_user, room_id, reason, servers, state_lock)
+			self.join_remote_room(sender_user, room_id, reason, servers)
 				.boxed()
 				.await?;
 		}
@@ -206,11 +200,11 @@ async fn join_local_room(
 		sender_user: &UserId,
 		room_id: &RoomId,
 		reason: Option<String>,
-		servers: &[OwnedServerName],
-		state_lock: RoomMutexGuard,
+		servers: Vec<OwnedServerName>,
 	) -> Result {
 		info!("Joining room locally");
 
+		let state_lock = self.services.state.mutex.lock(room_id).await;
 		let (room_version, join_rules, is_invited) = join!(
 			self.services.state.get_room_version(room_id),
 			self.services.state_accessor.get_join_rules(room_id),
@@ -264,6 +258,7 @@ async fn join_local_room(
 			info!("Joined room locally");
 			return Ok(());
 		};
+		drop(state_lock);
 
 		if servers.is_empty()
 			|| servers.len() == 1 && self.services.globals.server_is_ours(&servers[0])
@@ -286,7 +281,7 @@ async fn join_local_room(
 			remote_servers = %servers.len(),
 			"Could not join room locally, attempting remote join",
 		);
-		Box::pin(self.join_remote_room(sender_user, room_id, reason, servers, state_lock)).await
+		Box::pin(self.join_remote_room(sender_user, room_id, reason, servers)).await
 	}
 
 	#[tracing::instrument(skip_all, fields(%sender_user, %room_id), name = "join_remote_room", level = "info")]
@@ -295,104 +290,83 @@ pub async fn join_remote_room(
 		sender_user: &UserId,
 		room_id: &RoomId,
 		reason: Option<String>,
-		servers: &[OwnedServerName],
-		state_lock: RoomMutexGuard,
+		vias: Vec<OwnedServerName>,
 	) -> Result {
-		// public so the admin command force-join-room-remotely works
 		info!("Joining {room_id} over federation.");
 
-		let (make_join_response, remote_server) = self
-			.make_join_request(sender_user, room_id, servers)
-			.await?;
+		// Treat the via list as a "priority queue", and each time a make_join fails,
+		// move the server to the back of the queue. This way, if a server is down, we
+		// will eventually try all servers in the list.
+		let mut priority_queue = VecDeque::from(vias.clone());
+		let mut template: Option<CanonicalJsonObject> = None;
+		let mut room_version: Option<RoomVersionId> = None;
+		let via_count = vias.len();
+
+		for via in vias {
+			let response = self.make_join(room_id, sender_user, &via).await;
+			match response {
+				| MakeJoinResult::Success((t, r)) => {
+					template = Some(t);
+					room_version = Some(r);
+					break;
+				},
+				| MakeJoinResult::Fatal(e) => {
+					return Err(e);
+				},
+				| MakeJoinResult::Retry => {
+					// Pop the front to remove this server from the queue, and
+					// then push it to the back.
+					// We need to do this vec iter + dequeue to avoid infinite
+					// looping when we reprioritise.
+					let _ = priority_queue.pop_front();
+					priority_queue.push_back(via);
+				},
+			}
+		}
+
+		if template.is_none() {
+			info!("All {} servers were unable to assist in joining {room_id} :(", via_count);
+			return Err!(BadServerResponse("No server available to assist in joining."));
+		}
+		let room_version =
+			room_version.expect("room_version cannot be None while template is Some");
 
 		info!("make_join finished");
 
-		let room_version = make_join_response.room_version.unwrap_or(RoomVersionId::V1);
-		let room_version_rules = room_version
-			.rules()
-			.expect("room version should have defined rules");
-
 		if !self.services.server.supported_room_version(&room_version) {
 			// How did we get here?
 			return Err!(BadServerResponse(
 				"Remote room version {room_version} is not supported"
 			));
 		}
+		let room_version_rules = room_version
+			.rules()
+			.expect("room version should have defined rules");
 
-		let mut join_event_stub: CanonicalJsonObject =
-			serde_json::from_str(make_join_response.event.get()).map_err(|e| {
-				err!(BadServerResponse(warn!(
-					"Invalid make_join event json received from server: {e:?}"
-				)))
-			})?;
-
-		let join_authorized_via_users_server = {
-			if room_version_rules
-				.signatures
-				.check_join_authorised_via_users_server
-			{
-				join_event_stub
-					.get("content")
-					.map(|s| {
-						s.as_object()?
-							.get("join_authorised_via_users_server")?
-							.as_str()
-					})
-					.and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok())
-			} else {
-				None
-			}
-		};
-
-		join_event_stub.insert(
-			"origin_server_ts".to_owned(),
-			CanonicalJsonValue::Integer(
-				utils::millis_since_unix_epoch()
-					.try_into()
-					.expect("Timestamp is valid js_int value"),
-			),
-		);
-
-		let mut join_content = RoomMemberEventContent::new(MembershipState::Join);
-		join_content.displayname = self.services.users.displayname(sender_user).await.ok();
-		join_content.avatar_url = self.services.users.avatar_url(sender_user).await.ok();
-		join_content.reason = reason;
-		join_content
-			.join_authorized_via_users_server
-			.clone_from(&join_authorized_via_users_server);
-
-		join_event_stub.insert(
-			"content".to_owned(),
-			to_canonical_value(join_content).expect("event is valid, we just created it"),
-		);
-
-		// Remove event id if it exists
-		join_event_stub.remove("event_id");
+		let mut template = self
+			.seed_local_membership_pdu(
+				room_id,
+				sender_user,
+				MembershipState::Join,
+				reason,
+				template.unwrap(),
+				&room_version_rules,
+			)
+			.await?;
 
 		// In order to create a compatible ref hash (EventID) the `hashes` field needs
 		// to be present
 		self.services
 			.server_keys
-			.hash_and_sign_event(&mut join_event_stub, &room_version_rules)?;
+			.hash_and_sign_event(&mut template, &room_version_rules)?;
 
 		// Generate event id
-		let event_id = gen_event_id(&join_event_stub, &room_version_rules)?;
+		let event_id = gen_event_id(&template, &room_version_rules)?;
 
 		// Add event_id back
-		join_event_stub
+		template
 			.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.clone().into()));
 
-		// It has enough fields to be called a proper event now
-		let mut join_event = join_event_stub;
-		let send_join_request = federation::membership::create_join_event::v2::Request::new(
-			room_id.to_owned(),
-			event_id.clone(),
-			self.services
-				.sending
-				.convert_to_outgoing_federation_event(join_event.clone())
-				.await,
-		);
-
 		// NOTE: send_join can take a long time to respond, but from the point of view
 		// of other servers, we may already have finished joining. This means they
 		// sometimes end up sending PDUs to us that we aren't yet ready to accept, and
@@ -404,368 +378,32 @@ pub async fn join_remote_room(
 			.mutex_federation
 			.lock(room_id.as_str())
 			.await;
-		info!("Asking {remote_server} for send_join in room {room_id}");
-		let send_join_response = match self
-			.services
-			.sending
-			.send_slow_federation_request(&remote_server, send_join_request)
-			.await
-		{
-			| Ok(response) => response,
-			| Err(e) => {
-				error!("send_join failed: {e}");
-				return Err(e);
-			},
-		};
-
-		info!("send_join finished");
-
-		if join_authorized_via_users_server.is_some() {
-			if let Some(signed_raw) = &send_join_response.room_state.event {
-				debug_info!(
-					"There is a signed event with join_authorized_via_users_server. This room \
-					 is probably using restricted joins. Adding signature to our event"
-				);
-
-				let (signed_event_id, signed_value) =
-					gen_event_id_canonical_json(signed_raw, &room_version_rules).map_err(
-						|e| {
-							err!(Request(BadJson(warn!(
-								"Could not convert event to canonical JSON: {e}"
-							))))
-						},
-					)?;
-
-				if signed_event_id != event_id {
-					return Err!(Request(BadJson(warn!(
-						%signed_event_id, %event_id,
-						"Server {remote_server} sent event with wrong event ID"
-					))));
-				}
-
-				match signed_value["signatures"]
-					.as_object()
-					.ok_or_else(|| {
-						err!(BadServerResponse(warn!(
-							"Server {remote_server} sent invalid signatures type"
-						)))
-					})
-					.and_then(|e| {
-						e.get(remote_server.as_str()).ok_or_else(|| {
-							err!(BadServerResponse(warn!(
-								"Server {remote_server} did not send its signature for a \
-								 restricted room"
-							)))
-						})
-					}) {
-					| Ok(signature) => {
-						join_event
-							.get_mut("signatures")
-							.expect("we created a valid pdu")
-							.as_object_mut()
-							.expect("we created a valid pdu")
-							.insert(remote_server.to_string(), signature.clone());
-					},
-					| Err(e) => {
-						warn!(
-							"Server {remote_server} sent invalid signature in send_join \
-							 signatures for event {signed_value:?}: {e:?}",
-						);
-					},
-				}
-			}
-		}
-
-		self.services.short.get_or_create_shortroomid(room_id).await;
-
-		info!("Parsing join event");
-		let parsed_join_pdu = Pdu::from_id_val(&event_id, join_event.clone())
-			.map_err(|e| err!(BadServerResponse("Invalid join event PDU: {e:?}")))?;
-
-		info!("Acquiring server signing keys for response events");
-		let resp_events = &send_join_response.room_state;
-		let resp_state = &resp_events.state;
-		let resp_auth = &resp_events.auth_chain;
-		self.services
-			.server_keys
-			.acquire_events_pubkeys(resp_auth.iter().chain(resp_state.iter()))
-			.await;
-
-		info!("Going through send_join response room_state");
-		let cork = self.services.db.cork_and_flush();
-		let state = send_join_response
-			.room_state
-			.state
-			.iter()
-			.stream()
-			.then(|pdu| {
-				self.services
-					.server_keys
-					.validate_and_add_event_id_no_fetch(pdu, &room_version_rules)
-					.inspect_err(|e| {
-						debug_warn!(
-							"Could not validate send_join response room_state event: {e:?}"
-						);
-					})
-					.inspect(|_| {
-						debug!("Completed validating send_join response room_state event");
-					})
-			})
-			.ready_filter_map(Result::ok)
-			.fold(HashMap::new(), |mut state, (event_id, value)| async move {
-				let pdu = match Pdu::from_id_val(&event_id, value.clone()) {
-					| Ok(pdu) => pdu,
-					| Err(e) => {
-						debug_warn!("Invalid PDU in send_join response: {e:?}: {value:#?}");
-						return state;
-					},
-				};
-				if !pdu_fits(&value.clone()) {
-					warn!(
-						"dropping incoming PDU {event_id} in room {room_id} from room join \
-						 because it exceeds 65535 bytes or is otherwise too large."
-					);
-					return state;
-				}
-				self.services.outlier.add_pdu_outlier(&event_id, &value);
-				self.services.pdu_metadata.clear_pdu_markers(&event_id);
-				if let Some(state_key) = &pdu.state_key {
-					let shortstatekey = self
-						.services
-						.short
-						.get_or_create_shortstatekey(&pdu.kind.to_string().into(), state_key)
-						.await;
-
-					state.insert(shortstatekey, pdu.event_id.clone());
-				}
-				state
-			})
-			.await;
-
-		drop(cork);
-
-		info!("Going through send_join response auth_chain");
-		let cork = self.services.db.cork_and_flush();
-		send_join_response
-			.room_state
-			.auth_chain
-			.iter()
-			.stream()
-			.then(|pdu| {
-				self.services
-					.server_keys
-					.validate_and_add_event_id_no_fetch(pdu, &room_version_rules)
-			})
-			.ready_filter_map(Result::ok)
-			.ready_for_each(|(event_id, value)| {
-				trace!(%event_id, "Adding PDU as an outlier from send_join auth_chain");
-				self.services.outlier.add_pdu_outlier(&event_id, &value);
-				self.services.pdu_metadata.clear_pdu_markers(&event_id);
-			})
-			.await;
-
-		drop(cork);
-
-		debug!("Running send_join auth check");
-		let fetch_state = &state;
-		let state_fetch = |k: StateEventType, s: StateKey| async move {
-			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
-
-			let event_id = fetch_state.get(&shortstatekey)?;
-			self.services.timeline.get_pdu(event_id).await.ok()
-		};
-
-		let auth_check = state_res::event_auth::auth_check(
-			&room_version.rules().unwrap(),
-			&parsed_join_pdu,
-			None, // TODO: third party invite
-			|k, s| state_fetch(k.clone(), s.into()),
-			&state_fetch(StateEventType::RoomCreate, "".into())
-				.await
-				.expect("create event is missing from send_join auth"),
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden(warn!("Auth check failed: {e:?}")))))?;
-
-		if !auth_check {
-			return Err!(Request(Forbidden("Auth check failed")));
-		}
-		let resident_before = self
-			.services
-			.state_cache
-			.server_in_room(self.services.globals.server_name(), room_id)
-			.await;
-
-		let cork = self.services.db.cork_and_flush();
-		info!("Compressing state from send_join");
-		let compressed: CompressedState = self
-			.services
-			.state_compressor
-			.compress_state_events(state.iter().map(|(ssk, eid)| (ssk, eid.as_ref())))
-			.collect()
-			.await;
-
-		debug!("Saving compressed state");
-		let HashSetCompressStateEvent {
-			shortstatehash: statehash_before_join,
-			added,
-			removed,
-		} = self
-			.services
-			.state_compressor
-			.save_state(room_id, Arc::new(compressed))
-			.await?;
-
-		debug!("Forcing state for new room");
-		self.services
-			.state
-			.force_state(room_id, statehash_before_join, added, removed, &state_lock)
-			.await?;
-
-		debug!("Updating joined counts for new room");
-		self.services.state_cache.update_joined_count(room_id).await;
-
-		// We append to state before appending the pdu, so we don't have a moment in
-		// time with the pdu without it's state. This is okay because append_pdu can't
-		// fail.
-		let statehash_after_join = self
-			.services
-			.state
-			.append_to_state(&parsed_join_pdu, room_id)
-			.await?;
-
-		info!("Appending new room join event");
-		self.services
-			.timeline
-			.append_pdu(
-				&parsed_join_pdu,
-				join_event,
-				std::iter::once(parsed_join_pdu.event_id.as_ref()),
-				&state_lock,
-				room_id,
-			)
-			.await?;
-		self.services
-			.metadata
-			.maybe_set_mindepth(room_id, parsed_join_pdu.depth.into())
-			.await;
-
-		info!("Setting final room state for new room");
-		// We set the room state after inserting the pdu, so that we never have a moment
-		// in time where events in the current room state do not exist
-		self.services
-			.state
-			.set_room_state(room_id, statehash_after_join, &state_lock);
-		if !resident_before {
-			// NOTE: We replace local extremities for this room if we were not a resident
-			// before. We might be doing a remote join to satisfy restricted join rules,
-			// so we don't want to do this if we're already a resident. Otherwise, we
-			// want to replace our forward extremities whole-sale in case we were
-			// desynced.
-			info!("Replacing local forward extremities");
-			self.services
-				.state
-				.set_forward_extremities(
-					room_id,
-					std::iter::once(parsed_join_pdu.event_id()),
+		let state_lock = self.services.state.mutex.lock(room_id).await;
+		while let Some(via) = priority_queue.pop_front() {
+			let cork = self.services.db.cork_and_sync();
+			let result = self
+				.send_join(
+					room_id.to_owned(),
+					event_id.clone(),
+					&template,
+					&via,
+					&room_version_rules,
 					&state_lock,
 				)
-				.await;
-		}
-		drop(cork);
-
-		self.services.sync.wake_all_joined(room_id).await;
-
-		Ok(())
-	}
-
-	async fn make_join_request(
-		&self,
-		sender_user: &UserId,
-		room_id: &RoomId,
-		servers: &[OwnedServerName],
-	) -> Result<(federation::membership::prepare_join_event::v1::Response, OwnedServerName)> {
-		let mut make_join_counter: usize = 1;
-
-		for remote_server in servers {
-			if self.services.globals.server_is_ours(remote_server) {
+				.await?;
+			if result.is_none() {
 				continue;
 			}
-			info!(
-				"Asking {remote_server} for make_join (attempt {make_join_counter}/{})",
-				servers.len()
-			);
-
-			let mut request = federation::membership::prepare_join_event::v1::Request::new(
-				room_id.to_owned(),
-				sender_user.to_owned(),
-			);
-			request.ver = self.services.server.supported_room_versions().collect();
-
-			let make_join_response = self
-				.services
-				.sending
-				.send_federation_request(remote_server, request)
-				.await;
-
-			trace!("make_join response: {:?}", make_join_response);
-			make_join_counter = make_join_counter.saturating_add(1);
-
-			match make_join_response {
-				| Ok(response) => {
-					info!("Received make_join response from {remote_server}");
-					if let Err(e) = validate_remote_member_event_stub(
-						&MembershipState::Join,
-						sender_user,
-						room_id,
-						&to_canonical_object(&response.event)?,
-					) {
-						warn!("make_join response from {remote_server} failed validation: {e}");
-						continue;
-					}
-					return Ok((response, remote_server.clone()));
-				},
-				| Err(e) => match e.kind() {
-					| ErrorKind::UnableToAuthorizeJoin => {
-						info!(
-							"{remote_server} was unable to verify the joining user satisfied \
-							 restricted join requirements: {e}. Will continue trying."
-						);
-					},
-					| ErrorKind::UnableToGrantJoin => {
-						info!(
-							"{remote_server} believes the joining user satisfies restricted \
-							 join rules, but is unable to authorise a join for us. Will \
-							 continue trying."
-						);
-					},
-					| ErrorKind::IncompatibleRoomVersion(IncompatibleRoomVersionErrorData {
-						room_version,
-						..
-					}) => {
-						warn!(
-							"{remote_server} reports the room we are trying to join is \
-							 v{room_version}, which we do not support."
-						);
-						return Err(e);
-					},
-					| ErrorKind::Forbidden => {
-						warn!("{remote_server} refuses to let us join: {e}.");
-						return Err(e);
-					},
-					| ErrorKind::NotFound => {
-						info!(
-							"{remote_server} does not know about {room_id}: {e}. Will continue \
-							 trying."
-						);
-					},
-					| _ => {
-						info!("{remote_server} failed to make_join: {e}. Will continue trying.");
-					},
-				},
-			}
+			info!("send_join finished");
+			drop(cork);
+			self.services.sync.wake_all_joined(room_id).await;
+			return Ok(());
 		}
-		info!("All {} servers were unable to assist in joining {room_id} :(", servers.len());
+
+		info!(
+			"Despite producing a membership template, no server was capable of actually \
+			 completing the join for us."
+		);
 		Err!(BadServerResponse("No server available to assist in joining."))
 	}
 

src/service/rooms/outlier/mod.rs

@@ -45,6 +45,10 @@ pub async fn get_pdu_outlier(&self, event_id: &EventId) -> Result<PduEvent> {
 
 	/// Append the PDU as an outlier.
 	pub fn add_pdu_outlier(&self, event_id: &EventId, pdu: &CanonicalJsonObject) {
+		debug_assert!(
+			pdu.get("event_id").is_some(),
+			"event_id must be present in database events"
+		);
 		self.db.eventid_outlierpdu.raw_put(event_id, Json(pdu));
 	}
 }

src/service/rooms/timeline/data.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use conduwuit::{
-	Err, PduCount, PduEvent, Result, at, err,
+	Err, Event, PduCount, PduEvent, Result, at, err,
 	result::NotFound,
 	utils::{self, stream::TryReadyExt},
 };
@@ -176,6 +176,15 @@ pub(super) async fn append_pdu(
 		count: PduCount,
 	) {
 		debug_assert!(matches!(count, PduCount::Normal(_)), "PduCount not Normal");
+		debug_assert!(
+			json.get("event_id").is_some(),
+			"event_id must be present in database events"
+		);
+		debug_assert_eq!(
+			json.get("event_id").unwrap().as_str().unwrap(),
+			pdu.event_id().as_str(),
+			"event ID mismatch between PDU JSON and PDU object"
+		);
 
 		self.pduid_pdu.raw_put(pdu_id, Json(json));
 		self.eventid_pduid.insert(pdu.event_id.as_bytes(), pdu_id);
@@ -188,6 +197,10 @@ pub(super) fn prepend_backfill_pdu(
 		event_id: &EventId,
 		json: &CanonicalJsonObject,
 	) {
+		debug_assert!(
+			json.get("event_id").is_some(),
+			"event_id must be present in database events"
+		);
 		self.pduid_pdu.raw_put(pdu_id, Json(json));
 		self.eventid_pduid.insert(event_id, pdu_id);
 		self.eventid_outlierpdu.remove(event_id);
@@ -202,7 +215,10 @@ pub(super) async fn replace_pdu(
 		if self.pduid_pdu.get(pdu_id).await.is_not_found() {
 			return Err!(Request(NotFound("PDU does not exist.")));
 		}
-
+		debug_assert!(
+			pdu_json.get("event_id").is_some(),
+			"event_id must be present in database events"
+		);
 		self.pduid_pdu.raw_put(pdu_id, Json(pdu_json));
 
 		Ok(())

src/service/server_keys/sign.rs

@@ -18,6 +18,12 @@ pub fn hash_and_sign_event(
 		object: &mut CanonicalJsonObject,
 		room_version_rules: &RoomVersionRules,
 	) -> Result {
+		#[cfg(debug_assertions)]
+		{
+			if object.contains_key("event_id") {
+				conduwuit::warn!("Signing a PDU that has an `event_id` present");
+			}
+		}
 		hash_and_sign_event(
 			self.services.globals.server_name().as_str(),
 			self.keypair(),

src/service/users/account.rs

@@ -244,7 +244,7 @@ pub async fn create_local_account(
 						user_id,
 						&room_id,
 						Some("Automatically joining this room upon registration".to_owned()),
-						&[
+						vec![
 							self.services.globals.server_name().to_owned(),
 							room_server_name.to_owned(),
 						],