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

queue: route most ready-queue-insertion errors via requeue_message event

Browse Source 
Previously, we would only trigger the requeue_message event in
situations where we were actively working on talking to the destination.
That left issues such as persistently NXDOMAIN destinations as being
unable to be caught and handled by the requeue_message event, which is
an issue for sites that want to fail out messages from the queue that
have bogus domains before they reach max_age.

This commit replaces all but one of the force_into_delayed calls with
requeue_message, and makes the call out to the event unconditional
(rather than dependent upon whether we were incrementing retries or
not).

The only case now that doesn't cause requeue_message to fire is when the
ready queue is full. The rationale is that that is a transient local
resource issue (rather than some external factor to which we need to
react), and that is likely to be a hot event when it triggers, so we
don't want to add CPU pressure with calling out to the requeue event for
them.

refs: https://github.com/KumoCorp/kumomta/issues/319
This commit is contained in:
Wez Furlong
2024-12-03 13:01:42 -07:00
parent 4a0a4d6a1e
commit e5127cb24b
2 changed files with 158 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/kumod/src/queue.rs
+152 -108
View File
@@ -1591,24 +1591,26 @@ impl Queue {
let remaining = suspend.get_duration();
tracing::trace!("{} is suspended, delay={remaining:?}", self.name);
let response = Response {
code: 451,
enhanced_code: Some(rfc5321::EnhancedStatusCode {
class: 4,
subject: 4,
detail: 4,
}),
content: format!(
"KumoMTA internal: scheduled queue is suspended: {}",
suspend.reason
),
command: None,
};
log_disposition(LogDisposition {
kind: RecordType::TransientFailure,
msg: msg.clone(),
site: &self.name,
peer_address: None,
response: rfc5321::Response {
code: 451,
enhanced_code: Some(rfc5321::EnhancedStatusCode {
class: 4,
subject: 4,
detail: 4,
}),
content: format!(
"KumoMTA internal: scheduled queue is suspended: {}",
suspend.reason
),
command: None,
},
response: response.clone(),
egress_source: None,
egress_pool: None,
relay_disposition: None,
@@ -1620,29 +1622,48 @@ impl Queue {
})
.await;
match self.increment_attempts_and_update_delay(msg).await? {
Some(msg) => {
self.force_into_delayed(msg).await?;
}
None => {
// It was expired and removed from the spool
}
}
Box::pin(QueueManager::requeue_message(
msg,
IncrementAttempts::Yes,
None,
response,
))
.await?;
return Ok(());
}
if let Some(result) = self.check_message_rate_throttle().await? {
if let Some(delay) = result.retry_after {
tracing::trace!("{} throttled message rate, delay={delay:?}", self.name);
let delay = chrono::Duration::from_std(delay).unwrap_or(kumo_chrono_helper::MINUTE);
// We're not using jitter here because the throttle should
// ideally result in smooth message flow and the jitter will
// (intentionally) perturb that.
msg.delay_by(delay).await?;
let delay = chrono::Duration::from_std(delay).unwrap_or(kumo_chrono_helper::MINUTE);
Box::pin(QueueManager::requeue_message(
msg,
IncrementAttempts::No,
Some(delay),
Response {
code: 451,
enhanced_code: Some(EnhancedStatusCode {
class: 4,
subject: 4,
detail: 4,
}),
content: format!(
"KumoMTA internal: {} throttled message rate, delay={delay:?}",
self.name
),
command: None,
},
))
.await?;
self.metrics().delay_due_to_message_rate_throttle().inc();
return self.force_into_delayed(msg).await;
return Ok(());
}
}
@@ -1659,7 +1680,26 @@ impl Queue {
);
self.metrics().delay_due_to_throttle_insert_ready().inc();
return self.force_into_delayed(msg).await;
Box::pin(QueueManager::requeue_message(
msg,
IncrementAttempts::No,
None,
Response {
code: 451,
enhanced_code: Some(EnhancedStatusCode {
class: 4,
subject: 4,
detail: 4,
}),
content: format!(
"KumoMTA internal: {} throttle_insert_ready_queue event throttled message rate, due={due:?}",
self.name
),
command: None,
},
))
.await?;
return Ok(());
}
}
@@ -1668,14 +1708,26 @@ impl Queue {
if err.downcast_ref::<ReadyQueueFull>().is_none() {
// It was a legit error while trying to do something useful
match self.increment_attempts_and_update_delay(msg).await? {
Some(msg) => {
self.force_into_delayed(msg).await?;
}
None => {
// It was expired and removed from the spool
}
}
Box::pin(QueueManager::requeue_message(
msg,
IncrementAttempts::Yes,
None,
Response {
code: 451,
enhanced_code: Some(EnhancedStatusCode {
class: 4,
subject: 4,
detail: 4,
}),
content: format!(
"KumoMTA internal: {} error while inserting into ready queue: {err:#}",
self.name
),
command: None,
},
))
.await?;
} else {
// Queue is full; try again shortly
self.metrics().delay_due_to_ready_queue_full().inc();
@@ -2205,89 +2257,81 @@ impl QueueManager {
}
let mut queue_name = msg.get_queue_name()?;
// When increment_attempts is true, the intent is to handle a transient
// failure for this message. In that circumstance we want to allow
// the requeue_message event the opportunity to rebind the message
// to an alternative scheduled queue.
// Moving to another queue will make the message immediately eligible
// for delivery in that new queue.
if increment_attempts == IncrementAttempts::Yes {
match load_config().await {
Ok(mut config) => {
let result: anyhow::Result<()> = config
.async_call_callback(
&REQUEUE_MESSAGE_SIG,
(msg.clone(), response.to_single_line()),
)
.await;
match load_config().await {
Ok(mut config) => {
let result: anyhow::Result<()> = config
.async_call_callback(
&REQUEUE_MESSAGE_SIG,
(msg.clone(), response.to_single_line()),
)
.await;
match result {
Ok(()) => {
let queue_name_after = msg.get_queue_name()?;
if queue_name != queue_name_after {
// We want to avoid the normal due-time adjustment
// that would kick in when incrementing attempts
// in Queue::requeue_message, but we still want the
// number to be incremented.
msg.increment_num_attempts();
increment_attempts = IncrementAttempts::No;
match result {
Ok(()) => {
let queue_name_after = msg.get_queue_name()?;
if queue_name != queue_name_after {
// We want to avoid the normal due-time adjustment
// that would kick in when incrementing attempts
// in Queue::requeue_message, but we still want the
// number to be incremented.
msg.increment_num_attempts();
increment_attempts = IncrementAttempts::No;
// Avoid adding jitter as part of the queue change
delay = Some(chrono::Duration::zero());
// and ensure that the message is due now
msg.set_due(None).await?;
// Avoid adding jitter as part of the queue change
delay = Some(chrono::Duration::zero());
// and ensure that the message is due now
msg.set_due(None).await?;
// and use the new queue name
queue_name = queue_name_after;
}
}
Err(err) => {
// If they did a kumo.reject() in the handler, translate that
// into a Bounce. We do this even if they used a 4xx code; it
// only makes sense to map it to a Bounce rather than a
// TransientFailure because we already just had a TransientFailure.
if let Some(rej) = RejectError::from_anyhow(&err) {
log_disposition(LogDisposition {
kind: RecordType::Bounce,
msg: msg.clone(),
// There is no site because this was a policy bounce
// triggered in an event handler
site: "",
peer_address: None,
response: Response {
code: rej.code,
enhanced_code: None,
content: rej.message,
command: None,
},
egress_pool: None,
egress_source: None,
relay_disposition: None,
delivery_protocol: None,
tls_info: None,
source_address: None,
provider: None,
session_id: None,
})
.await;
SpoolManager::remove_from_spool(*msg.id()).await.ok();
return Ok(());
}
tracing::error!(
"Error while calling requeue_message event: {err:#}. \
will reuse current queue"
);
// and use the new queue name
queue_name = queue_name_after;
}
}
Err(err) => {
// If they did a kumo.reject() in the handler, translate that
// into a Bounce. We do this even if they used a 4xx code; it
// only makes sense to map it to a Bounce rather than a
// TransientFailure because we already just had a TransientFailure.
if let Some(rej) = RejectError::from_anyhow(&err) {
log_disposition(LogDisposition {
kind: RecordType::Bounce,
msg: msg.clone(),
// There is no site because this was a policy bounce
// triggered in an event handler
site: "",
peer_address: None,
response: Response {
code: rej.code,
enhanced_code: None,
content: rej.message,
command: None,
},
egress_pool: None,
egress_source: None,
relay_disposition: None,
delivery_protocol: None,
tls_info: None,
source_address: None,
provider: None,
session_id: None,
})
.await;
SpoolManager::remove_from_spool(*msg.id()).await.ok();
return Ok(());
}
tracing::error!(
"Error while calling requeue_message event: {err:#}. \
will reuse current queue"
);
}
}
Err(err) => {
tracing::error!(
"ReadyQueue::requeue_message: error getting \
}
Err(err) => {
tracing::error!(
"ReadyQueue::requeue_message: error getting \
lua config in order to call requeue_message event: \
{err:#}, will reuse current queue"
);
}
);
}
}
docs/changelog/main.md
+6
View File
@@ -15,6 +15,12 @@
we now display it as `some.host./10.0.0.1` which is a bit easier to
understand and occupies less space in the logs.
* We will now trigger the
[requeue_message](../reference/events/requeue_message.md) event in the
case of an error resolving the ready queue, such as DNS related errors.
This gives an opportunity to rebind or reject messages which are
experiencing persistent DNS resolution issues. #319
## Fixes
* When `enable_tls` is set to `Required` or `RequiredInsecure`, ignore the