Fix futures not Send

Signed-off-by: Jason Volk <jason@zemos.net>

src/api/client/keys.rs

@@ -248,7 +248,7 @@ pub(crate) async fn get_key_changes_route(
 	})
 }
 
-pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool>(
+pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool + Send>(
 	sender_user: Option<&UserId>, device_keys_input: &BTreeMap<OwnedUserId, Vec<OwnedDeviceId>>, allowed_signatures: F,
 	include_display_names: bool,
 ) -> Result<get_keys::v3::Response> {

src/router/mod.rs

@@ -14,16 +14,16 @@ conduit::mod_ctor! {}
 conduit::mod_dtor! {}
 
 #[no_mangle]
-pub extern "Rust" fn start(server: &Arc<Server>) -> Pin<Box<dyn Future<Output = Result<()>>>> {
+pub extern "Rust" fn start(server: &Arc<Server>) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
 	Box::pin(run::start(server.clone()))
 }
 
 #[no_mangle]
-pub extern "Rust" fn stop(server: &Arc<Server>) -> Pin<Box<dyn Future<Output = Result<()>>>> {
+pub extern "Rust" fn stop(server: &Arc<Server>) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
 	Box::pin(run::stop(server.clone()))
 }
 
 #[no_mangle]
-pub extern "Rust" fn run(server: &Arc<Server>) -> Pin<Box<dyn Future<Output = Result<()>>>> {
+pub extern "Rust" fn run(server: &Arc<Server>) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
 	Box::pin(run::run(server.clone()))
 }

src/service/globals/migrations.rs

@@ -528,15 +528,20 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
 
 		#[cfg(feature = "sha256_media")]
 		{
+			use std::path::PathBuf;
 			if services().globals.database_version()? < 14 && cfg!(feature = "sha256_media") {
 				warn!("sha256_media feature flag is enabled, migrating legacy base64 file names to sha256 file names");
 				// Move old media files to new names
+				let mut changes = Vec::<(PathBuf, PathBuf)>::new();
 				for (key, _) in db.mediaid_file.iter() {
 					let old_path = services().globals.get_media_file(&key);
 					debug!("Old file path: {old_path:?}");
 					let path = services().globals.get_media_file_new(&key);
 					debug!("New file path: {path:?}");
-					// move the file to the new location
+					changes.push((old_path, path));
+				}
+				// move the file to the new location
+				for (old_path, path) in changes {
 					if old_path.exists() {
 						tokio::fs::rename(&old_path, &path).await?;
 					}

src/service/pusher/mod.rs

@@ -47,7 +47,7 @@ impl Service {
 	#[tracing::instrument(skip(self, dest, request))]
 	pub async fn send_request<T>(&self, dest: &str, request: T) -> Result<T::IncomingResponse>
 	where
-		T: OutgoingRequest + Debug,
+		T: OutgoingRequest + Debug + Send,
 	{
 		const VERSIONS: [MatrixVersion; 1] = [MatrixVersion::V1_0];
 

src/service/rooms/event_handler/signing_keys.rs

@@ -28,7 +28,7 @@ impl super::Service {
 		&'a self, events: E, pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
 	) -> Result<()>
 	where
-		E: IntoIterator<Item = &'a BTreeMap<String, CanonicalJsonValue>>,
+		E: IntoIterator<Item = &'a BTreeMap<String, CanonicalJsonValue>> + Send,
 	{
 		let mut server_key_ids = HashMap::new();
 		for event in events {

src/service/sending/appservice.rs

@@ -12,7 +12,7 @@ use crate::{debug_error, services, utils, Error, Result};
 /// registration file
 pub(crate) async fn send_request<T>(registration: Registration, request: T) -> Result<Option<T::IncomingResponse>>
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	const VERSIONS: [MatrixVersion; 1] = [MatrixVersion::V1_0];
 

src/service/sending/mod.rs

@@ -219,7 +219,7 @@ impl Service {
 	#[tracing::instrument(skip(self, request), name = "request")]
 	pub async fn send_federation_request<T>(&self, dest: &ServerName, request: T) -> Result<T::IncomingResponse>
 	where
-		T: OutgoingRequest + Debug,
+		T: OutgoingRequest + Debug + Send,
 	{
 		let client = &services().globals.client.federation;
 		send::send(client, dest, request).await
@@ -233,7 +233,7 @@ impl Service {
 		&self, registration: Registration, request: T,
 	) -> Result<Option<T::IncomingResponse>>
 	where
-		T: OutgoingRequest + Debug,
+		T: OutgoingRequest + Debug + Send,
 	{
 		appservice::send_request(registration, request).await
 	}

src/service/sending/send.rs

@@ -18,7 +18,7 @@ use crate::{debug_error, debug_warn, services, Error, Result};
 #[tracing::instrument(skip_all, name = "send")]
 pub async fn send<T>(client: &Client, dest: &ServerName, req: T) -> Result<T::IncomingResponse>
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	if !services().globals.allow_federation() {
 		return Err(Error::bad_config("Federation is disabled."));
@@ -33,7 +33,7 @@ async fn execute<T>(
 	client: &Client, dest: &ServerName, actual: &ActualDest, request: Request,
 ) -> Result<T::IncomingResponse>
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	let method = request.method().clone();
 	let url = request.url().clone();
@@ -50,7 +50,7 @@ where
 
 async fn prepare<T>(dest: &ServerName, actual: &ActualDest, req: T) -> Result<Request>
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	const VERSIONS: [MatrixVersion; 1] = [MatrixVersion::V1_5];
 
@@ -72,7 +72,7 @@ async fn handle_response<T>(
 	dest: &ServerName, actual: &ActualDest, method: &Method, url: &Url, mut response: Response,
 ) -> Result<T::IncomingResponse>
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	trace!("Received response from {} for {} with {}", actual.string, url, response.url());
 	let status = response.status();
@@ -121,7 +121,7 @@ fn handle_error<T>(
 	_dest: &ServerName, actual: &ActualDest, method: &Method, url: &Url, mut e: reqwest::Error,
 ) -> Result<T::IncomingResponse>
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	if e.is_timeout() || e.is_connect() {
 		e = e.without_url();
@@ -144,7 +144,7 @@ where
 
 fn sign_request<T>(dest: &ServerName, http_request: &mut http::Request<Vec<u8>>)
 where
-	T: OutgoingRequest + Debug,
+	T: OutgoingRequest + Debug + Send,
 {
 	let mut req_map = serde_json::Map::new();
 	if !http_request.body().is_empty() {