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

Zero copy tests passing for JMAP

Browse Source
This commit is contained in:
mdecimus
2025-03-01 20:01:48 +01:00
parent 9ec047d284
commit c8abc9213a
67 changed files with 1103 additions and 846 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,13 +5,13 @@
*/
use common::Server;
use mail_builder::headers::content_type::ContentType;
use mail_builder::headers::HeaderType;
use mail_builder::mime::{make_boundary, BodyPart, MimePart};
use mail_builder::MessageBuilder;
use mail_builder::headers::HeaderType;
use mail_builder::headers::content_type::ContentType;
use mail_builder::mime::{BodyPart, MimePart, make_boundary};
use mail_parser::DateTime;
use smtp_proto::{
Response, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS,
RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, Response,
};
use std::fmt::Write;
use std::future::Future;
@@ -23,8 +23,8 @@ 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,
Domain, Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope,
RCPT_DSN_SENT, RCPT_STATUS_CHANGED, Recipient, Status,
};
pub trait SendDsn: Sync + Send {
@@ -80,7 +80,7 @@ impl SendDsn for Server {
continue;
}
let domain = &message.domains[rcpt.domain_idx];
let domain = &message.domains[rcpt.domain_idx as usize];
match &rcpt.status {
Status::Completed(response) => {
trc::event!(
@@ -173,7 +173,7 @@ impl Message {
if rcpt.has_flag(RCPT_DSN_SENT | RCPT_NOTIFY_NEVER) {
continue;
}
let domain = &self.domains[rcpt.domain_idx];
let domain = &self.domains[rcpt.domain_idx as usize];
match &rcpt.status {
Status::Completed(response) => {
rcpt.flags |= RCPT_DSN_SENT | RCPT_STATUS_CHANGED;
@@ -404,10 +404,11 @@ impl Message {
String::new()
}
Err(err) => {
trc::error!(err
.span_id(self.span_id)
.details("Failed to fetch blobId")
.caused_by(trc::location!()));
trc::error!(
err.span_id(self.span_id)
.details("Failed to fetch blobId")
.caused_by(trc::location!())
);
String::new()
}
@@ -452,7 +453,7 @@ impl Message {
is_double_bounce.push(dsn);
}
Status::Scheduled => {
let domain = &self.domains[rcpt.domain_idx];
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();

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

@@ -11,8 +11,7 @@ use std::{
};
use common::expr::{self, functions::ResolveVariable, *};
use serde::{Deserialize, Serialize};
use smtp_proto::Response;
use smtp_proto::{ArchivedResponse, Response};
use store::write::now;
use utils::BlobHash;
@@ -24,7 +23,7 @@ pub mod throttle;
pub type QueueId = u64;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
pub struct Schedule<T> {
pub due: u64,
pub inner: T,
@@ -45,7 +44,7 @@ pub enum MessageSource {
Autogenerated,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Clone, PartialEq, Eq)]
pub struct Message {
pub queue_id: QueueId,
pub created: u64,
@@ -61,20 +60,20 @@ pub struct Message {
pub env_id: Option<String>,
pub priority: i16,
pub size: usize,
pub size: u64,
pub quota_keys: Vec<QuotaKey>,
#[serde(skip)]
#[rkyv(with = rkyv::with::Skip)]
pub span_id: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Clone, PartialEq, Eq)]
pub enum QuotaKey {
Size { key: Vec<u8>, id: u64 },
Count { key: Vec<u8>, id: u64 },
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Clone, PartialEq, Eq)]
pub struct Domain {
pub domain: String,
pub retry: Schedule<u32>,
@@ -83,9 +82,9 @@ pub struct Domain {
pub status: Status<(), Error>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Clone, PartialEq, Eq)]
pub struct Recipient {
pub domain_idx: usize,
pub domain_idx: u32,
pub address: String,
pub address_lcase: String,
pub status: Status<HostResponse<String>, HostResponse<ErrorDetails>>,
@@ -96,7 +95,17 @@ pub struct Recipient {
pub const RCPT_DSN_SENT: u64 = 1 << 32;
pub const RCPT_STATUS_CHANGED: u64 = 2 << 32;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Serialize,
serde::Deserialize,
)]
pub enum Status<T, E> {
#[serde(rename = "scheduled")]
Scheduled,
@@ -108,13 +117,13 @@ pub enum Status<T, E> {
PermanentFailure(E),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
pub struct HostResponse<T> {
pub hostname: T,
pub response: Response<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
pub enum Error {
DnsError(String),
UnexpectedResponse(HostResponse<ErrorDetails>),
@@ -127,7 +136,9 @@ pub enum Error {
Io(String),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
#[derive(
Debug, Clone, PartialEq, Eq, rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Default,
)]
pub struct ErrorDetails {
pub entity: String,
pub details: String,
@@ -419,6 +430,57 @@ impl Display for Error {
}
}
impl Display for ArchivedError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ArchivedError::UnexpectedResponse(response) => {
write!(
f,
"Unexpected response from '{}': {}",
response.hostname.entity,
response.response.to_string()
)
}
ArchivedError::DnsError(err) => {
write!(f, "DNS lookup failed: {err}")
}
ArchivedError::ConnectionError(details) => {
write!(
f,
"Connection to '{}' failed: {}",
details.entity, details.details
)
}
ArchivedError::TlsError(details) => {
write!(
f,
"TLS error from '{}': {}",
details.entity, details.details
)
}
ArchivedError::DaneError(details) => {
write!(
f,
"DANE failed to authenticate '{}': {}",
details.entity, details.details
)
}
ArchivedError::MtaStsError(details) => {
write!(f, "MTA-STS auth failed: {details}")
}
ArchivedError::RateLimited => {
write!(f, "Rate limited")
}
ArchivedError::ConcurrencyLimited => {
write!(f, "Too many concurrent connections to remote server")
}
ArchivedError::Io(err) => {
write!(f, "Queue error: {err}")
}
}
}
}
impl Display for Status<(), Error> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
@@ -440,3 +502,16 @@ impl Display for Status<HostResponse<String>, HostResponse<ErrorDetails>> {
}
}
}
pub trait DisplayArchivedResponse {
fn to_string(&self) -> String;
}
impl DisplayArchivedResponse for ArchivedResponse<String> {
fn to_string(&self) -> String {
format!(
"Code: {}, Enhanced code: {}.{}.{}, Message: {}",
self.code, self.esc[0], self.esc[1], self.esc[2], self.message,
)
}
}

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

@@ -6,10 +6,10 @@
use std::future::Future;
use common::{config::smtp::queue::QueueQuota, expr::functions::ResolveVariable, Server};
use common::{Server, config::smtp::queue::QueueQuota, expr::functions::ResolveVariable};
use store::{
write::{BatchBuilder, QueueClass, ValueClass},
ValueKey,
write::{BatchBuilder, QueueClass, ValueClass},
};
use trc::QueueEvent;
@@ -23,7 +23,7 @@ pub trait HasQueueQuota: Sync + Send {
&'x self,
quota: &'x QueueQuota,
envelope: &impl ResolveVariable,
size: usize,
size: u64,
id: u64,
refs: &mut Vec<QuotaKey>,
session_id: u64,
@@ -89,7 +89,7 @@ impl HasQueueQuota for Server {
if !self
.check_quota(
quota,
&QueueEnvelope::new_rcpt(message, rcpt.domain_idx, rcpt_idx),
&QueueEnvelope::new_rcpt(message, rcpt.domain_idx as usize, rcpt_idx),
message.size,
(rcpt_idx + 1) as u64,
&mut quota_keys,
@@ -118,7 +118,7 @@ impl HasQueueQuota for Server {
&'x self,
quota: &'x QueueQuota,
envelope: &impl ResolveVariable,
size: usize,
size: u64,
id: u64,
refs: &mut Vec<QuotaKey>,
session_id: u64,
@@ -139,7 +139,7 @@ impl HasQueueQuota for Server {
key.as_ref().to_vec(),
))))
.await
.unwrap_or(0) as usize;
.unwrap_or(0) as u64;
if used_size + size > max_size {
return false;
} else {
@@ -159,7 +159,7 @@ impl HasQueueQuota for Server {
key.as_ref().to_vec(),
))))
.await
.unwrap_or(0) as usize;
.unwrap_or(0) as u64;
if total_messages + 1 > max_messages {
return false;
} else {

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

@@ -11,14 +11,14 @@ use std::borrow::Cow;
use std::future::Future;
use std::time::{Duration, SystemTime};
use store::write::key::DeserializeBigEndian;
use store::write::{BatchBuilder, BlobOp, LegacyBincode, QueueClass, ValueClass, now};
use store::write::{Archive, Archiver, BatchBuilder, BlobOp, QueueClass, ValueClass, now};
use store::{IterateParams, Serialize, SerializeInfallible, U64_LEN, ValueKey};
use trc::ServerEvent;
use utils::BlobHash;
use super::{
Domain, Message, MessageSource, QueueEnvelope, QueueId, QueuedMessage, QuotaKey, Recipient,
Schedule, Status,
ArchivedMessage, ArchivedStatus, Domain, Message, MessageSource, QueueEnvelope, QueueId,
QueuedMessage, QuotaKey, Recipient, Schedule, Status,
};
pub const LOCK_EXPIRY: u64 = 300;
@@ -40,6 +40,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_archive(
&self,
id: QueueId,
) -> impl Future<Output = trc::Result<Option<Archive>>> + Send;
}
impl SmtpSpool for Server {
@@ -149,14 +154,11 @@ impl SmtpSpool for Server {
}
async fn read_message(&self, id: QueueId) -> Option<Message> {
match self
.store()
.get_value::<LegacyBincode<Message>>(ValueKey::from(ValueClass::Queue(
QueueClass::Message(id),
)))
.await
{
Ok(Some(message)) => Some(message.inner),
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),
Ok(None) => None,
Err(err) => {
trc::error!(
@@ -168,6 +170,12 @@ impl SmtpSpool for Server {
}
}
}
async fn read_message_archive(&self, id: QueueId) -> trc::Result<Option<Archive>> {
self.store()
.get_value::<Archive>(ValueKey::from(ValueClass::Queue(QueueClass::Message(id))))
.await
}
}
impl Message {
@@ -192,7 +200,7 @@ impl Message {
// Generate id
if self.size == 0 {
self.size = message.len();
self.size = message.len() as u64;
}
// Reserve and write blob
@@ -298,7 +306,7 @@ impl Message {
)
.set(
ValueClass::Queue(QueueClass::Message(self.queue_id)),
match LegacyBincode::new(self).serialize() {
match Archiver::new(self).serialize() {
Ok(data) => data,
Err(err) => {
trc::error!(
@@ -380,7 +388,7 @@ impl Message {
idx
};
self.recipients.push(Recipient {
domain_idx,
domain_idx: domain_idx as u32,
address: rcpt.into(),
address_lcase: rcpt_lcase.into(),
status: Status::Scheduled,
@@ -430,7 +438,7 @@ impl Message {
let span_id = self.span_id;
batch.set(
ValueClass::Queue(QueueClass::Message(self.queue_id)),
match LegacyBincode::new(self).serialize() {
match Archiver::new(self).serialize() {
Ok(data) => data,
Err(err) => {
trc::error!(
@@ -506,3 +514,37 @@ impl Message {
.is_some_and(|(_, domain)| domains.contains(&domain.to_string()))
}
}
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.contains(&domain.to_string()))
}
pub fn next_delivery_event(&self) -> u64 {
let mut next_delivery = now();
for (pos, domain) in self
.domains
.iter()
.filter(|d| {
matches!(
d.status,
ArchivedStatus::Scheduled | ArchivedStatus::TemporaryFailure(_)
)
})
.enumerate()
{
if pos == 0 || domain.retry.due < next_delivery {
next_delivery = domain.retry.due.into();
}
}
next_delivery
}
}