fix: Address review comments

src/api/client/message.rs

@@ -380,5 +380,6 @@ pub(crate) async fn is_ignored_invite(
 	services
 		.users
 		.invite_filter_level(&sender_user, recipient_user)
-		.await == FilterLevel::Ignore
+		.await
+		== FilterLevel::Ignore
 }

src/api/client/room/upgrade.rs

@@ -305,8 +305,8 @@ pub(crate) async fn upgrade_room_route(
 		return Err!(Request(Forbidden("Upgrading the admin room this way is not allowed.")));
 	}
 
-	// 1. Check that the user has permission to send m.room.tombstone events in the
-	//    room.
+	// 1. Check that the user has permission to send m.room.tombstone events in
+	//    the room.
 	let old_room_state_lock = services.rooms.state.mutex.lock(body.room_id.as_str()).await;
 
 	// Check tombstone permission by attempting to create (but not send) the
@@ -556,8 +556,8 @@ pub(crate) async fn upgrade_room_route(
 		)?;
 	}
 
-	// 5. Send a `m.room.tombstone` event to the old room to indicate that it is not
-	//    intended to be used any further.
+	// 5. Send a `m.room.tombstone` event to the old room to indicate that it is
+	//    not intended to be used any further.
 	debug!(target=?body.room_id, "Sending tombstone to old room");
 	services
 		.rooms
