filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Project Alouette: je te plume la queue (et je la remplace par mieux)

Browse Source
This commit is contained in:
mdecimus
2025-07-09 20:38:27 +02:00
parent 698ce0de2e
commit d04a4331c3
83 changed files with 3479 additions and 2987 deletions
Download Patch File Download Diff File Expand all files Collapse all files

413
crates/smtp/src/queue/dsn.rs
View File

@@ -4,8 +4,14 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use super::spool::SmtpSpool;
use super::{
Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope, RCPT_DSN_SENT,
RCPT_STATUS_CHANGED, Recipient, Status,
};
use crate::queue::{MessageWrapper, UnexpectedResponse};
use crate::reporting::SmtpReporting;
use common::Server;
use mail_builder::MessageBuilder;
use mail_builder::headers::HeaderType;
use mail_builder::headers::content_type::ContentType;
@@ -16,37 +22,26 @@ use smtp_proto::{
};
use std::fmt::Write;
use std::future::Future;
use std::time::Duration;
use store::write::now;
use crate::outbound::client::from_error_status;
use crate::reporting::SmtpReporting;
use super::spool::SmtpSpool;
use super::{
Domain, Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope,
RCPT_DSN_SENT, RCPT_STATUS_CHANGED, Recipient, Status,
};
pub trait SendDsn: Sync + Send {
fn send_dsn(&self, message: &mut Message) -> impl Future<Output = ()> + Send;
fn log_dsn(&self, message: &Message) -> impl Future<Output = ()> + Send;
fn send_dsn(&self, message: &mut MessageWrapper) -> impl Future<Output = ()> + Send;
fn log_dsn(&self, message: &MessageWrapper) -> impl Future<Output = ()> + Send;
}
impl SendDsn for Server {
async fn send_dsn(&self, message: &mut Message) {
async fn send_dsn(&self, message: &mut MessageWrapper) {
// Send DSN events
self.log_dsn(message).await;
if !message.return_path.is_empty() {
if !message.message.return_path.is_empty() {
// Build DSN
if let Some(dsn) = message.build_dsn(self).await {
let mut dsn_message = self.new_message("", "", "", message.span_id);
dsn_message
.add_recipient_parts(
message.return_path.as_str(),
message.return_path_lcase.as_str(),
message.return_path_domain.as_str(),
message.message.return_path.as_str(),
message.message.return_path_lcase.as_str(),
self,
)
.await;
@@ -73,15 +68,14 @@ impl SendDsn for Server {
}
}
async fn log_dsn(&self, message: &Message) {
async fn log_dsn(&self, message: &MessageWrapper) {
let now = now();
for rcpt in &message.recipients {
for rcpt in &message.message.recipients {
if rcpt.has_flag(RCPT_DSN_SENT) {
continue;
}
let domain = &message.domains[rcpt.domain_idx as usize];
match &rcpt.status {
Status::Completed(response) => {
trc::event!(
@@ -93,17 +87,18 @@ impl SendDsn for Server {
Details = response.response.message.to_string(),
);
}
Status::TemporaryFailure(response) if domain.notify.due <= now => {
Status::TemporaryFailure(response) if rcpt.notify.due <= now => {
trc::event!(
Delivery(trc::DeliveryEvent::DsnTempFail),
SpanId = message.span_id,
To = rcpt.address_lcase.clone(),
Hostname = response.hostname.entity.clone(),
Code = response.response.code,
Details = response.response.message.to_string(),
NextRetry = trc::Value::Timestamp(domain.retry.due),
Expires = trc::Value::Timestamp(domain.expires),
Total = domain.retry.inner,
Hostname = response.entity.clone(),
Details = response.details.to_string(),
NextRetry = trc::Value::Timestamp(rcpt.retry.due),
Expires = rcpt
.expiration_time(message.message.created)
.map(trc::Value::Timestamp),
Total = rcpt.retry.inner,
);
}
Status::PermanentFailure(response) => {
@@ -111,48 +106,23 @@ impl SendDsn for Server {
Delivery(trc::DeliveryEvent::DsnPermFail),
SpanId = message.span_id,
To = rcpt.address_lcase.clone(),
Hostname = response.hostname.entity.clone(),
Code = response.response.code,
Details = response.response.message.to_string(),
Total = domain.retry.inner,
Hostname = response.entity.clone(),
Details = response.details.to_string(),
Total = rcpt.retry.inner,
);
}
Status::Scheduled => {
// There is no status for this address, use the domain's status.
match &domain.status {
Status::PermanentFailure(_) => {
trc::event!(
Delivery(trc::DeliveryEvent::DsnPermFail),
SpanId = message.span_id,
To = rcpt.address_lcase.clone(),
Details = from_error_status(&domain.status),
Total = domain.retry.inner,
);
}
Status::TemporaryFailure(_) if domain.notify.due <= now => {
trc::event!(
Delivery(trc::DeliveryEvent::DsnTempFail),
SpanId = message.span_id,
To = rcpt.address_lcase.clone(),
Details = from_error_status(&domain.status),
NextRetry = trc::Value::Timestamp(domain.retry.due),
Expires = trc::Value::Timestamp(domain.expires),
Total = domain.retry.inner,
);
}
Status::Scheduled if domain.notify.due <= now => {
trc::event!(
Delivery(trc::DeliveryEvent::DsnTempFail),
SpanId = message.span_id,
To = rcpt.address_lcase.clone(),
Details = "Concurrency limited",
NextRetry = trc::Value::Timestamp(domain.retry.due),
Expires = trc::Value::Timestamp(domain.expires),
Total = domain.retry.inner,
);
}
_ => continue,
}
Status::Scheduled if rcpt.notify.due <= now => {
trc::event!(
Delivery(trc::DeliveryEvent::DsnTempFail),
SpanId = message.span_id,
To = rcpt.address_lcase.clone(),
Details = "Concurrency limited",
NextRetry = trc::Value::Timestamp(rcpt.retry.due),
Expires = rcpt
.expiration_time(message.message.created)
.map(trc::Value::Timestamp),
Total = rcpt.retry.inner,
);
}
_ => continue,
}
@@ -160,7 +130,7 @@ impl SendDsn for Server {
}
}
impl Message {
impl MessageWrapper {
pub async fn build_dsn(&mut self, server: &Server) -> Option<Vec<u8>> {
let config = &server.core.smtp.queue;
let now = now();
@@ -170,11 +140,10 @@ impl Message {
let mut txt_failed = String::new();
let mut dsn = String::new();
for rcpt in &mut self.recipients {
for rcpt in &mut self.message.recipients {
if rcpt.has_flag(RCPT_DSN_SENT | RCPT_NOTIFY_NEVER) {
continue;
}
let domain = &self.domains[rcpt.domain_idx as usize];
match &rcpt.status {
Status::Completed(response) => {
rcpt.flags |= RCPT_DSN_SENT | RCPT_STATUS_CHANGED;
@@ -186,11 +155,11 @@ impl Message {
response.write_dsn_text(&rcpt.address, &mut txt_success);
}
Status::TemporaryFailure(response)
if domain.notify.due <= now && rcpt.has_flag(RCPT_NOTIFY_DELAY) =>
if rcpt.notify.due <= now && rcpt.has_flag(RCPT_NOTIFY_DELAY) =>
{
rcpt.write_dsn(&mut dsn);
rcpt.status.write_dsn(&mut dsn);
domain.write_dsn_will_retry_until(&mut dsn);
rcpt.write_dsn_will_retry_until(self.message.created, &mut dsn);
response.write_dsn_text(&rcpt.address, &mut txt_delay);
}
Status::PermanentFailure(response) => {
@@ -202,45 +171,16 @@ impl Message {
rcpt.status.write_dsn(&mut dsn);
response.write_dsn_text(&rcpt.address, &mut txt_failed);
}
Status::Scheduled => {
// There is no status for this address, use the domain's status.
match &domain.status {
Status::PermanentFailure(err) => {
rcpt.flags |= RCPT_DSN_SENT | RCPT_STATUS_CHANGED;
if !rcpt.has_flag(RCPT_NOTIFY_FAILURE) {
continue;
}
rcpt.write_dsn(&mut dsn);
domain.status.write_dsn(&mut dsn);
err.write_dsn_text(&rcpt.address, &domain.domain, &mut txt_failed);
}
Status::TemporaryFailure(err)
if domain.notify.due <= now && rcpt.has_flag(RCPT_NOTIFY_DELAY) =>
{
rcpt.write_dsn(&mut dsn);
domain.status.write_dsn(&mut dsn);
domain.write_dsn_will_retry_until(&mut dsn);
err.write_dsn_text(&rcpt.address, &domain.domain, &mut txt_delay);
}
Status::Scheduled
if domain.notify.due <= now && rcpt.has_flag(RCPT_NOTIFY_DELAY) =>
{
// This case should not happen under normal circumstances
rcpt.write_dsn(&mut dsn);
domain.status.write_dsn(&mut dsn);
domain.write_dsn_will_retry_until(&mut dsn);
Error::ConcurrencyLimited.write_dsn_text(
&rcpt.address,
&domain.domain,
&mut txt_delay,
);
}
Status::Completed(_) => {
#[cfg(feature = "test_mode")]
panic!("This should not have happened.");
}
_ => continue,
Status::Scheduled if rcpt.notify.due <= now && rcpt.has_flag(RCPT_NOTIFY_DELAY) => {
// This case should not happen under normal circumstances
rcpt.write_dsn(&mut dsn);
rcpt.status.write_dsn(&mut dsn);
rcpt.write_dsn_will_retry_until(self.message.created, &mut dsn);
ErrorDetails {
entity: "localhost".into(),
details: Error::ConcurrencyLimited,
}
.write_dsn_text(&rcpt.address, &mut txt_delay);
}
_ => continue,
}
@@ -315,58 +255,69 @@ impl Message {
// Update next delay notification time
if has_delay {
let mut changes = Vec::new();
for (domain_idx, domain) in self.domains.iter().enumerate() {
for (rcpt_idx, rcpt) in self.message.recipients.iter().enumerate() {
if matches!(
&domain.status,
&rcpt.status,
Status::TemporaryFailure(_) | Status::Scheduled
) && domain.notify.due <= now
) && rcpt.notify.due <= now
{
let envelope = QueueEnvelope::new(self, domain_idx);
let envelope = QueueEnvelope::new_rcpt(&self.message, rcpt_idx);
if let Some(next_notify) = server
.eval_if::<Vec<Duration>, _>(&config.notify, &envelope, self.span_id)
let queue_id = server
.eval_if::<String, _>(
&server.core.smtp.queue.queue,
&envelope,
self.span_id,
)
.await
.and_then(|notify| {
notify.into_iter().nth((domain.notify.inner + 1) as usize)
})
.unwrap_or_else(|| "default".to_string());
let queue = server.get_queue_or_default(&queue_id, self.span_id);
if let Some(next_notify) =
queue.notify.get((rcpt.notify.inner + 1) as usize).copied()
{
changes.push((domain_idx, 1, now + next_notify.as_secs()));
changes.push((rcpt_idx, 1, now + next_notify));
} else {
changes.push((domain_idx, 0, domain.expires + 10));
changes.push((rcpt_idx, 0, u64::MAX));
}
}
}
for (domain_idx, inner, due) in changes {
let domain = &mut self.domains[domain_idx];
domain.notify.inner += inner;
domain.notify.due = due;
for (rcpt_idx, inner, due) in changes {
let rcpt = &mut self.message.recipients[rcpt_idx];
rcpt.notify.inner += inner;
rcpt.notify.due = due;
}
}
// Obtain hostname and sender addresses
let from_name = server
.eval_if(&config.dsn.name, self, self.span_id)
.eval_if(&config.dsn.name, &self.message, self.span_id)
.await
.unwrap_or_else(|| String::from("Mail Delivery Subsystem"));
let from_addr = server
.eval_if(&config.dsn.address, self, self.span_id)
.eval_if(&config.dsn.address, &self.message, self.span_id)
.await
.unwrap_or_else(|| String::from("MAILER-DAEMON@localhost"));
let reporting_mta = server
.eval_if(&server.core.smtp.report.submitter, self, self.span_id)
.eval_if(
&server.core.smtp.report.submitter,
&self.message,
self.span_id,
)
.await
.unwrap_or_else(|| String::from("localhost"));
// Prepare DSN
let mut dsn_header = String::with_capacity(dsn.len() + 128);
self.write_dsn_headers(&mut dsn_header, &reporting_mta);
self.message
.write_dsn_headers(&mut dsn_header, &reporting_mta);
let dsn = dsn_header + dsn.as_str();
// Fetch up to 1024 bytes of message headers
let headers = match server
.blob_store()
.get_blob(self.blob_hash.as_slice(), 0..1024)
.get_blob(self.message.blob_hash.as_slice(), 0..1024)
.await
{
Ok(Some(mut buf)) => {
@@ -398,7 +349,7 @@ impl Message {
trc::event!(
Queue(trc::QueueEvent::BlobNotFound),
SpanId = self.span_id,
BlobId = self.blob_hash.to_hex(),
BlobId = self.message.blob_hash.to_hex(),
CausedBy = trc::location!()
);
@@ -418,7 +369,10 @@ impl Message {
// Build message
MessageBuilder::new()
.from((from_name.as_str(), from_addr.as_str()))
.header("To", HeaderType::Text(self.return_path.as_str().into()))
.header(
"To",
HeaderType::Text(self.message.return_path.as_str().into()),
)
.header("Auto-Submitted", HeaderType::Text("auto-generated".into()))
.message_id(format!("<{}@{}>", make_boundary("."), reporting_mta))
.subject(subject)
@@ -443,34 +397,23 @@ impl Message {
fn handle_double_bounce(&mut self) {
let mut is_double_bounce = Vec::with_capacity(0);
let now = now();
for rcpt in &mut self.recipients {
for rcpt in &mut self.message.recipients {
if !rcpt.has_flag(RCPT_DSN_SENT | RCPT_NOTIFY_NEVER) {
match &rcpt.status {
Status::PermanentFailure(err) => {
rcpt.flags |= RCPT_DSN_SENT;
let mut dsn = String::new();
err.write_dsn_text(&rcpt.address, &mut dsn);
is_double_bounce.push(dsn);
}
Status::Scheduled => {
let domain = &self.domains[rcpt.domain_idx as usize];
if let Status::PermanentFailure(err) = &domain.status {
rcpt.flags |= RCPT_DSN_SENT;
let mut dsn = String::new();
err.write_dsn_text(&rcpt.address, &domain.domain, &mut dsn);
is_double_bounce.push(dsn);
}
}
_ => (),
if let Status::PermanentFailure(err) = &rcpt.status {
rcpt.flags |= RCPT_DSN_SENT;
let mut dsn = String::new();
err.write_dsn_text(&rcpt.address, &mut dsn);
is_double_bounce.push(dsn);
}
}
}
let now = now();
for domain in &mut self.domains {
if domain.notify.due <= now {
domain.notify.due = domain.expires + 10;
if rcpt.notify.due <= now {
rcpt.notify.due = rcpt
.expiration_time(self.message.created)
.map(|d| d + 10)
.unwrap_or(u64::MAX);
}
}
@@ -501,12 +444,12 @@ impl HostResponse<String> {
}
}
impl HostResponse<ErrorDetails> {
fn write_dsn_text(&self, addr: &str, dsn: &mut String) {
let _ = write!(dsn, "<{}> (host '{}' rejected ", addr, self.hostname.entity);
impl UnexpectedResponse {
fn write_dsn_text(&self, host: &str, addr: &str, dsn: &mut String) {
let _ = write!(dsn, "<{addr}> (host '{host}' rejected ");
if !self.hostname.details.is_empty() {
let _ = write!(dsn, "command '{}'", self.hostname.details,);
if !self.command.is_empty() {
let _ = write!(dsn, "command '{}'", self.command);
} else {
dsn.push_str("transaction");
}
@@ -521,40 +464,35 @@ impl HostResponse<ErrorDetails> {
}
}
impl Error {
fn write_dsn_text(&self, addr: &str, domain: &str, dsn: &mut String) {
match self {
impl ErrorDetails {
fn write_dsn_text(&self, addr: &str, dsn: &mut String) {
let entity = self.entity.as_str();
match &self.details {
Error::UnexpectedResponse(response) => {
response.write_dsn_text(addr, dsn);
response.write_dsn_text(entity, addr, dsn);
}
Error::DnsError(err) => {
let _ = write!(dsn, "<{addr}> (failed to lookup '{domain}': {err})\r\n",);
let _ = write!(dsn, "<{addr}> (failed to lookup '{entity}': {err})\r\n",);
}
Error::ConnectionError(details) => {
let _ = write!(
dsn,
"<{}> (connection to '{}' failed: {})\r\n",
addr, details.entity, details.details
"<{addr}> (connection to '{entity}' failed: {details})\r\n",
);
}
Error::TlsError(details) => {
let _ = write!(
dsn,
"<{}> (TLS error from '{}': {})\r\n",
addr, details.entity, details.details
);
let _ = write!(dsn, "<{addr}> (TLS error from '{entity}': {details})\r\n",);
}
Error::DaneError(details) => {
let _ = write!(
dsn,
"<{}> (DANE failed to authenticate '{}': {})\r\n",
addr, details.entity, details.details
"<{addr}> (DANE failed to authenticate '{entity}': {details})\r\n",
);
}
Error::MtaStsError(details) => {
let _ = write!(
dsn,
"<{addr}> (MTA-STS failed to authenticate '{domain}': {details})\r\n",
"<{addr}> (MTA-STS failed to authenticate '{entity}': {details})\r\n",
);
}
Error::RateLimited => {
@@ -593,15 +531,14 @@ impl Recipient {
}
let _ = write!(dsn, "Final-Recipient: rfc822;{}\r\n", self.address);
}
}
impl Domain {
fn write_dsn_will_retry_until(&self, dsn: &mut String) {
let now = now();
if self.expires > now {
dsn.push_str("Will-Retry-Until: ");
dsn.push_str(&DateTime::from_timestamp(self.expires as i64).to_rfc822());
dsn.push_str("\r\n");
fn write_dsn_will_retry_until(&self, created: u64, dsn: &mut String) {
if let Some(expires) = self.expiration_time(created) {
if expires > now() {
dsn.push_str("Will-Retry-Until: ");
dsn.push_str(&DateTime::from_timestamp(expires as i64).to_rfc822());
dsn.push_str("\r\n");
}
}
}
}
@@ -636,7 +573,7 @@ impl<T, E> Status<T, E> {
}
}
impl Status<HostResponse<String>, HostResponse<ErrorDetails>> {
impl Status<HostResponse<String>, ErrorDetails> {
fn write_dsn(&self, dsn: &mut String) {
self.write_dsn_action(dsn);
self.write_dsn_status(dsn);
@@ -646,95 +583,55 @@ impl Status<HostResponse<String>, HostResponse<ErrorDetails>> {
fn write_dsn_status(&self, dsn: &mut String) {
dsn.push_str("Status: ");
if let Status::Completed(HostResponse { response, .. })
| Status::PermanentFailure(HostResponse { response, .. })
| Status::TemporaryFailure(HostResponse { response, .. }) = self
{
response.write_dsn_status(dsn);
}
dsn.push_str("\r\n");
}
fn write_dsn_remote_mta(&self, dsn: &mut String) {
dsn.push_str("Remote-MTA: dns;");
match self {
Status::Completed(HostResponse { hostname, .. }) => {
dsn.push_str(hostname);
Status::Completed(response) => {
response.response.write_dsn_status(dsn);
}
Status::PermanentFailure(HostResponse {
hostname: ErrorDetails {
entity: hostname, ..
},
..
})
| Status::TemporaryFailure(HostResponse {
hostname: ErrorDetails {
entity: hostname, ..
},
..
}) => {
dsn.push_str(hostname);
Status::TemporaryFailure(err) | Status::PermanentFailure(err) => {
if let Error::UnexpectedResponse(response) = &err.details {
response.response.write_dsn_status(dsn);
} else {
dsn.push_str(if matches!(self, Status::PermanentFailure(_)) {
"5.0.0"
} else {
"4.0.0"
});
}
}
Status::Scheduled => {
dsn.push_str("4.0.0");
}
_ => (),
}
dsn.push_str("\r\n");
}
fn write_dsn_diagnostic(&self, dsn: &mut String) {
if let Status::PermanentFailure(details) | Status::TemporaryFailure(details) = self {
details.response.write_dsn_diagnostic(dsn);
}
}
}
impl Status<(), Error> {
fn write_dsn(&self, dsn: &mut String) {
self.write_dsn_action(dsn);
self.write_dsn_status(dsn);
self.write_dsn_diagnostic(dsn);
self.write_dsn_remote_mta(dsn);
}
fn write_dsn_status(&self, dsn: &mut String) {
if let Status::PermanentFailure(err) | Status::TemporaryFailure(err) = self {
dsn.push_str("Status: ");
if let Error::UnexpectedResponse(response) = err {
response.response.write_dsn_status(dsn);
} else {
dsn.push_str(if matches!(self, Status::PermanentFailure(_)) {
"5.0.0"
} else {
"4.0.0"
});
}
dsn.push_str("\r\n");
}
}
fn write_dsn_remote_mta(&self, dsn: &mut String) {
if let Status::PermanentFailure(err) | Status::TemporaryFailure(err) = self {
match err {
Error::UnexpectedResponse(HostResponse {
hostname: details, ..
})
| Error::ConnectionError(details)
| Error::TlsError(details)
| Error::DaneError(details) => {
match self {
Status::Completed(response) => {
dsn.push_str("Remote-MTA: dns;");
dsn.push_str(&response.hostname);
dsn.push_str("\r\n");
}
Status::TemporaryFailure(err) | Status::PermanentFailure(err) => match &err.details {
Error::UnexpectedResponse(_)
| Error::ConnectionError(_)
| Error::TlsError(_)
| Error::DaneError(_) => {
dsn.push_str("Remote-MTA: dns;");
dsn.push_str(&details.entity);
dsn.push_str(&err.entity);
dsn.push_str("\r\n");
}
_ => (),
}
},
Status::Scheduled => (),
}
}
fn write_dsn_diagnostic(&self, dsn: &mut String) {
if let Status::PermanentFailure(Error::UnexpectedResponse(response))
| Status::TemporaryFailure(Error::UnexpectedResponse(response)) = self
{
response.response.write_dsn_diagnostic(dsn);
if let Status::PermanentFailure(err) | Status::TemporaryFailure(err) = self {
if let Error::UnexpectedResponse(response) = &err.details {
response.response.write_dsn_diagnostic(dsn);
}
}
}
}

314
crates/smtp/src/queue/manager.rs
View File

@@ -4,26 +4,26 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{
sync::{Arc, atomic::Ordering},
time::{Duration, Instant},
};
use ahash::{AHashMap, AHashSet};
use common::{
Inner,
core::BuildServer,
ipc::{QueueEvent, QueueEventStatus},
listener::limiter::ConcurrencyLimiter,
};
use rand::seq::SliceRandom;
use store::write::now;
use tokio::sync::mpsc;
use super::{
Message, QueueId, Status,
spool::{QUEUE_REFRESH, SmtpSpool},
};
use crate::queue::Recipient;
use ahash::AHashMap;
use common::{
Inner,
config::smtp::queue::{QueueExpiry, QueueName},
core::BuildServer,
ipc::{QueueEvent, QueueEventStatus},
};
use rand::seq::SliceRandom;
use std::{
collections::hash_map::Entry,
sync::{Arc, atomic::Ordering},
time::{Duration, Instant},
};
use store::write::now;
use tokio::sync::mpsc;
pub struct Queue {
pub core: Arc<Inner>,
@@ -35,13 +35,7 @@ pub struct Queue {
#[derive(Debug)]
pub enum OnHold {
InFlight,
ConcurrencyLimited {
limiters: Vec<ConcurrencyLimiter>,
next_due: Option<u64>,
},
Locked {
until: u64,
},
Locked { until: u64 },
}
impl SpawnQueue for mpsc::Receiver<QueueEvent> {
@@ -122,7 +116,8 @@ impl Queue {
if refresh_queue || self.next_wake_up <= Instant::now() {
// If the number of in-flight messages is greater than the maximum allowed, skip the queue
let server = self.core.build_server();
let max_in_flight = server.core.smtp.queue.max_threads;
let todo = "fix + implement virtual queues";
let max_in_flight = 4; //server.core.smtp.queue.max_threads;
has_back_pressure = in_flight_count >= max_in_flight;
if has_back_pressure {
self.next_wake_up = Instant::now() + Duration::from_secs(QUEUE_REFRESH);
@@ -138,11 +133,10 @@ impl Queue {
Details = self
.on_hold
.values()
.fold([0, 0, 0], |mut acc, v| {
.fold([0, 0], |mut acc, v| {
match v {
OnHold::InFlight => acc[0] += 1,
OnHold::ConcurrencyLimited { .. } => acc[1] += 1,
OnHold::Locked { .. } => acc[2] += 1,
OnHold::Locked { .. } => acc[1] += 1,
}
acc
})
@@ -180,13 +174,10 @@ impl Queue {
Details = self
.on_hold
.values()
.fold([0, 0, 0], |mut acc, v| {
.fold([0, 0], |mut acc, v| {
match v {
OnHold::InFlight => acc[0] += 1,
OnHold::ConcurrencyLimited { .. } => {
acc[1] += 1
}
OnHold::Locked { .. } => acc[2] += 1,
OnHold::Locked { .. } => acc[1] += 1,
}
acc
})
@@ -211,14 +202,6 @@ impl Queue {
continue;
}
}
OnHold::ConcurrencyLimited { limiters, next_due } => {
if !(limiters.iter().any(|l| {
l.concurrent.load(Ordering::Relaxed) < l.max_concurrent
}) || next_due.is_some_and(|due| due <= now))
{
continue;
}
}
OnHold::InFlight => continue,
}
@@ -243,17 +226,10 @@ impl Queue {
next_cleanup = now + CLEANUP_INTERVAL;
if !self.on_hold.is_empty() {
let active_queue_ids = queue_events
.into_iter()
.map(|e| e.queue_id)
.collect::<AHashSet<_>>();
let now = store::write::now();
self.on_hold.retain(|queue_id, status| match status {
OnHold::InFlight => true,
OnHold::Locked { until } => *until > now,
OnHold::ConcurrencyLimited { .. } => {
active_queue_ids.contains(queue_id)
}
});
}
}
@@ -269,112 +245,162 @@ impl Queue {
}
impl Message {
pub fn next_event(&self) -> Option<u64> {
let mut next_event = now();
let mut has_events = false;
for domain in &self.domains {
if matches!(
domain.status,
Status::Scheduled | Status::TemporaryFailure(_)
) {
if !has_events || domain.retry.due < next_event {
next_event = domain.retry.due;
has_events = true;
}
if domain.notify.due < next_event {
next_event = domain.notify.due;
}
if domain.expires < next_event {
next_event = domain.expires;
}
}
}
if has_events { next_event.into() } else { None }
}
pub fn next_delivery_event(&self) -> u64 {
let mut next_delivery = now();
for (pos, domain) in self
.domains
.iter()
.filter(|d| matches!(d.status, Status::Scheduled | Status::TemporaryFailure(_)))
.enumerate()
{
if pos == 0 || domain.retry.due < next_delivery {
next_delivery = domain.retry.due;
}
}
next_delivery
}
pub fn next_dsn(&self) -> u64 {
let mut next_dsn = now();
for (pos, domain) in self
.domains
.iter()
.filter(|d| matches!(d.status, Status::Scheduled | Status::TemporaryFailure(_)))
.enumerate()
{
if pos == 0 || domain.notify.due < next_dsn {
next_dsn = domain.notify.due;
}
}
next_dsn
}
pub fn expires(&self) -> u64 {
let mut expires = now();
for (pos, domain) in self
.domains
.iter()
.filter(|d| matches!(d.status, Status::Scheduled | Status::TemporaryFailure(_)))
.enumerate()
{
if pos == 0 || domain.expires < expires {
expires = domain.expires;
}
}
expires
}
pub fn next_event_after(&self, instant: u64) -> Option<u64> {
pub fn next_event(&self, queue: Option<QueueName>) -> Option<u64> {
let mut next_event = None;
for domain in &self.domains {
if matches!(
domain.status,
Status::Scheduled | Status::TemporaryFailure(_)
) {
if domain.retry.due > instant
&& next_event.as_ref().is_none_or(|ne| domain.retry.due.lt(ne))
{
next_event = domain.retry.due.into();
for rcpt in &self.recipients {
if matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_))
&& queue.is_none_or(|q| rcpt.queue == q)
{
let mut earlier_event = std::cmp::min(rcpt.retry.due, rcpt.notify.due);
if let Some(expires) = rcpt.expiration_time(self.created) {
earlier_event = std::cmp::min(earlier_event, expires);
}
if domain.notify.due > instant
&& next_event
.as_ref()
.is_none_or(|ne| domain.notify.due.lt(ne))
{
next_event = domain.notify.due.into();
}
if domain.expires > instant
&& next_event.as_ref().is_none_or(|ne| domain.expires.lt(ne))
{
next_event = domain.expires.into();
if let Some(next_event) = &mut next_event {
if earlier_event < *next_event {
*next_event = earlier_event;
}
} else {
next_event = Some(earlier_event);
}
}
}
next_event
}
pub fn next_delivery_event(&self, queue: Option<QueueName>) -> Option<u64> {
let mut next_delivery = None;
for rcpt in self.recipients.iter().filter(|rcpt| {
matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_))
&& queue.is_none_or(|q| rcpt.queue == q)
}) {
if let Some(next_delivery) = &mut next_delivery {
if rcpt.retry.due < *next_delivery {
*next_delivery = rcpt.retry.due;
}
} else {
next_delivery = Some(rcpt.retry.due);
}
}
next_delivery
}
pub fn next_dsn(&self, queue: Option<QueueName>) -> Option<u64> {
let mut next_dsn = None;
for rcpt in self.recipients.iter().filter(|rcpt| {
matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_))
&& queue.is_none_or(|q| rcpt.queue == q)
}) {
if let Some(next_dsn) = &mut next_dsn {
if rcpt.notify.due < *next_dsn {
*next_dsn = rcpt.notify.due;
}
} else {
next_dsn = Some(rcpt.notify.due);
}
}
next_dsn
}
pub fn expires(&self, queue: Option<QueueName>) -> Option<u64> {
let mut expires = None;
for rcpt in self.recipients.iter().filter(|d| {
matches!(d.status, Status::Scheduled | Status::TemporaryFailure(_))
&& queue.is_none_or(|q| d.queue == q)
}) {
if let Some(rcpt_expires) = rcpt.expiration_time(self.created) {
if let Some(expires) = &mut expires {
if rcpt_expires > *expires {
*expires = rcpt_expires;
}
} else {
expires = Some(rcpt_expires)
}
}
}
expires
}
pub fn next_event_after(&self, queue: Option<QueueName>, instant: u64) -> Option<u64> {
let mut next_event = None;
for rcpt in &self.recipients {
if matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_))
&& queue.is_none_or(|q| rcpt.queue == q)
{
if rcpt.retry.due > instant
&& next_event.as_ref().is_none_or(|ne| rcpt.retry.due.lt(ne))
{
next_event = rcpt.retry.due.into();
}
if rcpt.notify.due > instant
&& next_event.as_ref().is_none_or(|ne| rcpt.notify.due.lt(ne))
{
next_event = rcpt.notify.due.into();
}
if let Some(expires) = rcpt.expiration_time(self.created) {
if expires > instant && next_event.as_ref().is_none_or(|ne| expires.lt(ne)) {
next_event = expires.into();
}
}
}
}
next_event
}
pub fn next_events(&self) -> AHashMap<QueueName, u64> {
let mut next_events = AHashMap::new();
for rcpt in &self.recipients {
if matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_)) {
let mut earlier_event = std::cmp::min(rcpt.retry.due, rcpt.notify.due);
if let Some(expires) = rcpt.expiration_time(self.created) {
earlier_event = std::cmp::min(earlier_event, expires);
}
match next_events.entry(rcpt.queue) {
Entry::Occupied(mut entry) => {
let entry = entry.get_mut();
if earlier_event < *entry {
*entry = earlier_event;
}
}
Entry::Vacant(entry) => {
entry.insert(earlier_event);
}
}
}
}
next_events
}
}
impl Recipient {
pub fn expiration_time(&self, created: u64) -> Option<u64> {
match self.expires {
QueueExpiry::Duration(time) => Some(created + time),
QueueExpiry::Count(_) => None,
}
}
pub fn is_expired(&self, created: u64, now: u64) -> bool {
match self.expires {
QueueExpiry::Duration(time) => created + time <= now,
QueueExpiry::Count(count) => self.retry.inner >= count,
}
}
}
pub trait SpawnQueue {

225
crates/smtp/src/queue/mod.rs
View File

@@ -4,16 +4,17 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::{
config::smtp::queue::{QueueExpiry, QueueName},
expr::{self, functions::ResolveVariable, *},
};
use compact_str::ToCompactString;
use smtp_proto::{ArchivedResponse, Response};
use std::{
fmt::Display,
net::{IpAddr, Ipv4Addr},
time::{Duration, Instant, SystemTime},
};
use common::expr::{self, functions::ResolveVariable, *};
use compact_str::ToCompactString;
use smtp_proto::{ArchivedResponse, Response};
use store::write::now;
use utils::BlobHash;
@@ -34,7 +35,8 @@ pub struct Schedule<T> {
#[derive(Debug, Clone, Copy)]
pub struct QueuedMessage {
pub due: u64,
pub queue_id: u64,
pub queue_id: QueueId,
pub queue_name: QueueName,
}
#[derive(Debug, Clone, Copy)]
@@ -48,15 +50,16 @@ pub enum MessageSource {
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Clone, PartialEq, Eq)]
pub struct Message {
pub queue_id: QueueId,
pub created: u64,
pub blob_hash: BlobHash,
pub received_from_ip: IpAddr,
pub received_via_port: u16,
pub return_path: String,
pub return_path_lcase: String,
pub return_path_domain: String,
pub recipients: Vec<Recipient>,
pub domains: Vec<Domain>,
pub flags: u64,
pub env_id: Option<String>,
@@ -64,9 +67,14 @@ pub struct Message {
pub size: u64,
pub quota_keys: Vec<QuotaKey>,
}
#[rkyv(with = rkyv::with::Skip)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MessageWrapper {
pub queue_id: QueueId,
pub queue_name: QueueName,
pub span_id: u64,
pub message: Message,
}
#[derive(
@@ -84,24 +92,6 @@ pub enum QuotaKey {
Count { key: Vec<u8>, id: u64 },
}
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
Clone,
PartialEq,
Eq,
serde::Deserialize,
)]
pub struct Domain {
pub domain: String,
pub retry: Schedule<u32>,
pub notify: Schedule<u32>,
pub expires: u64,
pub status: Status<(), Error>,
}
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
@@ -113,10 +103,15 @@ pub struct Domain {
serde::Deserialize,
)]
pub struct Recipient {
pub domain_idx: u32,
pub address: String,
pub address_lcase: String,
pub status: Status<HostResponse<String>, HostResponse<ErrorDetails>>,
pub retry: Schedule<u32>,
pub notify: Schedule<u32>,
pub expires: QueueExpiry,
pub queue: QueueName,
pub status: Status<HostResponse<String>, ErrorDetails>,
pub flags: u64,
pub orcpt: Option<String>,
}
@@ -173,19 +168,36 @@ pub struct HostResponse<T> {
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
Default,
)]
pub enum Error {
DnsError(String),
UnexpectedResponse(HostResponse<ErrorDetails>),
ConnectionError(ErrorDetails),
TlsError(ErrorDetails),
DaneError(ErrorDetails),
UnexpectedResponse(UnexpectedResponse),
ConnectionError(String),
TlsError(String),
DaneError(String),
MtaStsError(String),
RateLimited,
#[default]
ConcurrencyLimited,
Io(String),
}
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct UnexpectedResponse {
pub command: String,
pub response: Response<String>,
}
#[derive(
Debug,
Clone,
@@ -199,7 +211,7 @@ pub enum Error {
)]
pub struct ErrorDetails {
pub entity: String,
pub details: String,
pub details: Error,
}
impl<T> Ord for Schedule<T> {
@@ -230,9 +242,9 @@ impl<T: Default> Schedule<T> {
}
}
pub fn later(duration: Duration) -> Self {
pub fn later(duration: u64) -> Self {
Schedule {
due: now() + duration.as_secs(),
due: now() + duration,
inner: T::default(),
}
}
@@ -243,37 +255,22 @@ pub struct QueueEnvelope<'x> {
pub mx: &'x str,
pub remote_ip: IpAddr,
pub local_ip: IpAddr,
pub current_domain: usize,
pub current_rcpt: usize,
}
impl<'x> QueueEnvelope<'x> {
pub fn new(message: &'x Message, current_domain: usize) -> Self {
pub fn new_rcpt(message: &'x Message, current_rcpt: usize) -> Self {
Self {
message,
current_domain,
current_rcpt: 0,
mx: "",
remote_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
local_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
}
}
pub fn new_rcpt(message: &'x Message, current_domain: usize, current_rcpt: usize) -> Self {
Self {
message,
current_domain,
current_rcpt,
mx: "",
remote_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
local_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
}
}
}
impl<'x> QueueEnvelope<'x> {
fn current_domain(&self) -> Option<&'x Domain> {
self.message.domains.get(self.current_domain)
fn current_recipient(&self) -> Option<&'x Recipient> {
self.message.recipients.get(self.current_rcpt)
}
}
@@ -283,14 +280,12 @@ impl<'x> ResolveVariable for QueueEnvelope<'x> {
V_SENDER => self.message.return_path_lcase.as_str().into(),
V_SENDER_DOMAIN => self.message.return_path_domain.as_str().into(),
V_RECIPIENT_DOMAIN => self
.current_domain()
.map(|d| d.domain.as_str())
.current_recipient()
.map(|d| d.address_lcase.domain_part())
.unwrap_or_default()
.into(),
V_RECIPIENT => self
.message
.recipients
.get(self.current_rcpt)
.current_recipient()
.map(|r| r.address_lcase.as_str())
.unwrap_or_default()
.into(),
@@ -302,40 +297,47 @@ impl<'x> ResolveVariable for QueueEnvelope<'x> {
.collect::<Vec<_>>()
.into(),
V_QUEUE_RETRY_NUM => self
.current_domain()
.current_recipient()
.map(|d| d.retry.inner)
.unwrap_or_default()
.into(),
V_QUEUE_NOTIFY_NUM => self
.current_domain()
.current_recipient()
.map(|d| d.notify.inner)
.unwrap_or_default()
.into(),
V_QUEUE_EXPIRES_IN => self
.current_domain()
.map(|d| d.expires.saturating_sub(now()))
.current_recipient()
.map(|d| match &d.expires {
QueueExpiry::Duration(time) => {
(*time + self.message.created).saturating_sub(now())
}
QueueExpiry::Count(count) => (*count) as u64,
})
.unwrap_or_default()
.into(),
V_QUEUE_LAST_STATUS => self
.current_domain()
.current_recipient()
.map(|d| d.status.to_compact_string())
.unwrap_or_default()
.into(),
V_QUEUE_LAST_ERROR => self
.current_domain()
.current_recipient()
.map(|d| match &d.status {
Status::Scheduled | Status::Completed(_) => "none",
Status::TemporaryFailure(err) | Status::PermanentFailure(err) => match err {
Error::DnsError(_) => "dns",
Error::UnexpectedResponse(_) => "unexpected-reply",
Error::ConnectionError(_) => "connection",
Error::TlsError(_) => "tls",
Error::DaneError(_) => "dane",
Error::MtaStsError(_) => "mta-sts",
Error::RateLimited => "rate",
Error::ConcurrencyLimited => "concurrency",
Error::Io(_) => "io",
},
Status::TemporaryFailure(err) | Status::PermanentFailure(err) => {
match &err.details {
Error::DnsError(_) => "dns",
Error::UnexpectedResponse(_) => "unexpected-reply",
Error::ConnectionError(_) => "connection",
Error::TlsError(_) => "tls",
Error::DaneError(_) => "dane",
Error::MtaStsError(_) => "mta-sts",
Error::RateLimited => "rate",
Error::ConcurrencyLimited => "concurrency",
Error::Io(_) => "io",
}
}
})
.unwrap_or_default()
.into(),
@@ -440,33 +442,21 @@ impl Display for Error {
Error::UnexpectedResponse(response) => {
write!(
f,
"Unexpected response from '{}': {}",
response.hostname.entity, response.response
"Unexpected response for {}: {}",
response.command, response.response
)
}
Error::DnsError(err) => {
write!(f, "DNS lookup failed: {err}")
}
Error::ConnectionError(details) => {
write!(
f,
"Connection to '{}' failed: {}",
details.entity, details.details
)
write!(f, "Connection failed: {details}",)
}
Error::TlsError(details) => {
write!(
f,
"TLS error from '{}': {}",
details.entity, details.details
)
write!(f, "TLS error: {details}",)
}
Error::DaneError(details) => {
write!(
f,
"DANE failed to authenticate '{}': {}",
details.entity, details.details
)
write!(f, "DANE authentication failure: {details}",)
}
Error::MtaStsError(details) => {
write!(f, "MTA-STS auth failed: {details}")
@@ -490,8 +480,8 @@ impl Display for ArchivedError {
ArchivedError::UnexpectedResponse(response) => {
write!(
f,
"Unexpected response from '{}': {}",
response.hostname.entity,
"Unexpected response for {}: {}",
response.command,
response.response.to_string()
)
}
@@ -499,25 +489,13 @@ impl Display for ArchivedError {
write!(f, "DNS lookup failed: {err}")
}
ArchivedError::ConnectionError(details) => {
write!(
f,
"Connection to '{}' failed: {}",
details.entity, details.details
)
write!(f, "Connection failed: {details}",)
}
ArchivedError::TlsError(details) => {
write!(
f,
"TLS error from '{}': {}",
details.entity, details.details
)
write!(f, "TLS error: {details}",)
}
ArchivedError::DaneError(details) => {
write!(
f,
"DANE failed to authenticate '{}': {}",
details.entity, details.details
)
write!(f, "DANE authentication failure: {details}",)
}
ArchivedError::MtaStsError(details) => {
write!(f, "MTA-STS auth failed: {details}")
@@ -535,28 +513,27 @@ impl Display for ArchivedError {
}
}
impl Display for Status<(), Error> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Status::Scheduled => write!(f, "Scheduled"),
Status::Completed(_) => write!(f, "Completed"),
Status::TemporaryFailure(err) => write!(f, "Temporary Failure: {err}"),
Status::PermanentFailure(err) => write!(f, "Permanent Failure: {err}"),
}
}
}
impl Display for Status<HostResponse<String>, HostResponse<ErrorDetails>> {
impl Display for Status<HostResponse<String>, ErrorDetails> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Status::Scheduled => write!(f, "Scheduled"),
Status::Completed(response) => write!(f, "Delivered: {}", response.response),
Status::TemporaryFailure(err) => write!(f, "Temporary Failure: {}", err.response),
Status::PermanentFailure(err) => write!(f, "Permanent Failure: {}", err.response),
Status::TemporaryFailure(err) => {
write!(f, "Temporary Failure for {}: {}", err.entity, err.details)
}
Status::PermanentFailure(err) => {
write!(f, "Permanent Failure for {}: {}", err.entity, err.details)
}
}
}
}
impl Display for ArchivedErrorDetails {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Error for {}: {}", self.entity, self.details)
}
}
pub trait DisplayArchivedResponse {
fn to_string(&self) -> String;
}

100
crates/smtp/src/queue/quota.rs
View File

@@ -4,21 +4,22 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::future::Future;
use super::{QueueEnvelope, QuotaKey, Status};
use crate::{
core::throttle::NewKey,
queue::{DomainPart, MessageWrapper},
};
use ahash::AHashSet;
use common::{Server, config::smtp::queue::QueueQuota, expr::functions::ResolveVariable};
use std::future::Future;
use store::{
ValueKey,
write::{BatchBuilder, QueueClass, ValueClass},
};
use trc::QueueEvent;
use crate::core::throttle::NewKey;
use super::{Message, QueueEnvelope, QuotaKey, Status};
pub trait HasQueueQuota: Sync + Send {
fn has_quota(&self, message: &mut Message) -> impl Future<Output = bool> + Send;
fn has_quota(&self, message: &mut MessageWrapper) -> impl Future<Output = bool> + Send;
fn check_quota<'x>(
&'x self,
quota: &'x QueueQuota,
@@ -31,7 +32,7 @@ pub trait HasQueueQuota: Sync + Send {
}
impl HasQueueQuota for Server {
async fn has_quota(&self, message: &mut Message) -> bool {
async fn has_quota(&self, message: &mut MessageWrapper) -> bool {
let mut quota_keys = Vec::new();
if !self.core.smtp.queue.quota.sender.is_empty() {
@@ -39,8 +40,8 @@ impl HasQueueQuota for Server {
if !self
.check_quota(
quota,
message,
message.size,
&message.message,
message.message.size,
0,
&mut quota_keys,
message.span_id,
@@ -59,38 +60,42 @@ impl HasQueueQuota for Server {
}
}
for quota in &self.core.smtp.queue.quota.rcpt_domain {
for domain_idx in 0..message.domains.len() {
if !self
.check_quota(
quota,
&QueueEnvelope::new(message, domain_idx),
message.size,
((domain_idx + 1) << 32) as u64,
&mut quota_keys,
message.span_id,
)
.await
{
trc::event!(
Queue(QueueEvent::QuotaExceeded),
SpanId = message.span_id,
Id = quota.id.clone(),
Type = "Domain"
);
if !self.core.smtp.queue.quota.rcpt_domain.is_empty() {
let mut seen_domains = AHashSet::new();
for quota in &self.core.smtp.queue.quota.rcpt_domain {
for (rcpt_idx, rcpt) in message.message.recipients.iter().enumerate() {
if seen_domains.insert(rcpt.address_lcase.domain_part())
&& !self
.check_quota(
quota,
&QueueEnvelope::new_rcpt(&message.message, rcpt_idx),
message.message.size,
((rcpt_idx + 1) << 32) as u64,
&mut quota_keys,
message.span_id,
)
.await
{
trc::event!(
Queue(QueueEvent::QuotaExceeded),
SpanId = message.span_id,
Id = quota.id.clone(),
Type = "Domain"
);
return false;
return false;
}
}
}
}
for quota in &self.core.smtp.queue.quota.rcpt {
for (rcpt_idx, rcpt) in message.recipients.iter().enumerate() {
for rcpt_idx in 0..message.message.recipients.len() {
if !self
.check_quota(
quota,
&QueueEnvelope::new_rcpt(message, rcpt.domain_idx as usize, rcpt_idx),
message.size,
&QueueEnvelope::new_rcpt(&message.message, rcpt_idx),
message.message.size,
(rcpt_idx + 1) as u64,
&mut quota_keys,
message.span_id,
@@ -109,7 +114,7 @@ impl HasQueueQuota for Server {
}
}
message.quota_keys = quota_keys;
message.message.quota_keys = quota_keys;
true
}
@@ -174,32 +179,29 @@ impl HasQueueQuota for Server {
}
}
impl Message {
impl MessageWrapper {
pub fn release_quota(&mut self, batch: &mut BatchBuilder) {
if self.quota_keys.is_empty() {
if self.message.quota_keys.is_empty() {
return;
}
let mut quota_ids = Vec::with_capacity(self.domains.len() + self.recipients.len());
for (pos, domain) in self.domains.iter().enumerate() {
if matches!(
&domain.status,
Status::Completed(_) | Status::PermanentFailure(_)
) {
quota_ids.push(((pos + 1) as u64) << 32);
}
}
for (pos, rcpt) in self.recipients.iter().enumerate() {
let mut quota_ids = Vec::with_capacity(self.message.recipients.len());
let mut seen_domains = AHashSet::new();
for (pos, rcpt) in self.message.recipients.iter().enumerate() {
if matches!(
&rcpt.status,
Status::Completed(_) | Status::PermanentFailure(_)
) {
if seen_domains.insert(rcpt.address_lcase.domain_part()) {
quota_ids.push(((pos + 1) as u64) << 32);
}
quota_ids.push((pos + 1) as u64);
}
}
if !quota_ids.is_empty() {
let mut quota_keys = Vec::new();
for quota_key in std::mem::take(&mut self.quota_keys) {
for quota_key in std::mem::take(&mut self.message.quota_keys) {
match quota_key {
QuotaKey::Count { id, key } if quota_ids.contains(&id) => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key)), -1);
@@ -207,7 +209,7 @@ impl Message {
QuotaKey::Size { id, key } if quota_ids.contains(&id) => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key)),
-(self.size as i64),
-(self.message.size as i64),
);
}
_ => {
@@ -215,7 +217,7 @@ impl Message {
}
}
}
self.quota_keys = quota_keys;
self.message.quota_keys = quota_keys;
}
}
}

323
crates/smtp/src/queue/spool.rs
View File

@@ -4,13 +4,18 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::queue::DomainPart;
use super::{
ArchivedMessage, ArchivedStatus, Message, MessageSource, QueueEnvelope, QueueId, QueuedMessage,
QuotaKey, Recipient, Schedule, Status,
};
use crate::queue::{DomainPart, MessageWrapper};
use common::config::smtp::queue::{QueueExpiry, QueueName};
use common::ipc::QueueEvent;
use common::{KV_LOCK_QUEUE_MESSAGE, Server};
use std::borrow::Cow;
use std::future::Future;
use std::time::{Duration, SystemTime};
use std::net::{IpAddr, Ipv4Addr};
use std::time::SystemTime;
use store::write::key::DeserializeBigEndian;
use store::write::{
AlignedBytes, Archive, Archiver, BatchBuilder, BlobOp, QueueClass, ValueClass, now,
@@ -19,11 +24,6 @@ use store::{IterateParams, Serialize, SerializeInfallible, U64_LEN, ValueKey};
use trc::ServerEvent;
use utils::BlobHash;
use super::{
ArchivedMessage, ArchivedStatus, Domain, Message, MessageSource, QueueEnvelope, QueueId,
QueuedMessage, QuotaKey, Recipient, Schedule, Status,
};
pub const LOCK_EXPIRY: u64 = 300;
pub const QUEUE_REFRESH: u64 = 300;
@@ -34,7 +34,7 @@ pub trait SmtpSpool: Sync + Send {
return_path_lcase: impl Into<String>,
return_path_domain: impl Into<String>,
span_id: u64,
) -> Message;
) -> MessageWrapper;
fn next_event(&self) -> impl Future<Output = Vec<QueuedMessage>> + Send;
@@ -42,7 +42,11 @@ pub trait SmtpSpool: Sync + Send {
fn unlock_event(&self, queue_id: QueueId) -> impl Future<Output = ()> + Send;
fn read_message(&self, id: QueueId) -> impl Future<Output = Option<Message>> + Send;
fn read_message(
&self,
id: QueueId,
queue_name: QueueName,
) -> impl Future<Output = Option<MessageWrapper>> + Send;
fn read_message_archive(
&self,
@@ -57,25 +61,30 @@ impl SmtpSpool for Server {
return_path_lcase: impl Into<String>,
return_path_domain: impl Into<String>,
span_id: u64,
) -> Message {
) -> MessageWrapper {
let created = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map_or(0, |d| d.as_secs());
Message {
MessageWrapper {
queue_id: self.inner.data.queue_id_gen.generate(),
queue_name: QueueName::default(),
span_id,
created,
return_path: return_path.into(),
return_path_lcase: return_path_lcase.into(),
return_path_domain: return_path_domain.into(),
recipients: Vec::with_capacity(1),
domains: Vec::with_capacity(1),
flags: 0,
env_id: None,
priority: 0,
size: 0,
blob_hash: Default::default(),
quota_keys: Vec::new(),
message: Message {
created,
return_path: return_path.into(),
return_path_lcase: return_path_lcase.into(),
return_path_domain: return_path_domain.into(),
recipients: Vec::with_capacity(1),
flags: 0,
env_id: None,
priority: 0,
size: 0,
blob_hash: Default::default(),
quota_keys: Vec::new(),
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,
},
}
}
@@ -85,12 +94,14 @@ impl SmtpSpool for Server {
store::write::QueueEvent {
due: 0,
queue_id: 0,
queue_name: [0; 8],
},
)));
let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: now + QUEUE_REFRESH,
queue_id: u64::MAX,
queue_name: [u8::MAX; 8],
},
)));
@@ -103,8 +114,15 @@ impl SmtpSpool for Server {
|key, _| {
let due = key.deserialize_be_u64(0)?;
let queue_id = key.deserialize_be_u64(U64_LEN)?;
let queue_name =
QueueName::from_bytes(key.get(U64_LEN + U64_LEN..).unwrap_or_default())
.unwrap_or_default();
events.push(QueuedMessage { due, queue_id });
events.push(QueuedMessage {
due,
queue_id,
queue_name,
});
Ok(due <= now)
},
@@ -156,12 +174,24 @@ impl SmtpSpool for Server {
}
}
async fn read_message(&self, id: QueueId) -> Option<Message> {
match self.read_message_archive(id).await.and_then(|a| match a {
Some(a) => a.deserialize::<Message>().map(Some),
None => Ok(None),
}) {
Ok(Some(message)) => Some(message),
async fn read_message(
&self,
queue_id: QueueId,
queue_name: QueueName,
) -> Option<MessageWrapper> {
match self
.read_message_archive(queue_id)
.await
.and_then(|a| match a {
Some(a) => a.deserialize::<Message>().map(Some),
None => Ok(None),
}) {
Ok(Some(message)) => Some(MessageWrapper {
queue_id,
queue_name,
span_id: 0,
message,
}),
Ok(None) => None,
Err(err) => {
trc::error!(
@@ -186,7 +216,7 @@ impl SmtpSpool for Server {
}
}
impl Message {
impl MessageWrapper {
pub async fn queue(
mut self,
raw_headers: Option<&[u8]>,
@@ -204,11 +234,11 @@ impl Message {
} else {
raw_message.into()
};
self.blob_hash = BlobHash::generate(message.as_ref());
self.message.blob_hash = BlobHash::generate(message.as_ref());
// Generate id
if self.size == 0 {
self.size = message.len() as u64;
if self.message.size == 0 {
self.message.size = message.len() as u64;
}
// Reserve and write blob
@@ -216,7 +246,7 @@ impl Message {
let reserve_until = now() + 120;
batch.set(
BlobOp::Reserve {
hash: self.blob_hash.clone(),
hash: self.message.blob_hash.clone(),
until: reserve_until,
},
0u32.serialize(),
@@ -232,7 +262,7 @@ impl Message {
}
if let Err(err) = server
.blob_store()
.put_blob(self.blob_hash.as_slice(), message.as_ref())
.put_blob(self.message.blob_hash.as_slice(), message.as_ref())
.await
{
trc::error!(
@@ -254,27 +284,31 @@ impl Message {
}),
SpanId = session_id,
QueueId = self.queue_id,
From = if !self.return_path.is_empty() {
trc::Value::String(self.return_path.as_str().into())
From = if !self.message.return_path.is_empty() {
trc::Value::String(self.message.return_path.as_str().into())
} else {
trc::Value::String("<>".into())
},
To = self
.message
.recipients
.iter()
.map(|r| trc::Value::String(r.address_lcase.as_str().into()))
.collect::<Vec<_>>(),
Size = self.size,
NextRetry = trc::Value::Timestamp(self.next_delivery_event()),
NextDsn = trc::Value::Timestamp(self.next_dsn()),
Expires = trc::Value::Timestamp(self.expires()),
Size = self.message.size,
NextRetry = self
.message
.next_delivery_event(None)
.map(trc::Value::Timestamp),
NextDsn = self.message.next_dsn(None).map(trc::Value::Timestamp),
Expires = self.message.expires(None).map(trc::Value::Timestamp),
);
// Write message to queue
let mut batch = BatchBuilder::new();
// Reserve quotas
for quota_key in &self.quota_keys {
for quota_key in &self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.clone())), 1);
@@ -282,39 +316,44 @@ impl Message {
QuotaKey::Size { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key.clone())),
self.size as i64,
self.message.size as i64,
);
}
}
}
batch
.set(
for (queue_name, due) in self.message.next_events() {
batch.set(
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due: self.next_event().unwrap_or_default(),
due,
queue_id: self.queue_id,
queue_name: queue_name.into_inner(),
})),
0u64.serialize(),
)
Vec::new(),
);
}
batch
.clear(BlobOp::Reserve {
hash: self.blob_hash.clone(),
hash: self.message.blob_hash.clone(),
until: reserve_until,
})
.set(
BlobOp::LinkId {
hash: self.blob_hash.clone(),
hash: self.message.blob_hash.clone(),
id: self.queue_id,
},
vec![],
)
.set(
BlobOp::Commit {
hash: self.blob_hash.clone(),
hash: self.message.blob_hash.clone(),
},
vec![],
)
.set(
ValueClass::Queue(QueueClass::Message(self.queue_id)),
match Archiver::new(self).serialize() {
match Archiver::new(self.message).serialize() {
Ok(data) => data,
Err(err) => {
trc::error!(
@@ -361,92 +400,77 @@ impl Message {
&mut self,
rcpt: impl Into<String>,
rcpt_lcase: impl Into<String>,
rcpt_domain: impl Into<String>,
server: &Server,
) {
let rcpt_domain = rcpt_domain.into();
let domain_idx =
if let Some(idx) = self.domains.iter().position(|d| d.domain == rcpt_domain) {
idx
} else {
let idx = self.domains.len();
self.domains.push(Domain {
domain: rcpt_domain,
retry: Schedule::now(),
notify: Schedule::now(),
expires: 0,
status: Status::Scheduled,
});
let expires = server
.eval_if(
&server.core.smtp.queue.expire,
&QueueEnvelope::new(self, idx),
self.span_id,
)
.await
.unwrap_or_else(|| Duration::from_secs(5 * 86400));
// Update expiration
let domain = self.domains.last_mut().unwrap();
domain.notify = Schedule::later(expires + Duration::from_secs(10));
domain.expires = now() + expires.as_secs();
idx
};
self.recipients.push(Recipient {
domain_idx: domain_idx as u32,
// Resolve queue
let idx = self.message.recipients.len();
self.message.recipients.push(Recipient {
address: rcpt.into(),
address_lcase: rcpt_lcase.into(),
status: Status::Scheduled,
flags: 0,
orcpt: None,
retry: Schedule::now(),
notify: Schedule::now(),
expires: QueueExpiry::Count(0),
queue: QueueName::default(),
});
let queue = server.get_queue_or_default(
&server
.eval_if::<String, _>(
&server.core.smtp.queue.queue,
&QueueEnvelope::new_rcpt(&self.message, idx),
self.span_id,
)
.await
.unwrap_or_else(|| "default".to_string()),
self.span_id,
);
// Update expiration
let now = now();
let recipient = self.message.recipients.last_mut().unwrap();
recipient.notify = Schedule::later(queue.notify.first().copied().unwrap_or(86400) + now);
recipient.expires = queue.expiry;
recipient.queue = queue.virtual_queue;
}
pub async fn add_recipient(&mut self, rcpt: impl Into<String>, server: &Server) {
let rcpt = rcpt.into();
let rcpt_lcase = rcpt.to_lowercase();
let rcpt_domain = rcpt_lcase.domain_part().to_string();
self.add_recipient_parts(rcpt, rcpt_lcase, rcpt_domain, server)
.await;
self.add_recipient_parts(rcpt, rcpt_lcase, server).await;
}
pub async fn save_changes(
mut self,
server: &Server,
prev_event: Option<u64>,
next_event: Option<u64>,
) -> bool {
debug_assert!(prev_event.is_some() == next_event.is_some());
pub async fn save_changes(mut self, server: &Server, prev_event: Option<u64>) -> bool {
// Release quota for completed deliveries
let mut batch = BatchBuilder::new();
self.release_quota(&mut batch);
// Update message queue
if let (Some(prev_event), Some(next_event)) = (prev_event, next_event) {
batch
.clear(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: prev_event,
queue_id: self.queue_id,
},
)))
.set(
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due: next_event,
queue_id: self.queue_id,
})),
0u64.serialize(),
);
if let Some(prev_event) = prev_event {
batch.clear(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: prev_event,
queue_id: self.queue_id,
queue_name: self.queue_name.into_inner(),
},
)));
}
for (queue_name, due) in self.message.next_events() {
batch.set(
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due,
queue_id: self.queue_id,
queue_name: queue_name.into_inner(),
})),
Vec::new(),
);
}
let span_id = self.span_id;
batch.set(
ValueClass::Queue(QueueClass::Message(self.queue_id)),
match Archiver::new(self).serialize() {
match Archiver::new(self.message).serialize() {
Ok(data) => data,
Err(err) => {
trc::error!(
@@ -471,11 +495,31 @@ impl Message {
}
}
pub async fn remove(self, server: &Server, prev_event: u64) -> bool {
pub async fn remove(self, server: &Server, prev_event: Option<u64>) -> bool {
let mut batch = BatchBuilder::new();
if let Some(prev_event) = prev_event {
batch.clear(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: prev_event,
queue_id: self.queue_id,
queue_name: self.queue_name.into_inner(),
},
)));
} else {
for (queue_name, due) in self.message.next_events() {
batch.clear(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due,
queue_id: self.queue_id,
queue_name: queue_name.into_inner(),
},
)));
}
}
// Release all quotas
for quota_key in self.quota_keys {
for quota_key in self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key)), -1);
@@ -483,7 +527,7 @@ impl Message {
QuotaKey::Size { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key)),
-(self.size as i64),
-(self.message.size as i64),
);
}
}
@@ -491,15 +535,9 @@ impl Message {
batch
.clear(BlobOp::LinkId {
hash: self.blob_hash.clone(),
hash: self.message.blob_hash.clone(),
id: self.queue_id,
})
.clear(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: prev_event,
queue_id: self.queue_id,
},
)))
.clear(ValueClass::Queue(QueueClass::Message(self.queue_id)));
if let Err(err) = server.store().write(batch.build_all()).await {
@@ -515,30 +553,33 @@ impl Message {
}
pub fn has_domain(&self, domains: &[String]) -> bool {
self.domains.iter().any(|d| domains.contains(&d.domain))
|| self
.return_path
.rsplit_once('@')
.is_some_and(|(_, domain)| domains.iter().any(|dd| dd == domain))
self.message.recipients.iter().any(|r| {
let domain = r.address_lcase.domain_part();
domains.iter().any(|dd| dd == domain)
}) || self
.message
.return_path
.rsplit_once('@')
.is_some_and(|(_, domain)| domains.iter().any(|dd| dd == domain))
}
}
impl ArchivedMessage {
pub fn has_domain(&self, domains: &[String]) -> bool {
self.domains
.iter()
.any(|d| domains.iter().any(|dd| dd == d.domain.as_str()))
|| self
.return_path
.rsplit_once('@')
.is_some_and(|(_, domain)| domains.iter().any(|dd| dd == domain))
self.recipients.iter().any(|r| {
let domain = r.address_lcase.domain_part();
domains.iter().any(|dd| dd == domain)
}) || self
.return_path
.rsplit_once('@')
.is_some_and(|(_, domain)| domains.iter().any(|dd| dd == domain))
}
pub fn next_delivery_event(&self) -> u64 {
let mut next_delivery = now();
for (pos, domain) in self
.domains
for (pos, rcpt) in self
.recipients
.iter()
.filter(|d| {
matches!(
@@ -548,8 +589,8 @@ impl ArchivedMessage {
})
.enumerate()
{
if pos == 0 || domain.retry.due < next_delivery {
next_delivery = domain.retry.due.into();
if pos == 0 || rcpt.retry.due < next_delivery {
next_delivery = rcpt.retry.due.into();
}
}

15
crates/smtp/src/queue/throttle.rs
View File

@@ -4,17 +4,13 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::future::Future;
use crate::core::throttle::NewKey;
use common::{
KV_RATE_LIMIT_SMTP, Server, config::smtp::QueueRateLimiter, expr::functions::ResolveVariable,
};
use std::future::Future;
use store::write::now;
use crate::core::throttle::NewKey;
use super::{Domain, Status};
pub trait IsAllowed: Sync + Send {
fn is_allowed<'x>(
&'x self,
@@ -69,10 +65,3 @@ impl IsAllowed for Server {
Ok(())
}
}
impl Domain {
pub fn set_rate_limiter_error(&mut self, retry_at: u64) {
self.retry.due = retry_at;
self.status = Status::TemporaryFailure(super::Error::RateLimited);
}
}