style: Document & refactor server_keys service

src/service/server_keys/acquire.rs

@@ -4,9 +4,7 @@ use std::{
 	time::Duration,
 };
 
-use conduwuit::{
-	debug, debug_error, debug_warn, error, implement, info, result::FlatOk, trace, warn,
-};
+use conduwuit::{debug, debug_error, debug_warn, error, info, result::FlatOk, trace, warn};
 use futures::{StreamExt, stream::FuturesUnordered};
 use ruma::{
 	CanonicalJsonObject, OwnedServerName, OwnedServerSigningKeyId, ServerName,
@@ -19,218 +17,228 @@ use super::key_exists;
 
 type Batch = BTreeMap<OwnedServerName, Vec<OwnedServerSigningKeyId>>;
 
-#[implement(super::Service)]
-pub async fn acquire_events_pubkeys<'a, I>(&self, events: I)
-where
-	I: Iterator<Item = &'a Box<RawJsonValue>> + Send,
-{
-	type Batch = BTreeMap<OwnedServerName, BTreeSet<OwnedServerSigningKeyId>>;
-	type Signatures = BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, String>>;
+impl super::Service {
+	/// Returns all verify keys required to verify the incoming events.
+	pub async fn acquire_events_pubkeys<'a, I>(&self, events: I)
+	where
+		I: Iterator<Item = &'a Box<RawJsonValue>> + Send,
+	{
+		type Batch = BTreeMap<OwnedServerName, BTreeSet<OwnedServerSigningKeyId>>;
+		type Signatures = BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, String>>;
 
-	let mut batch = Batch::new();
-	events
-		.cloned()
-		.map(Raw::<CanonicalJsonObject>::from_json)
-		.map(|event| event.get_field::<Signatures>("signatures"))
-		.filter_map(FlatOk::flat_ok)
-		.flat_map(IntoIterator::into_iter)
-		.for_each(|(server, sigs)| {
-			batch.entry(server).or_default().extend(sigs.into_keys());
-		});
+		let mut batch = Batch::new();
+		events
+			.cloned()
+			.map(Raw::<CanonicalJsonObject>::from_json)
+			.map(|event| event.get_field::<Signatures>("signatures"))
+			.filter_map(FlatOk::flat_ok)
+			.flat_map(IntoIterator::into_iter)
+			.for_each(|(server, sigs)| {
+				batch.entry(server).or_default().extend(sigs.into_keys());
+			});
 
-	let batch = batch
-		.iter()
-		.map(|(server, keys)| (server.borrow(), keys.iter().map(Borrow::borrow)));
-
-	self.acquire_pubkeys(batch).await;
-}
-
-#[implement(super::Service)]
-pub async fn acquire_pubkeys<'a, S, K>(&self, batch: S)
-where
-	S: Iterator<Item = (&'a ServerName, K)> + Send + Clone,
-	K: Iterator<Item = &'a ServerSigningKeyId> + Send + Clone,
-{
-	let notary_only = self.services.server.config.only_query_trusted_key_servers;
-	let notary_first_always = self.services.server.config.query_trusted_key_servers_first;
-	let notary_first_on_join = self
-		.services
-		.server
-		.config
-		.query_trusted_key_servers_first_on_join;
-
-	let requested_servers = batch.clone().count();
-	let requested_keys = batch.clone().flat_map(|(_, key_ids)| key_ids).count();
-
-	debug!("acquire {requested_keys} keys from {requested_servers}");
-
-	let mut missing = self.acquire_locals(batch).await;
-	let mut missing_keys = keys_count(&missing);
-	let mut missing_servers = missing.len();
-	if missing_servers == 0 {
-		return;
-	}
-
-	info!("{missing_keys} keys for {missing_servers} servers will be acquired");
-
-	if notary_first_always || notary_first_on_join {
-		missing = self.acquire_notary(missing.into_iter()).await;
-		missing_keys = keys_count(&missing);
-		missing_servers = missing.len();
-		if missing_keys == 0 {
-			return;
-		}
-
-		warn!(
-			"missing {missing_keys} keys for {missing_servers} servers from all notaries first"
-		);
-	}
-
-	if !notary_only {
-		missing = self.acquire_origins(missing.into_iter()).await;
-		missing_keys = keys_count(&missing);
-		missing_servers = missing.len();
-		if missing_keys == 0 {
-			return;
-		}
-
-		debug_warn!("missing {missing_keys} keys for {missing_servers} servers unreachable");
-	}
-
-	if !notary_first_always && !notary_first_on_join {
-		missing = self.acquire_notary(missing.into_iter()).await;
-		missing_keys = keys_count(&missing);
-		missing_servers = missing.len();
-		if missing_keys == 0 {
-			return;
-		}
-
-		debug_warn!(
-			"still missing {missing_keys} keys for {missing_servers} servers from all notaries."
-		);
-	}
-
-	if missing_keys > 0 {
-		warn!(
-			"did not obtain {missing_keys} keys for {missing_servers} servers out of \
-			 {requested_keys} total keys for {requested_servers} total servers."
-		);
-	}
-
-	for (server, key_ids) in missing {
-		debug_warn!(%server, ?key_ids, "missing");
-	}
-}
-
-#[implement(super::Service)]
-async fn acquire_locals<'a, S, K>(&self, batch: S) -> Batch
-where
-	S: Iterator<Item = (&'a ServerName, K)> + Send,
-	K: Iterator<Item = &'a ServerSigningKeyId> + Send,
-{
-	let mut missing = Batch::new();
-	for (server, key_ids) in batch {
-		for key_id in key_ids {
-			if !self.verify_key_exists(server, key_id).await {
-				missing
-					.entry(server.into())
-					.or_default()
-					.push(key_id.into());
-			}
-		}
-	}
-
-	missing
-}
-
-#[implement(super::Service)]
-async fn acquire_origins<I>(&self, batch: I) -> Batch
-where
-	I: Iterator<Item = (OwnedServerName, Vec<OwnedServerSigningKeyId>)> + Send,
-{
-	let timeout = Instant::now()
-		.checked_add(Duration::from_secs(45))
-		.expect("timeout overflows");
-
-	let mut requests: FuturesUnordered<_> = batch
-		.map(|(origin, key_ids)| self.acquire_origin(origin, key_ids, timeout))
-		.collect();
-
-	let mut missing = Batch::new();
-	while let Some((origin, key_ids)) = requests.next().await {
-		if !key_ids.is_empty() {
-			missing.insert(origin, key_ids);
-		}
-	}
-
-	missing
-}
-
-#[implement(super::Service)]
-async fn acquire_origin(
-	&self,
-	origin: OwnedServerName,
-	mut key_ids: Vec<OwnedServerSigningKeyId>,
-	timeout: Instant,
-) -> (OwnedServerName, Vec<OwnedServerSigningKeyId>) {
-	match timeout_at(timeout, self.server_request(&origin)).await {
-		| Err(e) => debug_warn!(%origin, "timed out: {e}"),
-		| Ok(Err(e)) => debug_error!(%origin, "{e}"),
-		| Ok(Ok(server_keys)) => {
-			trace!(
-				%origin,
-				?key_ids,
-				?server_keys,
-				"received server_keys"
-			);
-
-			self.add_signing_keys(server_keys.clone()).await;
-			key_ids.retain(|key_id| !key_exists(&server_keys, key_id));
-		},
-	}
-
-	(origin, key_ids)
-}
-
-#[implement(super::Service)]
-async fn acquire_notary<I>(&self, batch: I) -> Batch
-where
-	I: Iterator<Item = (OwnedServerName, Vec<OwnedServerSigningKeyId>)> + Send,
-{
-	let mut missing: Batch = batch.collect();
-	for notary in self.services.globals.trusted_servers() {
-		let missing_keys = keys_count(&missing);
-		let missing_servers = missing.len();
-		debug!(
-			"Asking notary {notary} for {missing_keys} missing keys from {missing_servers} \
-			 servers"
-		);
-
-		let batch = missing
+		let batch = batch
 			.iter()
 			.map(|(server, keys)| (server.borrow(), keys.iter().map(Borrow::borrow)));
 
-		match self.batch_notary_request(notary, batch).await {
-			| Err(e) => error!("Failed to contact notary {notary:?}: {e}"),
-			| Ok(results) =>
-				for server_keys in results {
-					self.acquire_notary_result(&mut missing, server_keys).await;
-				},
+		self.acquire_pubkeys(batch).await;
+	}
+
+	/// Acquires all the verify keys required in the provided batch, using
+	/// notaries and direct-fetch where applicable.
+	pub async fn acquire_pubkeys<'a, S, K>(&self, batch: S)
+	where
+		S: Iterator<Item = (&'a ServerName, K)> + Send + Clone,
+		K: Iterator<Item = &'a ServerSigningKeyId> + Send + Clone,
+	{
+		let notary_only = self.services.server.config.only_query_trusted_key_servers;
+		let notary_first_always = self.services.server.config.query_trusted_key_servers_first;
+		let notary_first_on_join = self
+			.services
+			.server
+			.config
+			.query_trusted_key_servers_first_on_join;
+
+		let requested_servers = batch.clone().count();
+		let requested_keys = batch.clone().flat_map(|(_, key_ids)| key_ids).count();
+
+		debug!("acquire {requested_keys} keys from {requested_servers}");
+
+		let mut missing = self.acquire_locals(batch).await;
+		let mut missing_keys = keys_count(&missing);
+		let mut missing_servers = missing.len();
+		if missing_servers == 0 {
+			return;
+		}
+
+		info!("{missing_keys} keys for {missing_servers} servers will be acquired");
+
+		if notary_first_always || notary_first_on_join {
+			missing = self.acquire_notary(missing.into_iter()).await;
+			missing_keys = keys_count(&missing);
+			missing_servers = missing.len();
+			if missing_keys == 0 {
+				return;
+			}
+
+			warn!(
+				"missing {missing_keys} keys for {missing_servers} servers from all notaries \
+				 first"
+			);
+		}
+
+		if !notary_only {
+			missing = self.acquire_origins(missing.into_iter()).await;
+			missing_keys = keys_count(&missing);
+			missing_servers = missing.len();
+			if missing_keys == 0 {
+				return;
+			}
+
+			debug_warn!("missing {missing_keys} keys for {missing_servers} servers unreachable");
+		}
+
+		if !notary_first_always && !notary_first_on_join {
+			missing = self.acquire_notary(missing.into_iter()).await;
+			missing_keys = keys_count(&missing);
+			missing_servers = missing.len();
+			if missing_keys == 0 {
+				return;
+			}
+
+			debug_warn!(
+				"still missing {missing_keys} keys for {missing_servers} servers from all \
+				 notaries."
+			);
+		}
+
+		if missing_keys > 0 {
+			warn!(
+				"did not obtain {missing_keys} keys for {missing_servers} servers out of \
+				 {requested_keys} total keys for {requested_servers} total servers."
+			);
+		}
+
+		for (server, key_ids) in missing {
+			debug_warn!(%server, ?key_ids, "missing");
 		}
 	}
 
-	missing
-}
+	/// Filters the batch to return only keys which we do not already have
+	/// locally.
+	async fn acquire_locals<'a, S, K>(&self, batch: S) -> Batch
+	where
+		S: Iterator<Item = (&'a ServerName, K)> + Send,
+		K: Iterator<Item = &'a ServerSigningKeyId> + Send,
+	{
+		let mut missing = Batch::new();
+		for (server, key_ids) in batch {
+			for key_id in key_ids {
+				if !self.verify_key_exists(server, key_id).await {
+					missing
+						.entry(server.into())
+						.or_default()
+						.push(key_id.into());
+				}
+			}
+		}
 
-#[implement(super::Service)]
-async fn acquire_notary_result(&self, missing: &mut Batch, server_keys: ServerSigningKeys) {
-	let server = &server_keys.server_name;
-	self.add_signing_keys(server_keys.clone()).await;
+		missing
+	}
 
-	if let Some(key_ids) = missing.get_mut(server) {
-		key_ids.retain(|key_id| !key_exists(&server_keys, key_id));
-		if key_ids.is_empty() {
-			missing.remove(server);
+	/// Fetches verification keys from the origin servers directly, returning
+	/// any keys that could not be fetched.
+	async fn acquire_origins<I>(&self, batch: I) -> Batch
+	where
+		I: Iterator<Item = (OwnedServerName, Vec<OwnedServerSigningKeyId>)> + Send,
+	{
+		let timeout = Instant::now()
+			.checked_add(Duration::from_secs(45))
+			.expect("timeout overflows");
+
+		let mut requests: FuturesUnordered<_> = batch
+			.map(|(origin, key_ids)| self.acquire_origin(origin, key_ids, timeout))
+			.collect();
+
+		let mut missing = Batch::new();
+		while let Some((origin, key_ids)) = requests.next().await {
+			if !key_ids.is_empty() {
+				missing.insert(origin, key_ids);
+			}
+		}
+
+		missing
+	}
+
+	/// Fetches verification keys from an origin directly, returning any keys
+	/// that were fetched.
+	async fn acquire_origin(
+		&self,
+		origin: OwnedServerName,
+		mut key_ids: Vec<OwnedServerSigningKeyId>,
+		timeout: Instant,
+	) -> (OwnedServerName, Vec<OwnedServerSigningKeyId>) {
+		match timeout_at(timeout, self.server_request(&origin)).await {
+			| Err(e) => debug_warn!(%origin, "timed out: {e}"),
+			| Ok(Err(e)) => debug_error!(%origin, "{e}"),
+			| Ok(Ok(server_keys)) => {
+				trace!(
+					%origin,
+					?key_ids,
+					?server_keys,
+					"received server_keys"
+				);
+
+				self.add_signing_keys(server_keys.clone()).await;
+				key_ids.retain(|key_id| !key_exists(&server_keys, key_id));
+			},
+		}
+
+		(origin, key_ids)
+	}
+
+	/// Fetches signing keys for the given batch from the configured notary
+	/// servers, returning any keys that could not be fetched.
+	async fn acquire_notary<I>(&self, batch: I) -> Batch
+	where
+		I: Iterator<Item = (OwnedServerName, Vec<OwnedServerSigningKeyId>)> + Send,
+	{
+		let mut missing: Batch = batch.collect();
+		for notary in self.services.globals.trusted_servers() {
+			let missing_keys = keys_count(&missing);
+			let missing_servers = missing.len();
+			debug!(
+				"Asking notary {notary} for {missing_keys} missing keys from {missing_servers} \
+				 servers"
+			);
+
+			let batch = missing
+				.iter()
+				.map(|(server, keys)| (server.borrow(), keys.iter().map(Borrow::borrow)));
+
+			match self.batch_notary_request(notary, batch).await {
+				| Err(e) => error!("Failed to contact notary {notary:?}: {e}"),
+				| Ok(results) =>
+					for server_keys in results {
+						self.acquire_notary_result(&mut missing, server_keys).await;
+					},
+			}
+		}
+
+		missing
+	}
+
+	/// Contacts a notary and fetches signing keys for the given batch, removing
+	/// keys from the missing batch when they are found.
+	async fn acquire_notary_result(&self, missing: &mut Batch, server_keys: ServerSigningKeys) {
+		let server = &server_keys.server_name;
+		self.add_signing_keys(server_keys.clone()).await;
+
+		if let Some(key_ids) = missing.get_mut(server) {
+			key_ids.retain(|key_id| !key_exists(&server_keys, key_id));
+			if key_ids.is_empty() {
+				missing.remove(server);
+			}
 		}
 	}
 }

src/service/server_keys/get.rs

@@ -1,6 +1,6 @@
 use std::borrow::Borrow;
 
-use conduwuit::{Err, Result, debug_error, implement, trace};
+use conduwuit::{Err, Result, trace};
 use ruma::{
 	CanonicalJsonObject, ServerName, ServerSigningKeyId, api::federation::discovery::VerifyKey,
 	room_version_rules::RoomVersionRules,
@@ -9,133 +9,138 @@ use ruma::{
 use super::{PubKeyMap, PubKeys, extract_key};
 use crate::server_keys::util::required_keys;
 
-#[implement(super::Service)]
-pub async fn get_event_keys(
-	&self,
-	object: &CanonicalJsonObject,
-	room_version_rules: &RoomVersionRules,
-) -> Result<PubKeyMap> {
-	let required = match required_keys(object, &room_version_rules.signatures) {
-		| Ok(required) => required,
-		| Err(e) => {
-			debug_error!("Failed to determine keys required to verify: {e}");
-			return Err!(BadServerResponse("Failed to determine keys required to verify: {e}"));
-		},
-	};
-	trace!(?required, "Keys required to verify event");
+impl super::Service {
+	/// Fetches the verify keys required to verify the incoming event.
+	pub async fn get_event_keys(
+		&self,
+		object: &CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<PubKeyMap> {
+		let required = match required_keys(object, &room_version_rules.signatures) {
+			| Ok(required) => required,
+			| Err(e) => {
+				return Err!(BadServerResponse(
+					"Failed to determine keys required to verify event: {e}"
+				));
+			},
+		};
 
-	let batch = required
-		.iter()
-		.map(|(s, ids)| (s.borrow(), ids.iter().map(Borrow::borrow)));
+		let batch = required
+			.iter()
+			.map(|(s, ids)| (s.borrow(), ids.iter().map(Borrow::borrow)));
 
-	Ok(self.get_pubkeys(batch).await)
-}
-
-#[implement(super::Service)]
-pub async fn get_pubkeys<'a, S, K>(&self, batch: S) -> PubKeyMap
-where
-	S: Iterator<Item = (&'a ServerName, K)> + Send,
-	K: Iterator<Item = &'a ServerSigningKeyId> + Send,
-{
-	let mut keys = PubKeyMap::new();
-	for (server, key_ids) in batch {
-		let pubkeys = self.get_pubkeys_for(server, key_ids).await;
-		keys.insert(server.into(), pubkeys);
+		Ok(self.get_pubkeys(batch).await)
 	}
 
-	keys
-}
-
-#[implement(super::Service)]
-pub async fn get_pubkeys_for<'a, I>(&self, origin: &ServerName, key_ids: I) -> PubKeys
-where
-	I: Iterator<Item = &'a ServerSigningKeyId> + Send,
-{
-	let mut keys = PubKeys::new();
-	for key_id in key_ids {
-		if let Ok(verify_key) = self.get_verify_key(origin, key_id).await {
-			keys.insert(key_id.into(), verify_key.key);
+	/// Fetches all verify keys for the given batch. Any keys which could not be
+	/// fetched are elided from the resulting map.
+	pub async fn get_pubkeys<'a, S, K>(&self, batch: S) -> PubKeyMap
+	where
+		S: Iterator<Item = (&'a ServerName, K)> + Send,
+		K: Iterator<Item = &'a ServerSigningKeyId> + Send,
+	{
+		let mut keys = PubKeyMap::new();
+		for (server, key_ids) in batch {
+			let pubkeys = self.get_pubkeys_for(server, key_ids).await;
+			keys.insert(server.into(), pubkeys);
 		}
+
+		keys
 	}
 
-	keys
-}
-
-#[implement(super::Service)]
-#[tracing::instrument(skip(self), level = "debug")]
-pub async fn get_verify_key(
-	&self,
-	origin: &ServerName,
-	key_id: &ServerSigningKeyId,
-) -> Result<VerifyKey> {
-	let notary_first = self.services.server.config.query_trusted_key_servers_first;
-	let notary_only = self.services.server.config.only_query_trusted_key_servers;
-
-	if let Some(result) = self.verify_keys_for(origin).await.remove(key_id) {
-		trace!("Found key in cache");
-		return Ok(result);
-	}
-
-	if notary_first {
-		if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await {
-			return Ok(result);
-		}
-	}
-
-	if !notary_only {
-		if let Ok(result) = self.get_verify_key_from_origin(origin, key_id).await {
-			return Ok(result);
-		}
-	}
-
-	if !notary_first {
-		if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await {
-			return Ok(result);
-		}
-	}
-
-	Err!(BadServerResponse(debug_error!(
-		%key_id,
-		%origin,
-		"Failed to fetch federation signing-key"
-	)))
-}
-
-#[implement(super::Service)]
-async fn get_verify_key_from_notaries(
-	&self,
-	origin: &ServerName,
-	key_id: &ServerSigningKeyId,
-) -> Result<VerifyKey> {
-	for notary in self.services.globals.trusted_servers() {
-		if let Ok(server_keys) = self.notary_request(notary, origin).await {
-			for server_key in server_keys.clone() {
-				self.add_signing_keys(server_key).await;
+	/// Gets verify keys for the given server name and a set of key IDs. Any
+	/// keys that cannot be found are elided from the resulting set.
+	pub async fn get_pubkeys_for<'a, I>(&self, origin: &ServerName, key_ids: I) -> PubKeys
+	where
+		I: Iterator<Item = &'a ServerSigningKeyId> + Send,
+	{
+		let mut keys = PubKeys::new();
+		for key_id in key_ids {
+			if let Ok(verify_key) = self.get_verify_key(origin, key_id).await {
+				keys.insert(key_id.into(), verify_key.key);
 			}
+		}
 
-			for server_key in server_keys {
-				if let Some(result) = extract_key(server_key, key_id) {
-					return Ok(result);
+		keys
+	}
+
+	/// Fetches a single verify key via origin and notaries.
+	pub async fn get_verify_key(
+		&self,
+		origin: &ServerName,
+		key_id: &ServerSigningKeyId,
+	) -> Result<VerifyKey> {
+		let notary_first = self.services.server.config.query_trusted_key_servers_first;
+		let notary_only = self.services.server.config.only_query_trusted_key_servers;
+
+		if let Some(result) = self.verify_keys_for(origin).await.remove(key_id) {
+			trace!("Found key in cache");
+			return Ok(result);
+		}
+
+		if notary_first {
+			if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await {
+				return Ok(result);
+			}
+		}
+
+		if !notary_only {
+			if let Ok(result) = self.get_verify_key_from_origin(origin, key_id).await {
+				return Ok(result);
+			}
+		}
+
+		if !notary_first {
+			if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await {
+				return Ok(result);
+			}
+		}
+
+		Err!(BadServerResponse(debug_error!(
+			%key_id,
+			%origin,
+			"Failed to fetch federation signing-key"
+		)))
+	}
+
+	/// Fetches verify keys from notary servers, adding them to the cache, and
+	/// returning the requested key if found.
+	async fn get_verify_key_from_notaries(
+		&self,
+		origin: &ServerName,
+		key_id: &ServerSigningKeyId,
+	) -> Result<VerifyKey> {
+		for notary in self.services.globals.trusted_servers() {
+			if let Ok(server_keys) = self.notary_request(notary, origin).await {
+				for server_key in server_keys.clone() {
+					self.add_signing_keys(server_key).await;
+				}
+
+				for server_key in server_keys {
+					if let Some(result) = extract_key(server_key, key_id) {
+						return Ok(result);
+					}
 				}
 			}
 		}
+
+		Err!(Request(NotFound("Failed to fetch signing-key from notaries")))
 	}
 
-	Err!(Request(NotFound("Failed to fetch signing-key from notaries")))
-}
-
-#[implement(super::Service)]
-async fn get_verify_key_from_origin(
-	&self,
-	origin: &ServerName,
-	key_id: &ServerSigningKeyId,
-) -> Result<VerifyKey> {
-	if let Ok(server_key) = self.server_request(origin).await {
-		self.add_signing_keys(server_key.clone()).await;
-		if let Some(result) = extract_key(server_key, key_id) {
-			return Ok(result);
+	/// Fetches verify keys from the origin server, adding them to the cache,
+	/// and returning the requested key if found.
+	async fn get_verify_key_from_origin(
+		&self,
+		origin: &ServerName,
+		key_id: &ServerSigningKeyId,
+	) -> Result<VerifyKey> {
+		if let Ok(server_key) = self.server_request(origin).await {
+			self.add_signing_keys(server_key.clone()).await;
+			if let Some(result) = extract_key(server_key, key_id) {
+				return Ok(result);
+			}
 		}
-	}
 
-	Err!(Request(NotFound("Failed to fetch signing-key from origin")))
+		Err!(Request(NotFound("Failed to fetch signing-key from origin")))
+	}
 }

src/service/server_keys/mod.rs

@@ -9,7 +9,7 @@ mod verify;
 use std::{collections::BTreeMap, sync::Arc, time::Duration};
 
 use conduwuit::{
-	Result, Server, debug_error, debug_warn, implement, trace,
+	Result, Server,
 	utils::{IterStream, timepoint_from_now},
 };
 use database::{Deserialized, Json, Map};
@@ -73,123 +73,124 @@ impl crate::Service for Service {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
-#[implement(Service)]
-#[inline]
-pub fn keypair(&self) -> &Ed25519KeyPair { &self.keypair }
+impl Service {
+	/// Fetches the server's active keypair.
+	pub fn keypair(&self) -> &Ed25519KeyPair { &self.keypair }
 
-#[implement(Service)]
-#[inline]
-pub fn active_key_id(&self) -> &ServerSigningKeyId { self.active_verify_key().0 }
+	/// Fetches the server's active signing key ID.
+	pub fn active_key_id(&self) -> &ServerSigningKeyId { self.active_verify_key().0 }
 
-#[implement(Service)]
-#[inline]
-pub fn active_verify_key(&self) -> (&ServerSigningKeyId, &VerifyKey) {
-	debug_assert!(self.verify_keys.len() <= 1, "more than one active verify_key");
-	self.verify_keys
-		.iter()
-		.next()
-		.map(|(id, key)| (id.as_ref(), key))
-		.expect("missing active verify_key")
-}
+	/// Fetches the server's active signing key. Panics if there's more than 1
+	/// active key.
+	pub fn active_verify_key(&self) -> (&ServerSigningKeyId, &VerifyKey) {
+		debug_assert!(self.verify_keys.len() <= 1, "more than one active verify_key");
+		self.verify_keys
+			.iter()
+			.next()
+			.map(|(id, key)| (id.as_ref(), key))
+			.expect("missing active verify_key")
+	}
 
-#[implement(Service)]
-async fn add_signing_keys(&self, new_keys: ServerSigningKeys) {
-	let origin = &new_keys.server_name;
+	/// Combines signing keys from the incoming response to the existing
+	/// database set.
+	async fn add_signing_keys(&self, new_keys: ServerSigningKeys) {
+		let origin = &new_keys.server_name;
 
-	// (timo) Not atomic, but this is not critical
-	let mut keys: ServerSigningKeys = self
-		.db
-		.server_signingkeys
-		.get(origin)
-		.await
-		.deserialized()
-		.unwrap_or_else(|_| {
-			// Just insert "now", it doesn't matter
-			ServerSigningKeys::new(origin.to_owned(), MilliSecondsSinceUnixEpoch::now())
-		});
+		// (timo) Not atomic, but this is not critical
+		let mut keys: ServerSigningKeys = self
+			.db
+			.server_signingkeys
+			.get(origin)
+			.await
+			.deserialized()
+			.unwrap_or_else(|_| {
+				// Just insert "now", it doesn't matter
+				ServerSigningKeys::new(origin.to_owned(), MilliSecondsSinceUnixEpoch::now())
+			});
 
-	keys.verify_keys.extend(new_keys.verify_keys);
-	keys.old_verify_keys.extend(new_keys.old_verify_keys);
-	self.db.server_signingkeys.raw_put(origin, Json(&keys));
-}
+		keys.verify_keys.extend(new_keys.verify_keys);
+		keys.old_verify_keys.extend(new_keys.old_verify_keys);
+		self.db.server_signingkeys.raw_put(origin, Json(&keys));
+	}
 
-#[implement(Service)]
-#[tracing::instrument(skip(self, object), level = "debug")]
-pub async fn required_keys_exist(
-	&self,
-	object: &CanonicalJsonObject,
-	room_version_rules: &RoomVersionRules,
-) -> bool {
-	trace!(?object, "Checking required keys exist");
-	let Ok(required_keys) = required_keys(object, &room_version_rules.signatures) else {
-		debug_error!("Failed to determine required keys");
-		return false;
-	};
-	trace!(?required_keys, "Required keys to verify event");
-	required_keys
-		.iter()
-		.flat_map(|(server, key_ids)| key_ids.iter().map(move |key_id| (server, key_id)))
-		.stream()
-		.all(|(server, key_id)| self.verify_key_exists(server, key_id))
-		.await
-}
+	/// Checks if the keys required to verify an incoming PDU are already known
+	/// locally.
+	pub async fn required_keys_exist(
+		&self,
+		object: &CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> bool {
+		let Ok(required_keys) = required_keys(object, &room_version_rules.signatures) else {
+			return false;
+		};
+		required_keys
+			.iter()
+			.flat_map(|(server, key_ids)| key_ids.iter().map(move |key_id| (server, key_id)))
+			.stream()
+			.all(|(server, key_id)| self.verify_key_exists(server, key_id))
+			.await
+	}
 
-#[implement(Service)]
-#[tracing::instrument(skip(self), level = "debug")]
-pub async fn verify_key_exists(&self, origin: &ServerName, key_id: &ServerSigningKeyId) -> bool {
-	type KeysMap = BTreeMap<OwnedServerSigningKeyId, Box<RawJsonValue>>;
+	/// Checks if a single verify key belonging to the target server is already
+	/// known locally.
+	pub async fn verify_key_exists(
+		&self,
+		origin: &ServerName,
+		key_id: &ServerSigningKeyId,
+	) -> bool {
+		type KeysMap = BTreeMap<OwnedServerSigningKeyId, Box<RawJsonValue>>;
 
-	let Ok(keys) = self
-		.db
-		.server_signingkeys
-		.get(origin)
-		.await
-		.deserialized::<Raw<ServerSigningKeys>>()
-	else {
-		debug_warn!("No known signing keys found for {origin}");
-		return false;
-	};
+		let Ok(keys) = self
+			.db
+			.server_signingkeys
+			.get(origin)
+			.await
+			.deserialized::<Raw<ServerSigningKeys>>()
+		else {
+			return false;
+		};
 
-	if let Ok(Some(verify_keys)) = keys.get_field::<KeysMap>("verify_keys") {
-		if verify_keys.contains_key(key_id) {
-			return true;
+		if let Ok(Some(verify_keys)) = keys.get_field::<KeysMap>("verify_keys") {
+			if verify_keys.contains_key(key_id) {
+				return true;
+			}
 		}
-	}
 
-	if let Ok(Some(old_verify_keys)) = keys.get_field::<KeysMap>("old_verify_keys") {
-		if old_verify_keys.contains_key(key_id) {
-			return true;
+		if let Ok(Some(old_verify_keys)) = keys.get_field::<KeysMap>("old_verify_keys") {
+			if old_verify_keys.contains_key(key_id) {
+				return true;
+			}
 		}
+
+		false
 	}
 
-	debug_warn!("Key {key_id} not found for {origin}");
-	false
-}
+	/// Returns all verify keys for the origin.
+	pub async fn verify_keys_for(&self, origin: &ServerName) -> VerifyKeys {
+		let mut keys = self
+			.signing_keys_for(origin)
+			.await
+			.map_or(BTreeMap::new(), |keys| merge_old_keys(keys).verify_keys);
 
-#[implement(Service)]
-pub async fn verify_keys_for(&self, origin: &ServerName) -> VerifyKeys {
-	let mut keys = self
-		.signing_keys_for(origin)
-		.await
-		.map_or(BTreeMap::new(), |keys| merge_old_keys(keys).verify_keys);
+		if self.services.globals.server_is_ours(origin) {
+			keys.extend(self.verify_keys.clone());
+		}
 
-	if self.services.globals.server_is_ours(origin) {
-		keys.extend(self.verify_keys.clone());
+		keys
 	}
 
-	keys
-}
+	/// Returns the stored server signing keys response for the origin.
+	pub async fn signing_keys_for(&self, origin: &ServerName) -> Result<ServerSigningKeys> {
+		self.db.server_signingkeys.get(origin).await.deserialized()
+	}
 
-#[implement(Service)]
-pub async fn signing_keys_for(&self, origin: &ServerName) -> Result<ServerSigningKeys> {
-	self.db.server_signingkeys.get(origin).await.deserialized()
-}
-
-#[implement(Service)]
-fn minimum_valid_ts(&self) -> MilliSecondsSinceUnixEpoch {
-	let timepoint =
-		timepoint_from_now(self.minimum_valid).expect("SystemTime should not overflow");
-	MilliSecondsSinceUnixEpoch::from_system_time(timepoint).expect("UInt should not overflow")
+	/// Returns the minimum_valid_ts for a key response from the current point
+	/// in time, plus the minimum valid time.
+	fn minimum_valid_ts(&self) -> MilliSecondsSinceUnixEpoch {
+		let timepoint =
+			timepoint_from_now(self.minimum_valid).expect("SystemTime should not overflow");
+		MilliSecondsSinceUnixEpoch::from_system_time(timepoint).expect("UInt should not overflow")
+	}
 }
 
 fn merge_old_keys(mut keys: ServerSigningKeys) -> ServerSigningKeys {

src/service/server_keys/request.rs

@@ -1,6 +1,7 @@
 use std::{collections::BTreeMap, fmt::Debug};
 
-use conduwuit::{Err, Result, debug, implement};
+use conduwuit::{Err, Result, debug};
+use get_remote_server_keys_batch::v2::Request;
 use ruma::{
 	OwnedServerName, OwnedServerSigningKeyId, ServerName, ServerSigningKeyId,
 	api::federation::discovery::{
@@ -9,51 +10,79 @@ use ruma::{
 		get_server_keys,
 	},
 };
+type RumaBatch = BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>;
 
-#[implement(super::Service)]
-pub(super) async fn batch_notary_request<'a, S, K>(
-	&self,
-	notary: &ServerName,
-	batch: S,
-) -> Result<Vec<ServerSigningKeys>>
-where
-	S: Iterator<Item = (&'a ServerName, K)> + Send,
-	K: Iterator<Item = &'a ServerSigningKeyId> + Send,
-{
-	use get_remote_server_keys_batch::v2::Request;
-	type RumaBatch = BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>;
-
-	let mut criteria = QueryCriteria::new();
-	criteria.minimum_valid_until_ts = Some(self.minimum_valid_ts());
-
-	let mut server_keys = batch.fold(RumaBatch::new(), |mut batch, (server, key_ids)| {
-		batch
-			.entry(server.into())
-			.or_default()
-			.extend(key_ids.map(|key_id| (key_id.into(), criteria.clone())));
-
-		batch
-	});
-
-	debug_assert!(!server_keys.is_empty(), "empty batch request to notary");
-
-	let mut results = Vec::new();
-	while let Some(batch) = server_keys
-		.keys()
-		.rev()
-		.take(self.services.server.config.trusted_server_batch_size)
-		.next_back()
-		.cloned()
+impl super::Service {
+	/// Handles a batched notary request.
+	/// If the batch is empty, an empty vec is returned.
+	pub(super) async fn batch_notary_request<'a, S, K>(
+		&self,
+		notary: &ServerName,
+		batch: S,
+	) -> Result<Vec<ServerSigningKeys>>
+	where
+		S: Iterator<Item = (&'a ServerName, K)> + Send,
+		K: Iterator<Item = &'a ServerSigningKeyId> + Send,
 	{
-		let request = Request::new(server_keys.split_off(&batch));
+		let mut criteria = QueryCriteria::new();
+		criteria.minimum_valid_until_ts = Some(self.minimum_valid_ts());
 
-		debug!(
-			?notary,
-			?batch,
-			remaining = %server_keys.len(),
-			requesting = ?request.server_keys.keys(),
-			"notary request"
-		);
+		let mut server_keys = batch.fold(RumaBatch::new(), |mut batch, (server, key_ids)| {
+			batch
+				.entry(server.into())
+				.or_default()
+				.extend(key_ids.map(|key_id| (key_id.into(), criteria.clone())));
+
+			batch
+		});
+
+		if server_keys.is_empty() {
+			return Ok(vec![]);
+		}
+
+		let mut results = Vec::new();
+		while let Some(batch) = server_keys
+			.keys()
+			.rev()
+			.take(self.services.server.config.trusted_server_batch_size)
+			.next_back()
+			.cloned()
+		{
+			let request = Request::new(server_keys.split_off(&batch));
+
+			debug!(
+				?notary,
+				?batch,
+				remaining = %server_keys.len(),
+				requesting = ?request.server_keys.keys(),
+				"notary request"
+			);
+
+			let response = self
+				.services
+				.sending
+				.send_unauthenticated_request(notary, request)
+				.await?
+				.server_keys
+				.into_iter()
+				.map(|key| key.deserialize())
+				.filter_map(Result::ok);
+
+			results.extend(response);
+		}
+
+		Ok(results)
+	}
+
+	/// Makes a request to a notary server to fetch signing keys.
+	pub async fn notary_request(
+		&self,
+		notary: &ServerName,
+		target: &ServerName,
+	) -> Result<impl Iterator<Item = ServerSigningKeys> + Clone + Debug + Send + use<>> {
+		use get_remote_server_keys::v2::Request;
+
+		let request = Request::new(target.into(), self.minimum_valid_ts());
 
 		let response = self
 			.services
@@ -65,54 +94,29 @@ where
 			.map(|key| key.deserialize())
 			.filter_map(Result::ok);
 
-		results.extend(response);
+		Ok(response)
 	}
 
-	Ok(results)
-}
+	/// Makes a request to an origin server to fetch signing keys.
+	pub async fn server_request(&self, target: &ServerName) -> Result<ServerSigningKeys> {
+		use get_server_keys::v2::Request;
 
-#[implement(super::Service)]
-pub async fn notary_request(
-	&self,
-	notary: &ServerName,
-	target: &ServerName,
-) -> Result<impl Iterator<Item = ServerSigningKeys> + Clone + Debug + Send + use<>> {
-	use get_remote_server_keys::v2::Request;
+		let server_signing_key = self
+			.services
+			.sending
+			.send_unauthenticated_request(target, Request::new())
+			.await
+			.map(|response| response.server_key)
+			.and_then(|key| key.deserialize().map_err(Into::into))?;
 
-	let request = Request::new(target.into(), self.minimum_valid_ts());
+		if server_signing_key.server_name != target {
+			return Err!(BadServerResponse(debug_warn!(
+				requested = ?target,
+				response = ?server_signing_key.server_name,
+				"Server responded with bogus server_name"
+			)));
+		}
 
-	let response = self
-		.services
-		.sending
-		.send_unauthenticated_request(notary, request)
-		.await?
-		.server_keys
-		.into_iter()
-		.map(|key| key.deserialize())
-		.filter_map(Result::ok);
-
-	Ok(response)
-}
-
-#[implement(super::Service)]
-pub async fn server_request(&self, target: &ServerName) -> Result<ServerSigningKeys> {
-	use get_server_keys::v2::Request;
-
-	let server_signing_key = self
-		.services
-		.sending
-		.send_unauthenticated_request(target, Request::new())
-		.await
-		.map(|response| response.server_key)
-		.and_then(|key| key.deserialize().map_err(Into::into))?;
-
-	if server_signing_key.server_name != target {
-		return Err!(BadServerResponse(debug_warn!(
-			requested = ?target,
-			response = ?server_signing_key.server_name,
-			"Server responded with bogus server_name"
-		)));
+		Ok(server_signing_key)
 	}
-
-	Ok(server_signing_key)
 }

src/service/server_keys/sign.rs

@@ -1,23 +1,29 @@
-use conduwuit::{Result, implement};
-use ruma::{CanonicalJsonObject, room_version_rules::RoomVersionRules};
+use conduwuit::Result;
+use ruma::{
+	CanonicalJsonObject, room_version_rules::RoomVersionRules, signatures::hash_and_sign_event,
+};
 
-#[implement(super::Service)]
-pub fn sign_json(&self, object: &mut CanonicalJsonObject) -> Result {
-	use ruma::signatures::sign_json;
+impl super::Service {
+	/// Signs an arbitrary JSON object. The object is mutated.
+	pub fn sign_json(&self, object: &mut CanonicalJsonObject) -> Result {
+		use ruma::signatures::sign_json;
 
-	let server_name = self.services.globals.server_name().as_str();
-	sign_json(server_name, self.keypair(), object).map_err(Into::into)
-}
+		let server_name = self.services.globals.server_name().as_str();
+		sign_json(server_name, self.keypair(), object).map_err(Into::into)
+	}
 
-#[implement(super::Service)]
-pub fn hash_and_sign_event(
-	&self,
-	object: &mut CanonicalJsonObject,
-	room_version_rules: &RoomVersionRules,
-) -> Result {
-	use ruma::signatures::hash_and_sign_event;
-
-	let server_name = self.services.globals.server_name().as_str();
-	hash_and_sign_event(server_name, self.keypair(), object, &room_version_rules.redaction)
+	/// Hashes and signs an event JSON object. The object is mutated.
+	pub fn hash_and_sign_event(
+		&self,
+		object: &mut CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result {
+		hash_and_sign_event(
+			self.services.globals.server_name().as_str(),
+			self.keypair(),
+			object,
+			&room_version_rules.redaction,
+		)
 		.map_err(Into::into)
+	}
 }

src/service/server_keys/verify.rs

@@ -1,77 +1,73 @@
-use conduwuit::{
-	Err, Result, debug_warn, implement, matrix::event::gen_event_id_canonical_json, trace,
-};
+use conduwuit::{Err, Result, matrix::event::gen_event_id_canonical_json};
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, room_version_rules::RoomVersionRules,
 	signatures::Verified,
 };
 use serde_json::value::RawValue as RawJsonValue;
 
-#[implement(super::Service)]
-pub async fn validate_and_add_event_id(
-	&self,
-	pdu: &RawJsonValue,
-	room_version_rules: &RoomVersionRules,
-) -> Result<(OwnedEventId, CanonicalJsonObject)> {
-	let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?;
-	if let Err(e) = self.verify_event(&value, room_version_rules).await {
-		return Err!(BadServerResponse(debug_error!(
-			"Event {event_id} failed verification: {e:?}"
-		)));
+impl super::Service {
+	/// Validates the incoming event, and then inserts the calculated event ID
+	/// into the event_id field.
+	pub async fn validate_and_add_event_id(
+		&self,
+		pdu: &RawJsonValue,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<(OwnedEventId, CanonicalJsonObject)> {
+		let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?;
+		if let Err(e) = self.verify_event(&value, room_version_rules).await {
+			return Err!(BadServerResponse(debug_error!(
+				"Event {event_id} failed verification: {e:?}"
+			)));
+		}
+
+		value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into()));
+
+		Ok((event_id, value))
 	}
 
-	value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into()));
+	/// Validates the incoming event only using locally known verification keys.
+	/// Inserts the calculated event ID into the event_id field upon success.
+	pub async fn validate_and_add_event_id_no_fetch(
+		&self,
+		pdu: &RawJsonValue,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<(OwnedEventId, CanonicalJsonObject)> {
+		let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?;
+		if !self.required_keys_exist(&value, room_version_rules).await {
+			return Err!(BadServerResponse(debug_warn!(
+				"Event {event_id} cannot be verified: missing keys."
+			)));
+		}
+		if let Err(e) = self.verify_event(&value, room_version_rules).await {
+			return Err!(BadServerResponse(debug_error!(
+				"Event {event_id} failed verification: {e:?}"
+			)));
+		}
 
-	Ok((event_id, value))
-}
+		value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into()));
 
-#[implement(super::Service)]
-pub async fn validate_and_add_event_id_no_fetch(
-	&self,
-	pdu: &RawJsonValue,
-	room_version_rules: &RoomVersionRules,
-) -> Result<(OwnedEventId, CanonicalJsonObject)> {
-	trace!(?pdu, "Validating PDU without fetching keys");
-	let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?;
-	trace!(event_id = event_id.as_str(), "Generated event ID, checking required keys");
-	if !self.required_keys_exist(&value, room_version_rules).await {
-		debug_warn!(
-			"Event {event_id} is missing required keys, cannot verify without fetching keys"
-		);
-		return Err!(BadServerResponse(debug_warn!(
-			"Event {event_id} cannot be verified: missing keys."
-		)));
+		Ok((event_id, value))
 	}
-	trace!("All required keys exist, verifying event");
-	if let Err(e) = self.verify_event(&value, room_version_rules).await {
-		debug_warn!("Event verification failed");
-		return Err!(BadServerResponse(debug_error!(
-			"Event {event_id} failed verification: {e:?}"
-		)));
+
+	/// Verifies the incoming event after fetching the keys required to do so.
+	pub async fn verify_event(
+		&self,
+		event: &CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<Verified> {
+		let keys = self.get_event_keys(event, room_version_rules).await?;
+		ruma::signatures::verify_event(&keys, event, room_version_rules).map_err(Into::into)
 	}
-	trace!("Event verified successfully");
 
-	value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into()));
-
-	Ok((event_id, value))
-}
-
-#[implement(super::Service)]
-pub async fn verify_event(
-	&self,
-	event: &CanonicalJsonObject,
-	room_version_rules: &RoomVersionRules,
-) -> Result<Verified> {
-	let keys = self.get_event_keys(event, room_version_rules).await?;
-	ruma::signatures::verify_event(&keys, event, room_version_rules).map_err(Into::into)
-}
-
-#[implement(super::Service)]
-pub async fn verify_json(
-	&self,
-	event: &CanonicalJsonObject,
-	room_version_rules: &RoomVersionRules,
-) -> Result {
-	let keys = self.get_event_keys(event, room_version_rules).await?;
-	ruma::signatures::verify_json(&keys, event).map_err(Into::into)
+	/// Verifies an arbitrary JSON object after fetching the keys required to do
+	/// so.
+	pub async fn verify_json(
+		&self,
+		event: &CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result {
+		// TODO: is this actually used anywhere?
+		let keys = self.get_event_keys(event, room_version_rules).await?;
+		ruma::signatures::verify_json(&keys, event).map_err(Into::into)
+	}
 }