feat: Implement MSC4524

Cargo.lock

@@ -871,6 +871,7 @@ dependencies = [
  "log",
  "resolvematrix",
  "ruma",
+ "ruminuwuity",
  "serde-saphyr",
  "serde_json",
  "tokio",

src/admin/Cargo.toml

@@ -11,65 +11,65 @@ version.workspace = true
 [lib]
 path = "mod.rs"
 crate-type = [
-	"rlib",
-#	"dylib",
+    "rlib",
+    #	"dylib",
 ]
 
 [features]
 brotli_compression = [
-	"conduwuit-api/brotli_compression",
-	"conduwuit-core/brotli_compression",
-	"conduwuit-service/brotli_compression",
+    "conduwuit-api/brotli_compression",
+    "conduwuit-core/brotli_compression",
+    "conduwuit-service/brotli_compression",
 ]
 gzip_compression = [
-	"conduwuit-api/gzip_compression",
-	"conduwuit-core/gzip_compression",
-	"conduwuit-service/gzip_compression",
+    "conduwuit-api/gzip_compression",
+    "conduwuit-core/gzip_compression",
+    "conduwuit-service/gzip_compression",
 ]
 io_uring = [
-	"conduwuit-api/io_uring",
-	"conduwuit-database/io_uring",
-	"conduwuit-service/io_uring",
+    "conduwuit-api/io_uring",
+    "conduwuit-database/io_uring",
+    "conduwuit-service/io_uring",
 ]
 jemalloc = [
-	"conduwuit-api/jemalloc",
-	"conduwuit-core/jemalloc",
-	"conduwuit-database/jemalloc",
-	"conduwuit-service/jemalloc",
+    "conduwuit-api/jemalloc",
+    "conduwuit-core/jemalloc",
+    "conduwuit-database/jemalloc",
+    "conduwuit-service/jemalloc",
 ]
 jemalloc_conf = [
-	"conduwuit-api/jemalloc_conf",
-	"conduwuit-core/jemalloc_conf",
-	"conduwuit-database/jemalloc_conf",
-	"conduwuit-service/jemalloc_conf",
+    "conduwuit-api/jemalloc_conf",
+    "conduwuit-core/jemalloc_conf",
+    "conduwuit-database/jemalloc_conf",
+    "conduwuit-service/jemalloc_conf",
 ]
 jemalloc_prof = [
-	"conduwuit-api/jemalloc_prof",
-	"conduwuit-core/jemalloc_prof",
-	"conduwuit-database/jemalloc_prof",
-	"conduwuit-service/jemalloc_prof",
+    "conduwuit-api/jemalloc_prof",
+    "conduwuit-core/jemalloc_prof",
+    "conduwuit-database/jemalloc_prof",
+    "conduwuit-service/jemalloc_prof",
 ]
 jemalloc_stats = [
-	"conduwuit-api/jemalloc_stats",
-	"conduwuit-core/jemalloc_stats",
-	"conduwuit-database/jemalloc_stats",
-	"conduwuit-service/jemalloc_stats",
+    "conduwuit-api/jemalloc_stats",
+    "conduwuit-core/jemalloc_stats",
+    "conduwuit-database/jemalloc_stats",
+    "conduwuit-service/jemalloc_stats",
 ]
 release_max_log_level = [
-	"conduwuit-api/release_max_log_level",
-	"conduwuit-core/release_max_log_level",
-	"conduwuit-database/release_max_log_level",
-	"conduwuit-service/release_max_log_level",
-	"tracing/max_level_trace",
-	"tracing/release_max_level_info",
-	"log/max_level_trace",
-	"log/release_max_level_info",
+    "conduwuit-api/release_max_log_level",
+    "conduwuit-core/release_max_log_level",
+    "conduwuit-database/release_max_log_level",
+    "conduwuit-service/release_max_log_level",
+    "tracing/max_level_trace",
+    "tracing/release_max_level_info",
+    "log/max_level_trace",
+    "log/release_max_level_info",
 ]
 zstd_compression = [
-	"conduwuit-api/zstd_compression",
-	"conduwuit-core/zstd_compression",
-	"conduwuit-database/zstd_compression",
-	"conduwuit-service/zstd_compression",
+    "conduwuit-api/zstd_compression",
+    "conduwuit-core/zstd_compression",
+    "conduwuit-database/zstd_compression",
+    "conduwuit-service/zstd_compression",
 ]
 
 [dependencies]
@@ -87,6 +87,7 @@ lettre.workspace = true
 log.workspace = true
 assign.workspace = true
 ruma.workspace = true
+ruminuwuity.workspace = true
 serde_json.workspace = true
 serde-saphyr.workspace = true
 tokio.workspace = true

src/admin/debug/commands.rs

@@ -6,13 +6,14 @@
 };
 
 use conduwuit::{
-	Err, Result, at, debug_error, err, info,
+	Err, Result, at, debug_error, debug_info, err, info,
 	matrix::{
 		Event,
 		pdu::{PduEvent, PduId, RawPduId},
 	},
 	trace, utils,
 	utils::{
+		random_string,
 		stream::{IterStream, ReadyExt},
 		string::EMPTY,
 	},
@@ -26,8 +27,10 @@
 };
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
-	OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, UInt,
-	api::federation::event::get_room_state, events::AnyStateEvent, serde::Raw,
+	OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, ServerName, UInt,
+	api::{error::ErrorKind, federation::event::get_room_state},
+	events::AnyStateEvent,
+	serde::Raw,
 };
 use service::rooms::{
 	short::{ShortEventId, ShortRoomId},
@@ -581,14 +584,60 @@ pub(super) async fn ping(&self, server: OwnedServerName) -> Result {
 		if server == self.services.globals.server_name() {
 			return Err!("Not allowed to send federation requests to ourselves.");
 		}
+		let modern = self.modern_pingpong_ping(&server).await?;
+		if let Some(w) = modern {
+			self.write_str(&w).await
+		} else {
+			info!("{server} does not support msc4524, falling back to legacy ping");
+			self.legacy_ping(&server).await
+		}
+	}
 
+	async fn modern_pingpong_ping(&self, server: &ServerName) -> Result<Option<String>> {
+		let timer = tokio::time::Instant::now();
+		let question = &random_string(64);
+
+		let response = self
+			.services
+			.sending
+			.send_federation_request(
+				server,
+				ruminuwuity::api::federation::ping::unstable::Request::new(question.to_owned()),
+			)
+			.await;
+
+		let answer = match response {
+			| Ok(resp) => resp.answer,
+			| Err(e) =>
+				return if e.is_not_found() && e.kind() == ErrorKind::Unrecognized {
+					Ok(None)
+				} else {
+					Err!("Failed sending federation request to specified server:\n\n{e}")
+				},
+		};
+		let recv = self
+			.services
+			.federation
+			.register_ping_answer(answer.clone())?;
+		debug_info!("Waiting for a pong for up to 60 seconds");
+		let result = tokio::time::timeout(std::time::Duration::from_mins(1), recv).await;
+		match result {
+			| Err(e) => Err!(BadServerResponse("Remote did not answer the ping in time: {e:?}")),
+			| Ok(Err(e)) =>
+				Err!(BadServerResponse("Unexpectedly could not read from pong channel: {e:?}")),
+			| Ok(Ok(())) =>
+				Ok(Some(format!("Ping got a pong in {:?}: {answer}", timer.elapsed()))),
+		}
+	}
+
+	async fn legacy_ping(&self, server: &ServerName) -> Result {
 		let timer = tokio::time::Instant::now();
 
 		match self
 			.services
 			.sending
 			.send_unauthenticated_request(
-				&server,
+				server,
 				ruma::api::federation::discovery::get_server_version::v1::Request::new(),
 			)
 			.await

src/api/router.rs

@@ -229,6 +229,8 @@ pub fn build(router: Router<State>, state: State) -> Router<State> {
 			.ruma_route(&server::well_known_server)
 			.ruma_route(&server::get_content_route)
 			.ruma_route(&server::get_content_thumbnail_route)
+			.ruma_route(&server::ping)
+			.ruma_route(&server::pong)
 			.route("/_continuwuity/local_user_count", get(client::continuwuity_local_user_count));
 	} else {
 		router = router

src/api/server/mod.rs

@@ -10,6 +10,8 @@
 pub(super) mod make_leave;
 pub(super) mod media;
 pub(super) mod openid;
+pub(super) mod ping;
+pub(super) mod pong;
 pub(super) mod publicrooms;
 pub(super) mod query;
 pub(super) mod send;
@@ -34,6 +36,8 @@
 pub(super) use make_leave::*;
 pub(super) use media::*;
 pub(super) use openid::*;
+pub(super) use ping::ping;
+pub(super) use pong::pong;
 pub(super) use publicrooms::*;
 pub(super) use query::*;
 pub(super) use send::*;
@@ -45,6 +49,6 @@
 pub(super) use user::*;
 pub(super) use version::*;
 pub(super) use well_known::*;
-
 mod utils;
+
 use utils::AccessCheck;

src/api/server/ping.rs

@@ -0,0 +1,55 @@
+use std::{sync::Arc, time::Duration};
+
+use axum::extract::State;
+use conduwuit::{
+	Result, info,
+	utils::{random_string, time::jitter},
+	warn,
+};
+use ruma::{OwnedServerName, api::error::ErrorKind};
+use ruminuwuity::api::federation::{
+	ping::unstable::{Request, Response},
+	pong,
+};
+
+use crate::Ruma;
+
+/// # `GET /_matrix/federation/unstable/uk.timedout.msc0000.tabletennis/ping`
+///
+/// Initiates a ping
+pub(crate) async fn ping(
+	State(services): State<crate::State>,
+	body: Ruma<Request>,
+) -> Result<Response> {
+	// TODO: rate-limiting
+	// TODO: make answer stable based on question
+	let answer = random_string(64);
+	services.federation.register_ping_answer(answer.clone())?;
+	services
+		.server
+		.runtime()
+		.spawn(send_ping(Arc::new(services), body.identity, answer.clone()));
+	Ok(Response::new(answer))
+}
+
+async fn send_ping(services: Arc<crate::State>, target: OwnedServerName, answer: String) {
+	tokio::time::sleep(jitter(Duration::from_secs(1), 1.0..=10.0)).await;
+	info!(%answer, "Sending a federation pong to {target}");
+	match services
+		.sending
+		.send_federation_request(&target, pong::unstable::Request::new(answer))
+		.await
+	{
+		| Ok(_) => info!(%target, "Federation pong succeeded"),
+		| Err(e) =>
+			if e.is_not_found() {
+				if e.kind() == ErrorKind::Unrecognized {
+					warn!(%target, "Remote requested a ping but does not recognize the pong.");
+				} else {
+					info!(%target, "Ping expired by the time we ponged.");
+				}
+			} else {
+				warn!(%target, "Pong failed");
+			},
+	}
+}

src/api/server/pong.rs

@@ -0,0 +1,17 @@
+use axum::extract::State;
+use conduwuit::Result;
+use ruminuwuity::api::federation::pong;
+
+use crate::Ruma;
+
+/// # `GET /_matrix/federation/unstable/uk.timedout.msc0000.tabletennis/pong`
+///
+/// Responds to a ping
+pub(crate) async fn pong(
+	State(services): State<crate::State>,
+	body: Ruma<pong::unstable::Request>,
+) -> Result<pong::unstable::Response> {
+	// TODO: verify incoming question is already registered
+	services.federation.answer_ping(&body.question)?;
+	Ok(pong::unstable::Response::default())
+}

src/ruminuwuity/api/federation/mod.rs

@@ -0,0 +1,2 @@
+pub mod ping;
+pub mod pong;

src/ruminuwuity/api/federation/ping.rs

@@ -0,0 +1,43 @@
+//! `GET /_matrix/federation/*/ping`
+//!
+//! Initiate a federation ping
+pub mod unstable {
+	use ruma::{
+		OwnedServerName,
+		api::{federation::authentication::ServerSignatures, request, response},
+		metadata,
+	};
+
+	metadata! {
+		method: POST,
+		rate_limited: true,
+		authentication: ServerSignatures,  // TODO: Needs to be optional
+		path: "/_matrix/federation/unstable/uk.timedout.msc4524.tabletennis/ping"
+	}
+
+	#[request]
+	pub struct Request {
+		#[serde(default, skip_serializing_if = "Option::is_none")]
+		pub origin: Option<OwnedServerName>,
+
+		pub question: String,
+	}
+
+	#[response]
+	pub struct Response {
+		pub answer: String,
+
+		#[serde(default, skip_serializing_if = "Vec::is_empty")]
+		pub details: Vec<String>,
+	}
+
+	impl Request {
+		#[must_use]
+		pub fn new(question: String) -> Self { Self { question, origin: None } }
+	}
+
+	impl Response {
+		#[must_use]
+		pub fn new(answer: String) -> Self { Self { answer, details: vec![] } }
+	}
+}

src/ruminuwuity/api/federation/pong.rs

@@ -0,0 +1,30 @@
+//! `GET /_matrix/federation/*/pong`
+//!
+//! Respond to a federation ping
+pub mod unstable {
+	use ruma::{
+		api::{federation::authentication::ServerSignatures, request, response},
+		metadata,
+	};
+
+	metadata! {
+		method: POST,
+		rate_limited: false,
+		authentication: ServerSignatures,
+		path: "/_matrix/federation/unstable/uk.timedout.msc4524.tabletennis/ping"
+	}
+
+	#[request]
+	pub struct Request {
+		pub question: String,
+	}
+
+	#[response]
+	#[derive(Default)]
+	pub struct Response;
+
+	impl Request {
+		#[must_use]
+		pub fn new(question: String) -> Self { Self { question } }
+	}
+}

src/ruminuwuity/api/mod.rs

@@ -0,0 +1,3 @@
+pub mod federation;
+
+pub use federation::*;

src/ruminuwuity/mod.rs

@@ -5,3 +5,5 @@
 pub mod invite_permission_config;
 pub mod meowlnir_antispam;
 pub mod policy;
+
+pub mod api;

src/service/federation/mod.rs

@@ -1,13 +1,16 @@
 mod execute;
-use std::sync::Arc;
 
-use conduwuit::{Result, Server};
+use std::{collections::HashMap, sync::Arc};
+
+use conduwuit::{Err, Result, Server, SyncRwLock, err};
 pub(crate) use execute::FederationPathBuilderInput;
 
 use crate::{Dep, client, moderation, server_keys};
 
 pub struct Service {
 	services: Services,
+	/// A map of {answer: channel}
+	pingpongs: SyncRwLock<HashMap<String, tokio::sync::oneshot::Sender<()>>>,
 }
 
 struct Services {
@@ -26,8 +29,42 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				server_keys: args.depend::<server_keys::Service>("server_keys"),
 				moderation: args.depend::<moderation::Service>("moderation"),
 			},
+			pingpongs: SyncRwLock::new(HashMap::new()),
 		}))
 	}
 
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
+
+impl Service {
+	/// Registers an outbound ping waiter based on the answer returned by the
+	/// remote. Returns a channel that is written to when the remote pongs.
+	pub fn register_ping_answer(
+		&self,
+		expected_answer: String,
+	) -> Result<tokio::sync::oneshot::Receiver<()>> {
+		let mut pingpongs = self.pingpongs.write();
+		if pingpongs.contains_key(&expected_answer) {
+			return Err!(Request(InvalidParam("Duplicate answer")));
+		}
+		let (tx, rx) = tokio::sync::oneshot::channel();
+		pingpongs.insert(expected_answer, tx);
+		Ok(rx)
+	}
+
+	/// "Answers" a registered outbound ping by sending an event to it. This is
+	/// called when the remote server that was pinged calls /pong.
+	///
+	/// `M_NOT_FOUND` is returned if the answer is not recognised.
+	pub fn answer_ping(&self, answer: &str) -> Result<()> {
+		let mut pingpongs = self.pingpongs.write();
+		let Some(tx) = pingpongs.remove(answer) else {
+			return Err!(Request(NotFound("Unknown answer")));
+		};
+		tx.send(()).map_err(|e| {
+			err!(BadServerResponse(error!(
+				error=?e, "Failed to handle pong"
+			)))
+		})
+	}
+}