From 891ec635fa9a9ad2d080f3d223245dad7f3fdc52 Mon Sep 17 00:00:00 2001 From: Wez Furlong Date: Wed, 20 Nov 2024 16:00:17 -0700 Subject: [PATCH] refactor: replace increment_attempts bool with an enum This improves grokability, and has no functional change. --- crates/kumod/src/lua_deliver.rs | 8 +++++- crates/kumod/src/queue.rs | 21 ++++++++++++---- crates/kumod/src/ready_queue.rs | 38 ++++++++++++++++++----------- crates/kumod/src/smtp_dispatcher.rs | 19 ++++++++++++--- 4 files changed, 62 insertions(+), 24 deletions(-) diff --git a/crates/kumod/src/lua_deliver.rs b/crates/kumod/src/lua_deliver.rs index 9186ebae..686343e2 100644 --- a/crates/kumod/src/lua_deliver.rs +++ b/crates/kumod/src/lua_deliver.rs @@ -1,5 +1,6 @@ use crate::delivery_metrics::MetricsWrappedConnection; use crate::logging::disposition::{log_disposition, LogDisposition}; +use crate::queue::IncrementAttempts; use crate::ready_queue::{Dispatcher, QueueDispatcher}; use crate::smtp_server::RejectError; use crate::spool::SpoolManager; @@ -243,7 +244,12 @@ impl QueueDispatcher for LuaQueueDispatcher { .await; spawn_local( "requeue message".to_string(), - Dispatcher::requeue_message(msg, true, None, response.clone()), + Dispatcher::requeue_message( + msg, + IncrementAttempts::Yes, + None, + response.clone(), + ), )?; dispatcher.metrics.inc_transfail(); } diff --git a/crates/kumod/src/queue.rs b/crates/kumod/src/queue.rs index 1685ccc2..d34b3439 100644 --- a/crates/kumod/src/queue.rs +++ b/crates/kumod/src/queue.rs @@ -1233,13 +1233,15 @@ impl Queue { } } - let increment_attempts = false; let mut delay = None; let queue_name = match msg.get_queue_name() { Err(err) => { tracing::error!("failed to determine queue name for msg: {err:#}"); - if let Err(err) = self.requeue_message(msg, increment_attempts, delay).await { + if let Err(err) = self + .requeue_message(msg, IncrementAttempts::No, delay) + .await + { tracing::error!( "failed to requeue message to {} after failed rebind: {err:#}", self.name @@ -1299,7 +1301,10 @@ impl Queue { .await; } - if let Err(err) = queue.requeue_message(msg, increment_attempts, delay).await { + if let Err(err) = queue + .requeue_message(msg, IncrementAttempts::No, delay) + .await + { tracing::error!( "failed to requeue message to {} after failed rebind: {err:#}", queue.name @@ -1414,10 +1419,10 @@ impl Queue { pub async fn requeue_message( &self, msg: Message, - increment_attempts: bool, + increment_attempts: IncrementAttempts, delay: Option, ) -> anyhow::Result<()> { - if increment_attempts { + if increment_attempts == IncrementAttempts::Yes { match self.increment_attempts_and_update_delay(msg).await? { Some(msg) => { return self.insert(msg).await; @@ -2262,3 +2267,9 @@ async fn maintain_named_queue(q: &QueueHandle) -> anyhow::Result<()> { } } } + +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum IncrementAttempts { + No, + Yes, +} diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index c11802df..cc2246f3 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/crates/kumod/src/ready_queue.rs @@ -9,7 +9,9 @@ use crate::http_server::inject_v1::HttpInjectionGeneratorDispatcher; use crate::logging::disposition::{log_disposition, LogDisposition, RecordType}; use crate::lua_deliver::LuaQueueDispatcher; use crate::metrics_helper::TOTAL_READYQ_RUNS; -use crate::queue::{DeliveryProto, Queue, QueueConfig, QueueManager, QMAINT_RUNTIME}; +use crate::queue::{ + DeliveryProto, IncrementAttempts, Queue, QueueConfig, QueueManager, QMAINT_RUNTIME, +}; use crate::smtp_dispatcher::{MxListEntry, OpportunisticInsecureTlsHandshakeError, SmtpDispatcher}; use crate::smtp_server::RejectError; use crate::spool::SpoolManager; @@ -967,8 +969,13 @@ impl Drop for Dispatcher { command: None, }; - if let Err(err) = - Dispatcher::requeue_message(msg, false, None, response).await + if let Err(err) = Dispatcher::requeue_message( + msg, + IncrementAttempts::No, + None, + response, + ) + .await { tracing::error!("error requeuing message on Drop: {err:#}"); } @@ -1146,7 +1153,8 @@ impl Dispatcher { session_id: Some(dispatcher.session_id), }) .await; - Dispatcher::requeue_message(msg, true, None, response).await?; + Dispatcher::requeue_message(msg, IncrementAttempts::Yes, None, response) + .await?; dispatcher.metrics.inc_transfail(); } @@ -1298,7 +1306,7 @@ impl Dispatcher { #[instrument(skip(msg))] pub async fn requeue_message( msg: Message, - mut increment_attempts: bool, + mut increment_attempts: IncrementAttempts, mut delay: Option, response: Response, ) -> anyhow::Result<()> { @@ -1313,7 +1321,7 @@ impl Dispatcher { // to an alternative scheduled queue. // Moving to another queue will make the message immediately eligible // for delivery in that new queue. - if increment_attempts { + if increment_attempts == IncrementAttempts::Yes { match load_config().await { Ok(mut config) => { let result: anyhow::Result<()> = config @@ -1332,7 +1340,7 @@ impl Dispatcher { // in Queue::requeue_message, but we still want the // number to be incremented. msg.increment_num_attempts(); - increment_attempts = false; + increment_attempts = IncrementAttempts::No; // Avoid adding jitter as part of the queue change delay = Some(chrono::Duration::zero()); @@ -1471,9 +1479,13 @@ impl Dispatcher { command: None, }; for msg in msgs { - if let Err(err) = - Self::requeue_message(msg, false, Some(delay), response.clone()) - .await + if let Err(err) = Self::requeue_message( + msg, + IncrementAttempts::No, + Some(delay), + response.clone(), + ) + .await { tracing::error!("error requeuing message for throttle: {err:#}"); } @@ -1499,7 +1511,6 @@ impl Dispatcher { self.metrics.inc_fail_by(msgs.len()); } let path_config = self.path_config.borrow(); - let increment_attempts = true; for msg in msgs { log_disposition(LogDisposition { kind: if response.is_transient() { @@ -1524,7 +1535,7 @@ impl Dispatcher { if response.is_transient() { if let Err(err) = - Self::requeue_message(msg, increment_attempts, None, response.clone()).await + Self::requeue_message(msg, IncrementAttempts::Yes, None, response.clone()).await { tracing::error!("error requeuing for bulk {} operation: {err:#}", self.name); } @@ -1612,8 +1623,7 @@ impl Dispatcher { }) .await; - let increment_attempts = true; - Self::requeue_message(msg, increment_attempts, None, response) + Self::requeue_message(msg, IncrementAttempts::Yes, None, response) .await .ok(); continue; diff --git a/crates/kumod/src/smtp_dispatcher.rs b/crates/kumod/src/smtp_dispatcher.rs index 5e6eca98..41d95d93 100644 --- a/crates/kumod/src/smtp_dispatcher.rs +++ b/crates/kumod/src/smtp_dispatcher.rs @@ -3,6 +3,7 @@ use crate::http_server::admin_trace_smtp_client_v1::{ SmtpClientTraceEventPayload, SmtpClientTracerImpl, }; use crate::logging::disposition::{log_disposition, LogDisposition, RecordType}; +use crate::queue::IncrementAttempts; use crate::ready_queue::{Dispatcher, QueueDispatcher}; use crate::spool::SpoolManager; use anyhow::Context; @@ -804,7 +805,12 @@ impl QueueDispatcher for SmtpDispatcher { .await; spawn_local( "requeue message".to_string(), - Dispatcher::requeue_message(msg, true, None, response), + Dispatcher::requeue_message( + msg, + IncrementAttempts::Yes, + None, + response, + ), )?; } dispatcher.metrics.inc_transfail(); @@ -827,7 +833,12 @@ impl QueueDispatcher for SmtpDispatcher { .await; spawn_local( "requeue message".to_string(), - Dispatcher::requeue_message(msg, true, None, response), + Dispatcher::requeue_message( + msg, + IncrementAttempts::Yes, + None, + response, + ), )?; } dispatcher.metrics.inc_transfail(); @@ -886,7 +897,7 @@ impl QueueDispatcher for SmtpDispatcher { .await; spawn_local( "requeue message".to_string(), - Dispatcher::requeue_message(msg, true, None, response), + Dispatcher::requeue_message(msg, IncrementAttempts::Yes, None, response), )?; } dispatcher.metrics.inc_transfail(); @@ -922,7 +933,7 @@ impl QueueDispatcher for SmtpDispatcher { .await; spawn_local( "requeue message".to_string(), - Dispatcher::requeue_message(msg, true, None, response), + Dispatcher::requeue_message(msg, IncrementAttempts::Yes, None, response), )?; } dispatcher.metrics.inc_transfail();