refactor: Use new make_join and send_join functions in membership service

src/admin/user/commands.rs

@@ -541,7 +541,12 @@ impl crate::Context<'_> {
 				.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) => {
@@ -623,7 +628,12 @@ impl crate::Context<'_> {
 				.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) => {
@@ -670,7 +680,7 @@ impl crate::Context<'_> {
 		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

@@ -67,7 +67,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?;
 
@@ -179,7 +179,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

@@ -132,7 +132,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
 }
@@ -142,7 +142,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;
 
@@ -233,7 +233,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())),
@@ -317,7 +317,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");
@@ -478,7 +478,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.");
@@ -686,7 +686,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."));
@@ -694,7 +694,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;
 		}
 
@@ -708,7 +708,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/service/rooms/membership/join.rs

@@ -181,15 +181,15 @@ impl super::Service {
 		}
 	}
 
-	async fn send_join(
+	pub async fn send_join(
 		&self,
 		room_id: OwnedRoomId,
 		event_id: OwnedEventId,
-		event: CanonicalJsonObject,
+		event: &CanonicalJsonObject,
 		via: &ServerName,
 		room_version_rules: &RoomVersionRules,
 		state_lock: &RoomMutexGuard,
-	) -> Result {
+	) -> Result<Option<()>> {
 		let send_join_request = create_join_event::v2::Request::new(
 			room_id.clone(),
 			event_id.clone(),
@@ -200,11 +200,7 @@ impl super::Service {
 		);
 
 		info!("Asking {via} for send_join in room {room_id}");
-		self.services
-			.short
-			.get_or_create_shortroomid(&room_id)
-			.await;
-		let send_join_response = self
+		let Ok(send_join_response) = self
 			.services
 			.sending
 			.send_slow_federation_request(via, send_join_request)
@@ -217,7 +213,14 @@ impl super::Service {
 					resp.room_state.auth_chain.len(),
 				);
 			})
-			.inspect_err(|e| debug_error!("{via} failed to send_join: {e:?}"))?;
+			.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
@@ -232,9 +235,19 @@ impl super::Service {
 			.await;
 
 		info!("Building state map");
-		let (create_event, untrusted_state_before) = self
+		let (create_event, untrusted_state_before) = match self
 			.build_state_map(&send_join_response, &room_id, room_version_rules)
-			.await?;
+			.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)",
@@ -338,6 +351,12 @@ impl super::Service {
 			.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
@@ -349,8 +368,20 @@ impl super::Service {
 				&room_id,
 			)
 			.await
-			.inspect(|_| info!("Finished joining {room_id}"))
-			.map(|_| ())
+			.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(

src/service/rooms/membership/mod.rs

@@ -1,30 +1,16 @@
 mod event;
 mod join;
 
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::VecDeque, sync::Arc};
 
 use conduwuit::{
-	Err, Event, Pdu, Result, Server, debug, debug_info, debug_warn, defer, 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, defer, info, is_true, pdu::PartialPdu, matrix::event::gen_event_id, 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::{
@@ -38,11 +24,13 @@ use ruma::{
 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,
 };
@@ -103,18 +91,19 @@ impl crate::Service for Service {
 }
 
 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
@@ -191,12 +180,12 @@ impl Service {
 		}
 
 		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?;
 		}
@@ -210,11 +199,11 @@ impl Service {
 		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),
@@ -269,6 +258,7 @@ impl Service {
 			info!("Joined room locally");
 			return Ok(());
 		};
+		drop(state_lock);
 
 		if servers.is_empty()
 			|| servers.len() == 1 && self.services.globals.server_is_ours(&servers[0])
@@ -291,7 +281,7 @@ impl Service {
 			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")]
@@ -300,104 +290,83 @@ impl Service {
 		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
@@ -411,401 +380,36 @@ impl Service {
 			.await;
 		self.services.event_handler.begin_remote_join(room_id);
 		defer! {{ self.services.event_handler.cancel_remote_join(room_id); }}
-		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
-			.get_events_keys(resp_auth.iter().chain(resp_state.iter()), &room_version_rules)
-			.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
-					.verify_event_json_no_fetch_add_event_id(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) => {
-						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();
-		let send_join_auth_chain = send_join_response
-			.room_state
-			.auth_chain
-			.iter()
-			.stream()
-			.then(|pdu| {
-				self.services
-					.server_keys
-					.verify_event_json_no_fetch_add_event_id(pdu, &room_version_rules)
-			})
-			.ready_filter_map(Result::ok)
-			.ready_fold(Vec::new(), |mut chain, (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);
-				let state_key = value
-					.get("state_key")
-					.and_then(|s| s.as_str())
-					.unwrap_or_default()
-					.to_owned();
-				let event_type = value
-					.get("type")
-					.and_then(|s| s.as_str())
-					.unwrap_or_default()
-					.to_owned();
-				chain.push((event_type, state_key));
-				chain
-			})
-			.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 Some(create_event) = state_fetch(StateEventType::RoomCreate, "".into()).await else {
-			let in_auth_chain = send_join_auth_chain
-				.iter()
-				.any(|(t, s)| t == "m.room.create" && s.is_empty());
-			return Err!(BadServerResponse(debug_error!(
-				?state,
-				"bad response from {remote_server}: m.room.create event is missing from \
-				 send_join state{}",
-				if in_auth_chain {
-					", but it was found in the auth_chain"
-				} else {
-					""
-				}
-			)));
-		};
-		drop(send_join_auth_chain);
-
-		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()),
-			&create_event,
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden(warn!("Auth check failed: {e:?}")))))?;
-
-		if !auth_check {
-			return Err!(Request(Forbidden(
-				"Join event returned by {remote_server} is not self-authorised (try again?)"
-			)));
-		}
-		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;
-		self.services
-			.event_handler
-			.process_pending_join_pdus(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)
+			info!("send_join finished");
+			drop(cork);
+			self.services.sync.wake_all_joined(room_id).await;
+			self.services
+				.event_handler
+				.process_pending_join_pdus(room_id)
 				.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.");
-					},
-				},
-			}
+			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/users/account.rs

@@ -304,7 +304,7 @@ impl super::Service {
 						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(),
 						],