diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 18cd6a33a..82251e3dc 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -10,7 +10,7 @@ use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use conduwuit::{ - debug_info, debug_warn, info, + debug_error, debug_info, debug_warn, info, utils::{should_continue_backoff, time::exponential_backoff::next_interval}, }; use conduwuit_core::{ @@ -219,12 +219,14 @@ async fn handle_request<'a>( statuses: &mut CurTransactionStatus, ) { let iv = vec![(msg.queue_id, msg.event)]; - if let Ok(Some(events)) = self.select_events(&msg.dest, iv, statuses).await { - if !events.is_empty() { + match self.select_events(&msg.dest, iv, statuses).await { + | Ok(Some(events)) if !events.is_empty() => { futures.push(self.send_events(msg.dest, events)); - } else { + }, + | Ok(_) => { statuses.remove(&msg.dest); - } + }, + | Err(e) => debug_error!(error=?e, ?msg.dest, "Failed to select events"), } } @@ -400,22 +402,24 @@ fn should_attempt_send( ) -> Result<(bool, bool)> { let (mut allow, mut retry) = (true, false); if let Destination::Federation(server_name) = dest { - statuses - .entry(dest.clone()) - .and_modify(|e| match e { - | TransactionStatus::Running | TransactionStatus::Retrying(_) => { - allow = false; // already running - }, - | TransactionStatus::Failed(tries, _) => { - *e = TransactionStatus::Retrying(*tries); - retry = true; - }, - }) - .or_insert(TransactionStatus::Running); - return match self.services.federation.retry_after(server_name) { - | None => Ok((allow, retry)), - | Some(t) => Ok((t.as_millis() > 0 && allow, retry)), - }; + let entry = statuses.entry(dest.clone()).and_modify(|e| match e { + | TransactionStatus::Running | TransactionStatus::Retrying(_) => { + allow = false; // already running + }, + | TransactionStatus::Failed(tries, _) => { + *e = TransactionStatus::Retrying(*tries); + retry = true; + }, + }); + if let Some(retry_after) = self.services.federation.retry_after(server_name) { + if retry_after.is_zero() && allow { + allow = true; + } + } + if allow { + entry.or_insert(TransactionStatus::Running); + } + return Ok((allow, retry)); } statuses .entry(dest.clone())