diff --git a/src/admin/debug/tester.rs b/src/admin/debug/tester.rs index 80f590f4c..9c1b57e06 100644 --- a/src/admin/debug/tester.rs +++ b/src/admin/debug/tester.rs @@ -11,6 +11,7 @@ pub enum TesterCommand { Timer, } +#[allow(clippy::unused_async_trait_impl)] impl crate::Context<'_> { #[rustfmt::skip] async fn panic(&self) -> Result { diff --git a/src/admin/processor.rs b/src/admin/processor.rs index f27476cf3..8881b820c 100644 --- a/src/admin/processor.rs +++ b/src/admin/processor.rs @@ -49,7 +49,7 @@ async fn handle_command(services: Arc, command: CommandInput) -> Proce async fn process_command(services: Arc, input: &CommandInput) -> ProcessorResult { let (command, args, body) = match parse(&services, input) { - | Err(error) => return Err(error), + | Err(error) => return Err(Box::new(error)), | Ok(parsed) => parsed, }; @@ -72,18 +72,26 @@ async fn process_command(services: Arc, input: &CommandInput) -> Proce String::from_utf8(take(output.get_mut())).expect("invalid utf8 in command output stream"); match result { - | Ok(()) if logs.is_empty() => - Ok(Some(reply(RoomMessageEventContent::notice_markdown(output), context.reply_id))), + | Ok(()) if logs.is_empty() => Ok(Some(Box::new(reply( + RoomMessageEventContent::notice_markdown(output), + context.reply_id, + )))), | Ok(()) => { logs.write_str(output.as_str()).expect("output buffer"); - Ok(Some(reply(RoomMessageEventContent::notice_markdown(logs), context.reply_id))) + Ok(Some(Box::new(reply( + RoomMessageEventContent::notice_markdown(logs), + context.reply_id, + )))) }, | Err(error) => { write!(&mut logs, "Command failed with error:\n```\n{error:#?}\n```") .expect("output buffer"); - Err(reply(RoomMessageEventContent::notice_markdown(logs), context.reply_id)) + Err(Box::new(reply( + RoomMessageEventContent::notice_markdown(logs), + context.reply_id, + ))) }, } } @@ -94,7 +102,7 @@ fn handle_panic(error: &Error, command: &CommandInput) -> ProcessorResult { let msg = format!("Panic occurred while processing command:\n```\n{error:#?}\n```\n{link}"); let content = RoomMessageEventContent::notice_markdown(msg); error!("Panic while processing command: {error:?}"); - Err(reply(content, command.reply_id.as_deref())) + Err(Box::new(reply(content, command.reply_id.as_deref()))) } /// Parse and process a message from the admin room diff --git a/src/api/client/device.rs b/src/api/client/device.rs index 843532581..0c3161808 100644 --- a/src/api/client/device.rs +++ b/src/api/client/device.rs @@ -68,8 +68,7 @@ pub(crate) async fn update_device_route( services .users - .update_device_metadata(sender_user, &body.device_id, &device) - .await?; + .update_device_metadata(sender_user, &body.device_id, &device)?; Ok(update_device::v3::Response::new()) }, diff --git a/src/api/client_ip.rs b/src/api/client_ip.rs index d477490e4..704938e18 100644 --- a/src/api/client_ip.rs +++ b/src/api/client_ip.rs @@ -104,12 +104,15 @@ impl FromRequestParts for ClientIp { type Rejection = ClientIpError; - async fn from_request_parts(parts: &mut Parts, state: &S) -> Result { - Self::for_config( + fn from_request_parts( + parts: &mut Parts, + state: &S, + ) -> impl Future> { + std::future::ready(Self::for_config( parts, &state.config.accepted_ip_sources, state.config.request_ip_source.as_ref(), - ) + )) } } diff --git a/src/api/router/auth.rs b/src/api/router/auth.rs index f5fdc3e5c..f62f83eb5 100644 --- a/src/api/router/auth.rs +++ b/src/api/router/auth.rs @@ -331,14 +331,14 @@ async fn verify + Sync>( impl CheckAuth for NoAuthentication { type Identity = (); - async fn verify + Sync>( + fn verify + Sync>( _services: &Services, _output: Self::Output, _request: &hyper::Request, _query: AuthQueryParams, _route: TypeId, - ) -> Result { - Ok(()) + ) -> impl Future> { + std::future::ready(Ok(())) } } diff --git a/src/service/admin/console.rs b/src/service/admin/console.rs index be2426e8c..68c1cf0d6 100644 --- a/src/service/admin/console.rs +++ b/src/service/admin/console.rs @@ -49,12 +49,14 @@ pub(super) async fn handle_signal(self: &Arc, sig: &'static str) { } } - pub async fn start(self: &Arc) { + pub fn start(self: &Arc) -> impl Future { let mut worker_join = self.worker_join.lock(); if worker_join.is_none() { let self_ = Arc::clone(self); _ = worker_join.insert(self.server.runtime().spawn(self_.worker())); } + + std::future::ready(()) } pub async fn close(self: &Arc) { diff --git a/src/service/admin/mod.rs b/src/service/admin/mod.rs index 2dda07cdb..0ca689292 100644 --- a/src/service/admin/mod.rs +++ b/src/service/admin/mod.rs @@ -104,7 +104,7 @@ pub fn allows_restricted(&self) -> bool { !matches!(self, Self::EscapedCommand) /// events which have digested any prior errors. The wrapping preserves whether /// the command failed without interpreting the text. Ok(None) outputs are /// dropped to produce no response. -pub type ProcessorResult = Result, CommandOutput>; +pub type ProcessorResult = Result>, Box>; /// Alias for the output structure. pub type CommandOutput = RoomMessageEventContent; @@ -204,7 +204,7 @@ pub async fn text_or_file( .text_to_file(message_content.body()) .await .expect("failed to create text file"); - let size_u64: u64 = message_content.body().len().try_into().map_or(0, |n| n); + let size_u64: u64 = message_content.body().len().try_into().unwrap_or(0); let mut metadata = FileInfo::new(); metadata.mimetype = Some("text/markdown".to_owned()); @@ -365,7 +365,7 @@ async fn handle_command(&self, command: CommandInput) { match self.process_command(command).await { | Ok(None) => debug!("Command successful with no response"), | Ok(Some(output)) | Err(output) => self - .handle_response(output) + .handle_response(*output) .await .unwrap_or_else(default_log), } diff --git a/src/service/manager.rs b/src/service/manager.rs index 7a2e50d50..25afe80a3 100644 --- a/src/service/manager.rs +++ b/src/service/manager.rs @@ -65,7 +65,7 @@ pub(super) async fn start(self: Arc) -> Result<()> { debug!("Starting service workers..."); for service in services { - self.start_worker(&mut workers, &service).await?; + self.start_worker(&mut workers, &service)?; } Ok(()) @@ -108,20 +108,14 @@ async fn handle_result( ) -> Result<()> { let (service, result) = result; match result { - | Ok(()) => self.handle_finished(workers, &service).await, + | Ok(()) => { + debug!("service {:?} worker finished", service.name()); + Ok(()) + }, | Err(error) => self.handle_error(workers, &service, error).await, } } - async fn handle_finished( - &self, - _workers: &mut WorkersLocked<'_>, - service: &Arc, - ) -> Result<()> { - debug!("service {:?} worker finished", service.name()); - Ok(()) - } - async fn handle_error( &self, workers: &mut WorkersLocked<'_>, @@ -144,11 +138,11 @@ async fn handle_error( warn!("service {name:?} worker restarting after {} delay", time::pretty(delay)); sleep(delay).await; - self.start_worker(workers, service).await + self.start_worker(workers, service) } /// Start the worker in a task for the service. - async fn start_worker( + fn start_worker( &self, workers: &mut WorkersLocked<'_>, service: &Arc, diff --git a/src/service/media/preview.rs b/src/service/media/preview.rs index 2d08e793a..33a05950e 100644 --- a/src/service/media/preview.rs +++ b/src/service/media/preview.rs @@ -49,18 +49,22 @@ pub struct UrlPreviewData { } impl Service { - pub async fn remove_url_preview(&self, url: &str) -> Result<()> { + pub fn remove_url_preview(&self, url: &str) -> impl Future> { // TODO: also remove the downloaded image - self.db.remove_url_preview(url) + std::future::ready(self.db.remove_url_preview(url)) } pub async fn clear_url_previews(&self) { self.db.clear_url_previews().await; } - pub async fn set_url_preview(&self, url: &str, data: &UrlPreviewData) -> Result<()> { + pub fn set_url_preview( + &self, + url: &str, + data: &UrlPreviewData, + ) -> impl Future> { let now = SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) .expect("valid system time"); - self.db.set_url_preview(url, data, now) + std::future::ready(self.db.set_url_preview(url, data, now)) } pub async fn get_url_preview(&self, url: &Url) -> Result { diff --git a/src/service/rooms/auth_chain/data.rs b/src/service/rooms/auth_chain/data.rs index e9e40979a..ab8bbdc5d 100644 --- a/src/service/rooms/auth_chain/data.rs +++ b/src/service/rooms/auth_chain/data.rs @@ -48,8 +48,10 @@ pub(super) async fn get_cached_eventid_authchain( .map_err(|_| err!(Request(NotFound("auth_chain not found"))))?; let chain = chain - .chunks_exact(size_of::()) - .map(utils::u64_from_u8) + .as_chunks::<{ size_of::() }>() + .0 + .iter() + .map(|s| utils::u64_from_u8(s)) .collect::>(); // Cache in RAM diff --git a/src/service/rooms/state_compressor/mod.rs b/src/service/rooms/state_compressor/mod.rs index d0919aefb..ddd25eac1 100644 --- a/src/service/rooms/state_compressor/mod.rs +++ b/src/service/rooms/state_compressor/mod.rs @@ -136,14 +136,14 @@ pub async fn load_shortstatehash_info( /// Returns a stack with info on shortstatehash, full state, added diff and /// removed diff for the selected shortstatehash and each parent layer. - async fn cache_shortstatehash_info( + fn cache_shortstatehash_info( &self, shortstatehash: ShortStateHash, stack: ShortStateInfoVec, - ) -> Result { + ) -> impl Future { self.stateinfo_cache.lock().insert(shortstatehash, stack); - Ok(()) + std::future::ready(Ok(())) } /// Inserts a new shortstatehash info entry. diff --git a/src/service/rooms/timeline/helpers.rs b/src/service/rooms/timeline/helpers.rs index 50384bba6..9175377f7 100644 --- a/src/service/rooms/timeline/helpers.rs +++ b/src/service/rooms/timeline/helpers.rs @@ -82,8 +82,7 @@ async fn assert_allowed_to_send_state_event( )))); }, | StateEventType::RoomServerAcl => - self.assert_allowed_to_send_room_server_acl_event(room_id, json) - .await?, + self.assert_allowed_to_send_room_server_acl_event(room_id, json)?, | StateEventType::RoomEncryption => // Forbid m.room.encryption if encryption is disabled if !self.services.config.allow_encryption { @@ -109,7 +108,7 @@ async fn assert_allowed_to_send_state_event( Ok(()) } - async fn assert_allowed_to_send_room_server_acl_event( + fn assert_allowed_to_send_room_server_acl_event( &self, room_id: &RoomId, json: &Raw, diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 3bf06f08c..1007bc067 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -61,7 +61,7 @@ enum TransactionStatus { Retrying(u32), // number of times failed } -type SendingError = (Destination, Error); +type SendingError = (Destination, Box); type SendingResult = Result; type SendingFuture<'a> = BoxFuture<'a, SendingResult>; type SendingFutures<'a> = FuturesUnordered>; @@ -779,7 +779,7 @@ async fn send_events_dest_appservice( let Some(appservice) = self.services.appservice.get_registration(&id).await else { return Err(( Destination::Appservice(id.clone()), - err!(Database(warn!(?id, "Missing appservice registration"))), + Box::new(err!(Database(warn!(?id, "Missing appservice registration")))), )); }; @@ -827,7 +827,7 @@ async fn send_events_dest_appservice( match self.send_appservice_request(appservice, request).await { | Ok(_) => Ok(Destination::Appservice(id)), - | Err(e) => Err((Destination::Appservice(id), e)), + | Err(e) => Err((Destination::Appservice(id), Box::new(e))), } } @@ -848,7 +848,7 @@ async fn send_events_dest_push( let Ok(pusher) = self.services.pusher.get_pusher(&user_id, &pushkey).await else { return Err(( Destination::Push(user_id.clone(), pushkey.clone()), - err!(Database(error!(%user_id, ?pushkey, "Missing pusher"))), + Box::new(err!(Database(error!(%user_id, ?pushkey, "Missing pusher")))), )); }; @@ -990,17 +990,17 @@ async fn send_events_dest_federation( } match result { - | Err(error) => Err((Destination::Federation(server), error)), + | Err(error) => Err((Destination::Federation(server), Box::new(error))), | Ok(_) => Ok(Destination::Federation(server)), } } /// Converts and sanitises an outgoing PDU object for federation /// transmission. - pub async fn convert_to_outgoing_federation_event( + pub fn convert_to_outgoing_federation_event( &self, mut pdu_json: CanonicalJsonObject, - ) -> Box { + ) -> impl Future> { if let Some(unsigned) = pdu_json .get_mut("unsigned") .and_then(|val| val.as_object_mut()) @@ -1012,6 +1012,8 @@ pub async fn convert_to_outgoing_federation_event( // so we can safely remove it here. pdu_json.remove("event_id"); - to_raw_value(&pdu_json).expect("CanonicalJson is valid serde_json::Value") + std::future::ready( + to_raw_value(&pdu_json).expect("CanonicalJson is valid serde_json::Value"), + ) } } diff --git a/src/service/users/device.rs b/src/service/users/device.rs index 3ad105dae..01b644ec5 100644 --- a/src/service/users/device.rs +++ b/src/service/users/device.rs @@ -263,7 +263,7 @@ pub async fn remove_to_device_events( } /// Updates device metadata and increments the device list version. - pub async fn update_device_metadata( + pub fn update_device_metadata( &self, user_id: &UserId, device_id: &DeviceId,