Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-04-17 21:46:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

Simplify ratelimiting

Browse Source
This commit is contained in:
link2xt
2022-06-18 12:02:27 +00:00
parent cddd38cdff
commit aef19cb0e0
3 changed files with 26 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -3,6 +3,7 @@
## Unreleased ## Unreleased
### Changes ### Changes
- refactorings #3437
### Fixes ### Fixes

40
src/scheduler.rs
View File

@@ -324,29 +324,25 @@ async fn smtp_loop(ctx: Context, started: Sender<()>, smtp_handlers: SmtpConnect
let mut timeout = None; let mut timeout = None;
loop { loop {
match send_smtp_messages(&ctx, &mut connection).await { if let Err(err) = send_smtp_messages(&ctx, &mut connection).await {
Err(err) => { warn!(ctx, "send_smtp_messages failed: {:#}", err);
warn!(ctx, "send_smtp_messages failed: {:#}", err); timeout = Some(timeout.map_or(30, |timeout: u64| timeout.saturating_mul(3)))
timeout = Some(timeout.map_or(30, |timeout: u64| timeout.saturating_mul(3))) } else {
} let duration_until_can_send = ctx.ratelimit.read().await.until_can_send();
Ok(ratelimited) => { if !duration_until_can_send.is_zero() {
if ratelimited { info!(
let duration_until_can_send = ctx.ratelimit.read().await.until_can_send(); ctx,
info!( "smtp got rate limited, waiting for {} until can send again",
ctx, duration_to_str(duration_until_can_send)
"smtp got rate limited, waiting for {} until can send again", );
duration_to_str(duration_until_can_send) tokio::time::timeout(duration_until_can_send, async {
); idle_interrupt_receiver.recv().await.unwrap_or_default()
tokio::time::timeout(duration_until_can_send, async { })
idle_interrupt_receiver.recv().await.unwrap_or_default() .await
}) .unwrap_or_default();
.await continue;
.unwrap_or_default();
continue;
} else {
timeout = None;
}
} }
timeout = None;
} }
// Fake Idle // Fake Idle

21
src/smtp.rs
View File

@@ -477,21 +477,17 @@ pub(crate) async fn send_msg_to_smtp(
} }
/// Attempts to send queued MDNs. /// Attempts to send queued MDNs.
/// async fn send_mdns(context: &Context, connection: &mut Smtp) -> Result<()> {
/// Returns true if there are more MDNs to send, but rate limiter does not
/// allow to send them. Returns false if there are no more MDNs to send.
/// If sending an MDN fails, returns an error.
async fn send_mdns(context: &Context, connection: &mut Smtp) -> Result<bool> {
loop { loop {
if !context.ratelimit.read().await.can_send() { if !context.ratelimit.read().await.can_send() {
info!(context, "Ratelimiter does not allow sending MDNs now"); info!(context, "Ratelimiter does not allow sending MDNs now");
return Ok(true); return Ok(());
} }
let more_mdns = send_mdn(context, connection).await?; let more_mdns = send_mdn(context, connection).await?;
if !more_mdns { if !more_mdns {
// No more MDNs to send. // No more MDNs to send.
return Ok(false); return Ok(());
} }
} }
} }
@@ -500,10 +496,8 @@ async fn send_mdns(context: &Context, connection: &mut Smtp) -> Result<bool> {
/// ///
/// Logs and ignores SMTP errors to ensure that a single SMTP message constantly failing to be sent /// Logs and ignores SMTP errors to ensure that a single SMTP message constantly failing to be sent
/// does not block other messages in the queue from being sent. /// does not block other messages in the queue from being sent.
/// pub(crate) async fn send_smtp_messages(context: &Context, connection: &mut Smtp) -> Result<()> {
/// Returns true if sending was ratelimited, false otherwise. Errors are propagated to the caller. let ratelimited = if context.ratelimit.read().await.can_send() {
pub(crate) async fn send_smtp_messages(context: &Context, connection: &mut Smtp) -> Result<bool> {
let mut ratelimited = if context.ratelimit.read().await.can_send() {
// add status updates and sync messages to end of sending queue // add status updates and sync messages to end of sending queue
context.flush_status_updates().await?; context.flush_status_updates().await?;
context.send_sync_msg().await?; context.send_sync_msg().await?;
@@ -538,12 +532,11 @@ pub(crate) async fn send_smtp_messages(context: &Context, connection: &mut Smtp)
// do not attempt to send MDNs if ratelimited happend before on status-updates/sync: // do not attempt to send MDNs if ratelimited happend before on status-updates/sync:
// instead, let the caller recall this function so that more important status-updates/sync are sent out. // instead, let the caller recall this function so that more important status-updates/sync are sent out.
if !ratelimited { if !ratelimited {
ratelimited = send_mdns(context, connection) send_mdns(context, connection)
.await .await
.context("failed to send MDNs")?; .context("failed to send MDNs")?;
} }
Ok(())
Ok(ratelimited)
} }
/// Tries to send MDN for message `msg_id` to `contact_id`. /// Tries to send MDN for message `msg_id` to `contact_id`.