mirror/kumomta
1
0
Fork 0
mirror of https://github.com/mailscope/kumomta.git synced 2026-10-26 00:12:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: replace increment_attempts bool with an enum

Browse Source 
This improves grokability, and has no functional change.
This commit is contained in:
Wez Furlong
2024-11-20 16:00:17 -07:00
parent 7761b1d8e9
commit 891ec635fa
4 changed files with 62 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/kumod/src/lua_deliver.rs
+7 -1
View File
@@ -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();
}
crates/kumod/src/queue.rs
+16 -5
View File
@@ -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<chrono::Duration>,
) -> 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,
}
crates/kumod/src/ready_queue.rs
+24 -14
View File
@@ -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<chrono::Duration>,
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;
crates/kumod/src/smtp_dispatcher.rs
+15 -4
View File
@@ -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();