@@ -598,7 +598,8 @@ pub(crate) async fn upgrade_room_route(
 	power_levels.events_default = new_level;
 	power_levels.invite = new_level;
 
-	// 6. Modify the power levels in the old room to prevent sending of events and
+	// 6. Modify the power levels in the old room to prevent sending of events
+	//    and
 	// inviting new users
 	// Spec dictates that this is allowed to fail.
 	debug!(target=?body.room_id, ?new_level, "Raising power level in old room to lock it");

src/api/client/sync/v3/left.rs

@@ -74,8 +74,8 @@ pub(super) async fn load_left_room(
 
 	// return early if:
 	// - this is an initial sync and the room filter doesn't include leaves, or
-	// - this is an incremental sync, and we've already synced the leave, and the
-	//   room filter doesn't include leaves
+	// - this is an incremental sync, and we've already synced the leave, and
+	//   the room filter doesn't include leaves
 	if last_sync_end_count.is_none_or(|last_sync_end_count| last_sync_end_count >= left_count)
 		&& !filter.room.include_leave
 	{

src/api/client/sync/v5.rs

@@ -726,7 +726,8 @@ async fn stripped_state_event(
 				.get_field::<StateEventType>("type")
 				.ok()
 				.flatten()
-				.as_ref() == Some(event_type)
+				.as_ref()
+				== Some(event_type)
 		})
 }
 

src/api/server/key.rs

@@ -143,6 +143,7 @@ async fn acquire_keys_as_notary(
 	mut queries: BTreeMap<OwnedServerSigningKeyId, QueryCriteria>,
 	start: MilliSecondsSinceUnixEpoch,
 ) -> Vec<Raw<ServerSigningKeys>> {
+	let fetch_all = queries.is_empty();
 	let mut results = Vec::with_capacity(queries.len().max(1));
 	let mut keymap = HashMap::with_capacity(queries.len().max(1));
 
@@ -155,12 +156,11 @@ async fn acquire_keys_as_notary(
 			let index = results.len();
 			results.push(ssk);
 			for key_id in res.verify_keys.keys().chain(res.old_verify_keys.keys()) {
-				if queries.is_empty()
-					|| queries
-						.get(key_id)
-						.and_then(|c| c.minimum_valid_until_ts)
-						.is_none_or(|m| res.valid_until_ts >= m)
-				{
+				if fetch_all
+					|| queries.get(key_id).is_some_and(|c| {
+						c.minimum_valid_until_ts
+							.is_none_or(|m| res.valid_until_ts >= m)
+					}) {
 					queries.remove(key_id);
 					keymap.insert(key_id.to_owned(), index);
 				}
@@ -170,7 +170,7 @@ async fn acquire_keys_as_notary(
 		debug!("Not asking remote for keys (already asked recently)");
 	}
 	debug!(keys=?keymap.keys(), "Live verify keys");
-	if queries.is_empty() && keymap.is_empty() {
+	if fetch_all && keymap.is_empty() {
 		// If the server asked for all keys, AND we didn't get anything from the
 		// origin, just fetch any fresh responses we have.
 		server_keys
@@ -228,14 +228,20 @@ async fn acquire_keys_as_notary(
 			);
 			continue;
 		}
+		if let Some(old_verify_key) = ssk.old_verify_keys.get(&key_id) {
+			if old_verify_key.expired_ts < minimum_valid_until_ts {
+				debug!(
+					%key_id,
+					expired_ts=?old_verify_key.expired_ts,
+					?minimum_valid_until_ts,
+					"Stored old verify key does not satisfy query criteria"
+				);
+				continue;
+			}
+		}
 
 		debug!(%key_id, ?ssk, "Found key locally");
-		let rep_key_ids = ssk
-			.verify_keys
-			.keys()
-			.chain(ssk.old_verify_keys.keys())
-			.cloned()
-			.collect::<Vec<_>>();
+		let rep_key_ids = ssk.verify_keys.keys().cloned().collect::<Vec<_>>();
 		let Ok(signed_ssk) = sign_ssk(&server_keys, ssk.clone(), &remote, &my_name)
 			.await
 			.inspect_err(
@@ -247,6 +253,7 @@ async fn acquire_keys_as_notary(
 
 		let idx = results.len();
 		results.push(signed_ssk);
+		keymap.insert(key_id, idx);
 		for key_id in rep_key_ids {
 			keymap.insert(key_id, idx);
 		}

src/core/matrix/state_res/event_auth.rs

@@ -763,8 +763,8 @@ where
 	Ok(match target_membership {
 		| MembershipState::Join => {
 			trace!("starting target_membership=join check");
-			// 1. If the only previous event is an m.room.create and the state_key is the
-			//    creator,
+			// 1. If the only previous event is an m.room.create and the
+			//    state_key is the creator,
 			// allow
 			let mut prev_events = current_event.prev_events();
 
@@ -1052,8 +1052,8 @@ where
 				allow
 			},
 		| MembershipState::Knock if room_version.authorization.knocking => {
-			// 1. If the `join_rule` is anything other than `knock` or `knock_restricted`,
-			//    reject.
+			// 1. If the `join_rule` is anything other than `knock` or
+			//    `knock_restricted`, reject.
 			if !matches!(join_rules, JoinRule::KnockRestricted(_) | JoinRule::Knock) {
 				warn!(
 					"Join rule is not set to knock or knock_restricted, knocking is not allowed"
@@ -1062,8 +1062,8 @@ where
 			} else if matches!(join_rules, JoinRule::KnockRestricted(_))
 				&& !room_version.authorization.knock_restricted_join_rule
 			{
-				// 2. If the `join_rule` is `knock_restricted`, but the room does not support
-				//    `knock_restricted`, reject.
+				// 2. If the `join_rule` is `knock_restricted`, but the room
+				//    does not support `knock_restricted`, reject.
 				warn!(
 					"Join rule is set to knock_restricted but room version does not support \
 					 knock_restricted, knocking is not allowed"
@@ -1081,8 +1081,8 @@ where
 				sender_membership,
 				MembershipState::Ban | MembershipState::Invite | MembershipState::Join
 			) {
-				// 4. If the `sender`'s current membership is not `ban`, `invite`, or `join`,
-				//    allow.
+				// 4. If the `sender`'s current membership is not `ban`,
+				//    `invite`, or `join`, allow.
 				// 5. Otherwise, reject.
 				warn!(
 					?target_user_membership_event_id,
@@ -1173,11 +1173,12 @@ fn check_power_levels(
 	}
 
 	// - If any of the keys users_default, events_default, state_default, ban,
-	//   redact, kick, or invite in content are present and not an integer, reject.
-	// - If either of the keys events or notifications in content are present and
-	//   not a dictionary with values that are integers, reject.
-	// - If users key in content is not a dictionary with keys that are valid user
-	//   IDs with values that are integers, reject.
+	//   redact, kick, or invite in content are present and not an integer,
+	//   reject.
+	// - If either of the keys events or notifications in content are present
+	//   and not a dictionary with values that are integers, reject.
+	// - If users key in content is not a dictionary with keys that are valid
+	//   user IDs with values that are integers, reject.
 	let user_content: RoomPowerLevelsEventContent =
 		deserialize_power_levels(power_event.content().get(), room_version)?;
 

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -161,8 +161,8 @@ impl super::Service {
 			}
 		}
 
-		// 4. Reject "due to auth events" if the event doesn't pass auth based on the
-		//    claimed auth events
+		// 4. Reject "due to auth events" if the event doesn't pass auth based
+		//    on the claimed auth events
 		debug!("Checking based on auth events");
 		let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len());
 		// Build map of auth events

src/service/rooms/summary/mod.rs

@@ -553,11 +553,12 @@ impl Service {
 				.services
 				.state_cache
 				.is_joined(querying_user, &summary.room_id)
-				.await || self
-				.services
-				.state_cache
-				.is_invited(querying_user, &summary.room_id)
-				.await)
+				.await
+				|| self
+					.services
+					.state_cache
+					.is_invited(querying_user, &summary.room_id)
+					.await)
 		{
 			return true;
 		}

src/service/sending/mod.rs

@@ -289,7 +289,8 @@ impl Service {
 					.services
 					.state_cache
 					.appservice_in_room(room_id, &appservice)
-					.await || matching_aliases(appservice.aliases.clone()).await
+					.await
+				|| matching_aliases(appservice.aliases.clone()).await
 			{
 				_ = self
 					.send_edu_appservice(&appservice.registration.id, serialized.clone())

src/service/server_keys/mod.rs

@@ -12,6 +12,7 @@ use std::{
 };
 
 use assign::assign;
+use async_trait::async_trait;
 use conduwuit::{
 	Result, Server, SyncRwLock,
 	utils::{IterStream, ReadyExt, stream::TryIgnore, timepoint_from_now, to_canonical_object},
@@ -56,6 +57,7 @@ pub type VerifyKeys = BTreeMap<OwnedServerSigningKeyId, VerifyKey>;
 pub type PubKeyMap = PublicKeyMap;
 pub type PubKeys = PublicKeySet;
 
+#[async_trait]
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		let (keypair, verify_keys) = keypair::init(args.db)?;
@@ -76,6 +78,8 @@ impl crate::Service for Service {
 		}))
 	}
 
+	async fn clear_cache(&self) { self.last_lookup.write().clear(); }
+
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
@@ -214,8 +218,9 @@ impl Service {
 	#[must_use]
 	pub fn notary_may_contact_origin(&self, server_name: &ServerName) -> bool {
 		if server_name == self.services.server.name {
-			// We always intercept origin requests to us with build_server_keys_response,
-			// so this is equivalent to calling `/_matrix/key/v2/server`.
+			// We always intercept origin requests to us with
+			// build_server_keys_response, so this is equivalent to
+			// calling `/_matrix/key/v2/server`.
 			return true;
 		}
 		self.last_lookup

src/service/transactions/mod.rs

@@ -198,7 +198,8 @@ impl Service {
 			&& let active_count = state
 				.values()
 				.filter(|v| matches!(v, TxnState::Active(_)))
-				.count() && active_count >= max_active_txns
+				.count()
+			&& active_count >= max_active_txns
 		{
 			warn!(
 				active = active_count,

src/service/users/account.rs

@@ -409,7 +409,8 @@ impl super::Service {
 				.services
 				.appservice
 				.is_exclusive_user_id(&user_id)
-				.await && !emergency_mode_enabled
+				.await
+				&& !emergency_mode_enabled
 			{
 				return Err!(Request(Exclusive("Username is reserved by an appservice.")));
 			}

src/service/users/profile.rs

@@ -81,7 +81,8 @@ impl super::Service {
 			if let Ok(canonical_profile) = to_canonical_object(full_profile) {
 				if serde_json::to_string(&canonical_profile)
 					.expect("should be able to serialize to string")
-					.len() > MAX_PROFILE_LENGTH_BYTES
+					.len()
+					> MAX_PROFILE_LENGTH_BYTES
 				{
 					return Err!(
 						"Profile data must not exceed {MAX_PROFILE_LENGTH_BYTES} bytes in \