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

Core refactoring

Browse Source
This commit is contained in:
mdecimus
2024-09-26 14:49:46 +02:00
parent 24967c1e86
commit ce8182ae07
267 changed files with 5886 additions and 4461 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,6 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::Server;
use mail_builder::headers::content_type::ContentType;
use mail_builder::headers::HeaderType;
use mail_builder::mime::{make_boundary, BodyPart, MimePart};
@@ -13,19 +14,26 @@ use smtp_proto::{
Response, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS,
};
use std::fmt::Write;
use std::future::Future;
use std::time::Duration;
use store::write::now;
use crate::core::SMTP;
use crate::outbound::client::from_error_status;
use crate::reporting::SmtpReporting;
use super::spool::SmtpSpool;
use super::{
Domain, Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope, Recipient,
Status, RCPT_DSN_SENT, RCPT_STATUS_CHANGED,
};
impl SMTP {
pub async fn send_dsn(&self, message: &mut Message) {
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;
}
impl SendDsn for Server {
async fn send_dsn(&self, message: &mut Message) {
// Send DSN events
self.log_dsn(message).await;
@@ -152,8 +160,8 @@ impl SMTP {
}
impl Message {
pub async fn build_dsn(&mut self, core: &SMTP) -> Option<Vec<u8>> {
let config = &core.core.smtp.queue;
pub async fn build_dsn(&mut self, server: &Server) -> Option<Vec<u8>> {
let config = &server.core.smtp.queue;
let now = now();
let mut txt_success = String::new();
@@ -314,8 +322,7 @@ impl Message {
{
let envelope = QueueEnvelope::new(self, domain_idx);
if let Some(next_notify) = core
.core
if let Some(next_notify) = server
.eval_if::<Vec<Duration>, _>(&config.notify, &envelope, self.span_id)
.await
.and_then(|notify| {
@@ -337,19 +344,16 @@ impl Message {
}
// Obtain hostname and sender addresses
let from_name = core
.core
let from_name = server
.eval_if(&config.dsn.name, self, self.span_id)
.await
.unwrap_or_else(|| String::from("Mail Delivery Subsystem"));
let from_addr = core
.core
let from_addr = server
.eval_if(&config.dsn.address, self, self.span_id)
.await
.unwrap_or_else(|| String::from("MAILER-DAEMON@localhost"));
let reporting_mta = core
.core
.eval_if(&core.core.smtp.report.submitter, self, self.span_id)
let reporting_mta = server
.eval_if(&server.core.smtp.report.submitter, self, self.span_id)
.await
.unwrap_or_else(|| String::from("localhost"));
@@ -359,10 +363,8 @@ impl Message {
let dsn = dsn_header + dsn.as_str();
// Fetch up to 1024 bytes of message headers
let headers = match core
.core
.storage
.blob
let headers = match server
.blob_store()
.get_blob(self.blob_hash.as_slice(), 0..1024)
.await
{

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

@@ -4,33 +4,39 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{sync::atomic::Ordering, time::Duration};
use std::{
sync::{atomic::Ordering, Arc},
time::Duration,
};
use common::{
core::BuildServer,
ipc::{OnHold, QueueEvent, QueueEventLock},
Inner,
};
use store::write::now;
use tokio::sync::mpsc;
use crate::core::{SmtpInstance, SMTP};
use super::{spool::QueueEventLock, DeliveryAttempt, Event, Message, OnHold, Status};
use super::{spool::SmtpSpool, DeliveryAttempt, Message, Status};
pub(crate) const SHORT_WAIT: Duration = Duration::from_millis(1);
pub(crate) const LONG_WAIT: Duration = Duration::from_secs(86400 * 365);
pub struct Queue {
pub core: SmtpInstance,
pub core: Arc<Inner>,
pub on_hold: Vec<OnHold<QueueEventLock>>,
pub next_wake_up: Duration,
}
impl SpawnQueue for mpsc::Receiver<Event> {
fn spawn(mut self, core: SmtpInstance) {
impl SpawnQueue for mpsc::Receiver<QueueEvent> {
fn spawn(mut self, core: Arc<Inner>) {
tokio::spawn(async move {
let mut queue = Queue::new(core);
loop {
let on_hold = match tokio::time::timeout(queue.next_wake_up, self.recv()).await {
Ok(Some(Event::OnHold(on_hold))) => on_hold.into(),
Ok(Some(Event::Stop)) | Ok(None) => {
Ok(Some(QueueEvent::OnHold(on_hold))) => on_hold.into(),
Ok(Some(QueueEvent::Stop)) | Ok(None) => {
break;
}
_ => None,
@@ -48,7 +54,7 @@ impl SpawnQueue for mpsc::Receiver<Event> {
}
impl Queue {
pub fn new(core: SmtpInstance) -> Self {
pub fn new(core: Arc<Inner>) -> Self {
Queue {
core,
on_hold: Vec::with_capacity(128),
@@ -58,20 +64,20 @@ impl Queue {
pub async fn process_events(&mut self) {
// Deliver any concurrency limited messages
let core = SMTP::from(self.core.clone());
let server = self.core.build_server();
while let Some(queue_event) = self.next_on_hold() {
DeliveryAttempt::new(queue_event)
.try_deliver(core.clone())
.try_deliver(server.clone())
.await;
}
// Deliver scheduled messages
let now = now();
self.next_wake_up = LONG_WAIT;
for queue_event in core.next_event().await {
for queue_event in server.next_event().await {
if queue_event.due <= now {
DeliveryAttempt::new(queue_event)
.try_deliver(core.clone())
.try_deliver(server.clone())
.await;
} else {
self.next_wake_up = Duration::from_secs(queue_event.due - now);
@@ -217,5 +223,5 @@ impl Message {
}
pub trait SpawnQueue {
fn spawn(self, core: SmtpInstance);
fn spawn(self, core: Arc<Inner>);
}

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

@@ -12,15 +12,14 @@ use std::{
use common::{
expr::{self, functions::ResolveVariable, *},
listener::limiter::{ConcurrencyLimiter, InFlight},
ipc::QueueEventLock,
listener::limiter::InFlight,
};
use serde::{Deserialize, Serialize};
use smtp_proto::Response;
use store::write::now;
use utils::BlobHash;
use self::spool::QueueEventLock;
pub mod dsn;
pub mod manager;
pub mod quota;
@@ -29,20 +28,6 @@ pub mod throttle;
pub type QueueId = u64;
#[derive(Debug)]
pub enum Event {
Reload,
OnHold(OnHold<QueueEventLock>),
Stop,
}
#[derive(Debug)]
pub struct OnHold<T> {
pub next_due: Option<u64>,
pub limiters: Vec<ConcurrencyLimiter>,
pub message: T,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Schedule<T> {
pub due: u64,

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

@@ -4,19 +4,34 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::{config::smtp::queue::QueueQuota, expr::functions::ResolveVariable};
use std::future::Future;
use common::{config::smtp::queue::QueueQuota, expr::functions::ResolveVariable, Server};
use store::{
write::{BatchBuilder, QueueClass, ValueClass},
ValueKey,
};
use trc::QueueEvent;
use crate::core::{throttle::NewKey, SMTP};
use crate::core::throttle::NewKey;
use super::{Message, QueueEnvelope, QuotaKey, Status};
impl SMTP {
pub async fn has_quota(&self, message: &mut Message) -> bool {
pub trait HasQueueQuota: Sync + Send {
fn has_quota(&self, message: &mut Message) -> impl Future<Output = bool> + Send;
fn check_quota<'x>(
&'x self,
quota: &'x QueueQuota,
envelope: &impl ResolveVariable,
size: usize,
id: u64,
refs: &mut Vec<QuotaKey>,
session_id: u64,
) -> impl Future<Output = bool> + Send;
}
impl HasQueueQuota for Server {
async fn has_quota(&self, message: &mut Message) -> bool {
let mut quota_keys = Vec::new();
if !self.core.smtp.queue.quota.sender.is_empty() {
@@ -110,7 +125,6 @@ impl SMTP {
) -> bool {
if !quota.expr.is_empty()
&& self
.core
.eval_expr(&quota.expr, envelope, "check_quota", session_id)
.await
.unwrap_or(false)

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

@@ -5,32 +5,44 @@
*/
use crate::queue::DomainPart;
use common::ipc::{QueueEvent, QueueEventLock};
use common::Server;
use std::borrow::Cow;
use std::future::Future;
use std::time::{Duration, SystemTime};
use store::write::key::DeserializeBigEndian;
use store::write::{now, BatchBuilder, Bincode, BlobOp, QueueClass, QueueEvent, ValueClass};
use store::write::{now, BatchBuilder, Bincode, BlobOp, QueueClass, ValueClass};
use store::{Deserialize, IterateParams, Serialize, ValueKey, U64_LEN};
use trc::ServerEvent;
use utils::BlobHash;
use crate::core::SMTP;
use super::{
Domain, Event, Message, MessageSource, QueueEnvelope, QueueId, QuotaKey, Recipient, Schedule,
Status,
Domain, Message, MessageSource, QueueEnvelope, QueueId, QuotaKey, Recipient, Schedule, Status,
};
pub const LOCK_EXPIRY: u64 = 300;
#[derive(Debug)]
pub struct QueueEventLock {
pub due: u64,
pub queue_id: u64,
pub lock_expiry: u64,
pub trait SmtpSpool: Sync + Send {
fn new_message(
&self,
return_path: impl Into<String>,
return_path_lcase: impl Into<String>,
return_path_domain: impl Into<String>,
span_id: u64,
) -> Message;
fn next_event(&self) -> impl Future<Output = Vec<QueueEventLock>> + Send;
fn try_lock_event(
&self,
event: QueueEventLock,
) -> impl Future<Output = Option<QueueEventLock>> + Send;
fn read_message(&self, id: QueueId) -> impl Future<Output = Option<Message>> + Send;
}
impl SMTP {
pub fn new_message(
impl SmtpSpool for Server {
fn new_message(
&self,
return_path: impl Into<String>,
return_path_lcase: impl Into<String>,
@@ -41,7 +53,7 @@ impl SMTP {
.duration_since(SystemTime::UNIX_EPOCH)
.map_or(0, |d| d.as_secs());
Message {
queue_id: self.inner.queue_id_gen.generate().unwrap_or(created),
queue_id: self.inner.data.queue_id_gen.generate().unwrap_or(created),
span_id,
created,
return_path: return_path.into(),
@@ -58,22 +70,24 @@ impl SMTP {
}
}
pub async fn next_event(&self) -> Vec<QueueEventLock> {
let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
due: 0,
queue_id: 0,
})));
let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
due: u64::MAX,
queue_id: u64::MAX,
})));
async fn next_event(&self) -> Vec<QueueEventLock> {
let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: 0,
queue_id: 0,
},
)));
let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: u64::MAX,
queue_id: u64::MAX,
},
)));
let mut events = Vec::new();
let now = now();
let result = self
.core
.storage
.data
.store()
.iterate(
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
@@ -107,10 +121,10 @@ impl SMTP {
events
}
pub async fn try_lock_event(&self, mut event: QueueEventLock) -> Option<QueueEventLock> {
async fn try_lock_event(&self, mut event: QueueEventLock) -> Option<QueueEventLock> {
let mut batch = BatchBuilder::new();
batch.assert_value(
ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due: event.due,
queue_id: event.queue_id,
})),
@@ -118,13 +132,13 @@ impl SMTP {
);
event.lock_expiry = now() + LOCK_EXPIRY;
batch.set(
ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due: event.due,
queue_id: event.queue_id,
})),
event.lock_expiry.serialize(),
);
match self.core.storage.data.write(batch.build()).await {
match self.store().write(batch.build()).await {
Ok(_) => Some(event),
Err(err) if err.is_assertion_failure() => {
trc::event!(
@@ -145,11 +159,9 @@ impl SMTP {
}
}
pub async fn read_message(&self, id: QueueId) -> Option<Message> {
async fn read_message(&self, id: QueueId) -> Option<Message> {
match self
.core
.storage
.data
.store()
.get_value::<Bincode<Message>>(ValueKey::from(ValueClass::Queue(QueueClass::Message(
id,
))))
@@ -174,7 +186,7 @@ impl Message {
raw_headers: Option<&[u8]>,
raw_message: &[u8],
session_id: u64,
core: &SMTP,
server: &Server,
source: MessageSource,
) -> bool {
// Write blob
@@ -203,7 +215,7 @@ impl Message {
},
0u32.serialize(),
);
if let Err(err) = core.core.storage.data.write(batch.build()).await {
if let Err(err) = server.store().write(batch.build()).await {
trc::error!(err
.details("Failed to write to store.")
.span_id(session_id)
@@ -211,10 +223,8 @@ impl Message {
return false;
}
if let Err(err) = core
.core
.storage
.blob
if let Err(err) = server
.blob_store()
.put_blob(self.blob_hash.as_slice(), message.as_ref())
.await
{
@@ -271,7 +281,7 @@ impl Message {
}
batch
.set(
ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due: self.next_event().unwrap_or_default(),
queue_id: self.queue_id,
})),
@@ -299,7 +309,7 @@ impl Message {
Bincode::new(self).serialize(),
);
if let Err(err) = core.core.storage.data.write(batch.build()).await {
if let Err(err) = server.store().write(batch.build()).await {
trc::error!(err
.details("Failed to write to store.")
.span_id(session_id)
@@ -309,7 +319,14 @@ impl Message {
}
// Queue the message
if core.inner.queue_tx.send(Event::Reload).await.is_err() {
if server
.inner
.ipc
.queue_tx
.send(QueueEvent::Reload)
.await
.is_err()
{
trc::event!(
Server(ServerEvent::ThreadError),
Reason = "Channel closed.",
@@ -326,7 +343,7 @@ impl Message {
rcpt: impl Into<String>,
rcpt_lcase: impl Into<String>,
rcpt_domain: impl Into<String>,
core: &SMTP,
server: &Server,
) {
let rcpt_domain = rcpt_domain.into();
let domain_idx =
@@ -343,10 +360,9 @@ impl Message {
status: Status::Scheduled,
});
let expires = core
.core
let expires = server
.eval_if(
&core.core.smtp.queue.expire,
&server.core.smtp.queue.expire,
&QueueEnvelope::new(self, idx),
self.span_id,
)
@@ -370,17 +386,17 @@ impl Message {
});
}
pub async fn add_recipient(&mut self, rcpt: impl Into<String>, core: &SMTP) {
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, core)
self.add_recipient_parts(rcpt, rcpt_lcase, rcpt_domain, server)
.await;
}
pub async fn save_changes(
mut self,
core: &SMTP,
server: &Server,
prev_event: Option<u64>,
next_event: Option<u64>,
) -> bool {
@@ -395,12 +411,14 @@ impl Message {
let mut batch = BatchBuilder::new();
if let (Some(prev_event), Some(next_event)) = (prev_event, next_event) {
batch
.clear(ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
due: prev_event,
queue_id: self.queue_id,
})))
.clear(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: prev_event,
queue_id: self.queue_id,
},
)))
.set(
ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
due: next_event,
queue_id: self.queue_id,
})),
@@ -414,7 +432,7 @@ impl Message {
Bincode::new(self).serialize(),
);
if let Err(err) = core.core.storage.data.write(batch.build()).await {
if let Err(err) = server.store().write(batch.build()).await {
trc::error!(err
.details("Failed to save changes.")
.span_id(span_id)
@@ -425,7 +443,7 @@ impl Message {
}
}
pub async fn remove(self, core: &SMTP, prev_event: u64) -> bool {
pub async fn remove(self, server: &Server, prev_event: u64) -> bool {
let mut batch = BatchBuilder::new();
// Release all quotas
@@ -448,13 +466,15 @@ impl Message {
hash: self.blob_hash.clone(),
id: self.queue_id,
})
.clear(ValueClass::Queue(QueueClass::MessageEvent(QueueEvent {
due: prev_event,
queue_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) = core.core.storage.data.write(batch.build()).await {
if let Err(err) = server.store().write(batch.build()).await {
trc::error!(err
.details("Failed to write to update queue.")
.span_id(self.span_id)

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

@@ -4,15 +4,18 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::future::Future;
use common::{
config::smtp::Throttle,
expr::functions::ResolveVariable,
listener::limiter::{ConcurrencyLimiter, InFlight},
Server,
};
use dashmap::mapref::entry::Entry;
use store::write::now;
use crate::core::{throttle::NewKey, SMTP};
use crate::core::throttle::NewKey;
use super::{Domain, Status};
@@ -22,8 +25,18 @@ pub enum Error {
Rate { retry_at: u64 },
}
impl SMTP {
pub async fn is_allowed<'x>(
pub trait IsAllowed: Sync + Send {
fn is_allowed<'x>(
&'x self,
throttle: &'x Throttle,
envelope: &impl ResolveVariable,
in_flight: &mut Vec<InFlight>,
session_id: u64,
) -> impl Future<Output = Result<(), Error>> + Send;
}
impl IsAllowed for Server {
async fn is_allowed<'x>(
&'x self,
throttle: &'x Throttle,
envelope: &impl ResolveVariable,
@@ -32,7 +45,6 @@ impl SMTP {
) -> Result<(), Error> {
if throttle.expr.is_empty()
|| self
.core
.eval_expr(&throttle.expr, envelope, "throttle", session_id)
.await
.unwrap_or(false)
@@ -64,7 +76,7 @@ impl SMTP {
}
if let Some(concurrency) = &throttle.concurrency {
match self.inner.queue_throttle.entry(key) {
match self.inner.data.smtp_queue_throttle.entry(key) {
Entry::Occupied(mut e) => {
let limiter = e.get_mut();
if let Some(inflight) = limiter.is_allowed() {