feat(transport): Destination worker pool

Implements a per-destination worker pool,
so that connections to the same destination
are not parallelized, but instead queued.
If a queue is full, new messages are immediately
deferred, before mail data is sent from postfix.

Closes: #141

Signed-off-by: Jagoda Ślązak <jslazak@jslazak.com>

filtermail/src/http_server.rs

@@ -1,4 +1,4 @@
-use crate::smtp_server::{Envelope, SmtpHandler};
+use crate::smtp_server::{SmtpHandler, Transaction};
 use http_body_util::combinators::BoxBody;
 use http_body_util::{BodyExt, Full};
 use hyper::body::{Bytes, Incoming};
@@ -93,6 +93,8 @@ impl<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
                 )?);
             }
 
+            let mut transaction = Transaction::default();
+
             let mail_from = req
                 .headers()
                 .get(crate::transport::HEADER_MAIL_FROM)
@@ -100,16 +102,14 @@ impl<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
                 .unwrap_or("")
                 .to_string();
 
-            match handler.handle_mail(&mail_from) {
+            if let Err(e) = handler.handle_mail_from(&mail_from) {
-                Ok(_) => {}
+                return Ok(Response::builder()
-                Err(e) => {
+                    .status(400)
-                    return Ok(Response::builder()
+                    .body(Full::new(Bytes::from(e)).boxed())?);
-                        .status(400)
+            }
-                        .body(Full::new(Bytes::from(e)).boxed())?);
+            transaction.envelope.mail_from = mail_from;
-                }
-            };
 
-            let rcpt_to = req
+            let rcpt_to: Vec<String> = req
                 .headers()
                 .get_all(crate::transport::HEADER_RCPT_TO)
                 .iter()
@@ -117,6 +117,21 @@ impl<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
                 .map(ToString::to_string)
                 .collect();
 
+            for r in &rcpt_to {
+                if let Err(e) = handler.handle_rcpt_to(r, &mut transaction) {
+                    return Ok(Response::builder()
+                        .status(400)
+                        .body(Full::new(Bytes::from(e)).boxed())?);
+                }
+            }
+            transaction.envelope.rcpt_to = rcpt_to;
+
+            if let Err(e) = handler.handle_data_start(&transaction) {
+                return Ok(Response::builder()
+                    .status(400)
+                    .body(Full::new(Bytes::from(e)).boxed())?);
+            }
+
             let body_limited = http_body_util::Limited::new(req.into_body(), max_size);
             let body_bytes = match body_limited.collect().await {
                 Ok(body) => body.to_bytes(),
@@ -127,24 +142,19 @@ impl<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
                 }
             };
 
-            let mut envelope = Envelope {
+            transaction.envelope.data = body_bytes.to_vec();
-                origin_ip: "".to_string(),
-                mail_from,
-                rcpt_to,
-                data: body_bytes.to_vec(),
-            };
 
-            log::debug!("(HTTP) MAIL FROM:<{}>", envelope.mail_from);
+            log::debug!("(HTTP) MAIL FROM:<{}>", transaction.envelope.mail_from);
-            for rcpt in &envelope.rcpt_to {
+            for rcpt in &transaction.envelope.rcpt_to {
                 log::debug!("(HTTP) RCPT TO:<{}>", rcpt);
             }
 
             log::trace!(
                 "(HTTP) DATA:\n{:?}",
-                String::from_utf8_lossy(&envelope.data)
+                String::from_utf8_lossy(&transaction.envelope.data)
             );
 
-            match handler.handle_data(&mut envelope).await {
+            match handler.handle_data_dot(&mut transaction).await {
                 Ok(response) => Ok(Response::builder()
                     .status(200)
                     .body(Full::new(Bytes::from(response)).boxed())?),

filtermail/src/inbound.rs

@@ -1,12 +1,12 @@
 //! Module for handling incoming SMTP messages.
 
-use crate::ENCRYPTION_NEEDED_523;
 use crate::config::Config;
 use crate::dkim_verifier::DkimVerifier;
 use crate::message::{check_encrypted, is_securejoin};
 use crate::smtp_client::SmtpConnectionPool;
+use crate::smtp_responses::ENCRYPTION_NEEDED_523;
 pub use crate::smtp_server::Envelope;
-use crate::smtp_server::SmtpHandler;
+use crate::smtp_server::{SmtpHandler, Transaction};
 use crate::utils::{AddressDomain, build_resolver, extract_address, log_eml};
 use async_trait::async_trait;
 use hickory_resolver::TokioResolver;
@@ -69,12 +69,10 @@ impl IncomingBeforeQueueHandler {
 
 #[async_trait]
 impl SmtpHandler for IncomingBeforeQueueHandler {
-    fn handle_mail(&self, _address: &str) -> Result<(), String> {
+    type State = ();
-        Ok(())
-    }
 
-    async fn check_data(&self, envelope: &mut Envelope) -> Result<(), String> {
+    async fn check_data(&self, transaction: &mut Transaction<Self::State>) -> Result<(), String> {
-        let message = match parse_mail(&envelope.data) {
+        let message = match parse_mail(&transaction.envelope.data) {
             Ok(m) => m,
             Err(e) => return Err(format!("500 Failed to parse message: {}", e)),
         };
@@ -92,16 +90,21 @@ impl SmtpHandler for IncomingBeforeQueueHandler {
 
         log::debug!("Processing DATA message from {from_addr}");
 
-        if !envelope.mail_from.eq_ignore_ascii_case(&from_addr) {
+        if !transaction
+            .envelope
+            .mail_from
+            .eq_ignore_ascii_case(&from_addr)
+        {
             // If the MAIL FROM doesn't match the From header, we do not reject the mail,
             // as this can be caused by e.g. SRS forwarding.
             // Instead, we reset the envelope address, so it is reinjected as
             // `MAIL FROM:<>` to prevent sending a bounce message.
             // <https://github.com/chatmail/filtermail/issues/67>
-            envelope.mail_from = String::new();
+            transaction.envelope.mail_from = String::new();
         }
 
-        envelope.rcpt_to = envelope
+        transaction.envelope.rcpt_to = transaction
+            .envelope
             .rcpt_to
             .iter()
             .filter(|s| {
@@ -121,7 +124,7 @@ impl SmtpHandler for IncomingBeforeQueueHandler {
         // Allow encrypted or securejoin messages
         if mail_encrypted || is_securejoin(&message) {
             log::info!("Incoming: Filtering encrypted mail.");
-            return self.verify_origin(envelope, &from_addr).await;
+            return self.verify_origin(&transaction.envelope, &from_addr).await;
         }
 
         log::info!("Incoming: Filtering unencrypted mail.");
@@ -132,26 +135,26 @@ impl SmtpHandler for IncomingBeforeQueueHandler {
             && from_addr.to_lowercase().starts_with("mailer-daemon@")
             && message.ctype.mimetype == "multipart/report"
         {
-            return self.verify_origin(envelope, &from_addr).await;
+            return self.verify_origin(&transaction.envelope, &from_addr).await;
         }
 
-        for recipient in &envelope.rcpt_to {
+        for recipient in &transaction.envelope.rcpt_to {
             if !self.config.is_cleartext_ok(recipient) {
                 log::warn!("Rejected unencrypted mail from: {from_addr}");
                 return Err(ENCRYPTION_NEEDED_523.to_string());
             }
         }
 
-        self.verify_origin(envelope, &from_addr).await
+        self.verify_origin(&transaction.envelope, &from_addr).await
     }
 
-    async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String> {
+    async fn reinject_mail(&self, transaction: &Transaction<Self::State>) -> Result<(), String> {
         log::debug!("Re-injecting the mail that passed checks");
         let hostname = format!("[{}]", self.config.filtermail_host);
         crate::smtp_client::send(
             &self.config.postfix_host,
             self.config.postfix_reinject_port_incoming,
-            envelope,
+            &transaction.envelope,
             &hostname,
             None,
             self.dns_resolver.clone(),
@@ -189,12 +192,14 @@ mod tests {
         config: Config,
     ) -> TestResult {
         let handler = IncomingBeforeQueueHandler::new(config, false)?;
-        let mut envelope = Envelope {
+        let mut transaction = Transaction {
-            mail_from: address.to_string(),
+            envelope: Envelope {
-            origin_ip: "".to_string(), // Currently shouldn't be relevant.
+                mail_from: address.to_string(),
-            data: eml.to_vec(),
+                data: eml.to_vec(),
-            rcpt_to: vec!["does.not.matter@example.org".to_string()],
+                rcpt_to: vec!["does.not.matter@example.org".to_string()],
+            },
+            ..Default::default()
         };
-        Ok(handler.check_data(&mut envelope).await?)
+        Ok(handler.check_data(&mut transaction).await?)
     }
 }

filtermail/src/main.rs

@@ -33,6 +33,7 @@ pub(crate) mod message;
 pub(crate) mod openpgp;
 pub(crate) mod outbound;
 pub(crate) mod smtp_client;
+mod smtp_responses;
 pub(crate) mod smtp_server;
 mod tls;
 mod transport;
@@ -50,8 +51,6 @@ use std::process;
 use std::str::FromStr;
 use std::sync::Arc;
 
-const ENCRYPTION_NEEDED_523: &str = "523 Encryption Needed: Invalid Unencrypted Mail";
-
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 enum Mode {
     Outgoing,

filtermail/src/outbound.rs

@@ -1,11 +1,11 @@
 //! Module for handling outgoing SMTP messages.
 
-use crate::ENCRYPTION_NEEDED_523;
 use crate::config::Config;
 use crate::message::{check_encrypted, is_securejoin};
 use crate::smtp_client::SmtpConnectionPool;
-pub use crate::smtp_server::Envelope;
+use crate::smtp_responses::ENCRYPTION_NEEDED_523;
-use crate::smtp_server::SmtpHandler;
+use crate::smtp_responses::OK_250;
+use crate::smtp_server::{SmtpHandler, Transaction};
 use crate::utils::{build_resolver, extract_address};
 use async_trait::async_trait;
 use governor::clock::MonotonicClock;
@@ -53,7 +53,9 @@ impl OutgoingBeforeQueueHandler {
 
 #[async_trait]
 impl SmtpHandler for OutgoingBeforeQueueHandler {
-    fn handle_mail(&self, address: &str) -> Result<(), String> {
+    type State = ();
+
+    fn handle_mail_from(&self, address: &str) -> Result<(), String> {
         log::debug!("handle_MAIL from {address}");
 
         let parts: Vec<&str> = address.split('@').collect();
@@ -79,8 +81,8 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
         Ok(())
     }
 
-    async fn check_data(&self, envelope: &mut Envelope) -> Result<(), String> {
+    async fn check_data(&self, transaction: &mut Transaction<Self::State>) -> Result<(), String> {
-        let message = match parse_mail(&envelope.data) {
+        let message = match parse_mail(&transaction.envelope.data) {
             Ok(m) => m,
             Err(e) => return Err(format!("500 Failed to parse message: {}", e)),
         };
@@ -97,7 +99,8 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
         let from_addr = extract_address(&from_header)
             .ok_or(format!("500 Invalid FROM header: {from_header}"))?;
 
-        envelope.rcpt_to = envelope
+        transaction.envelope.rcpt_to = transaction
+            .envelope
             .rcpt_to
             .iter()
             .filter(|s| {
@@ -113,12 +116,19 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
         // MAIL FROM is our source of truth for outbound messages,
         // as this address is checked by postfix against the username before sending it
         // to filtermail.
-        log::debug!("Processing DATA message from {}", envelope.mail_from);
+        log::debug!(
+            "Processing DATA message from {}",
+            transaction.envelope.mail_from
+        );
 
-        if !envelope.mail_from.eq_ignore_ascii_case(&from_addr) {
+        if !transaction
+            .envelope
+            .mail_from
+            .eq_ignore_ascii_case(&from_addr)
+        {
             return Err(format!(
                 "500 Invalid FROM <{}> for <{}>",
-                from_addr, envelope.mail_from
+                from_addr, transaction.envelope.mail_from
             ));
         }
 
@@ -131,8 +141,8 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
         log::info!("Outgoing: Filtering unencrypted mail.");
 
         // Allow self-sent Autocrypt Setup Message
-        if envelope.rcpt_to.len() == 1
+        if transaction.envelope.rcpt_to.len() == 1
-            && let Some(rcpt_to) = envelope.rcpt_to.first()
+            && let Some(rcpt_to) = transaction.envelope.rcpt_to.first()
             && *rcpt_to == from_addr
         {
             let subject = message
@@ -148,13 +158,13 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
         Err(ENCRYPTION_NEEDED_523.to_string())
     }
 
-    async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String> {
+    async fn reinject_mail(&self, transaction: &Transaction<Self::State>) -> Result<(), String> {
         log::debug!("Re-injecting the mail that passed checks");
         let hostname = format!("[{}]", self.config.filtermail_host);
         crate::smtp_client::send(
             &self.config.postfix_host,
             self.config.postfix_reinject_port,
-            envelope,
+            &transaction.envelope,
             &hostname,
             None,
             self.dns_resolver.clone(),
@@ -169,21 +179,27 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
         Ok(())
     }
 
-    async fn handle_data(&self, envelope: &mut Envelope) -> Result<String, String> {
+    async fn handle_data_dot(
+        &self,
+        transaction: &mut Transaction<Self::State>,
+    ) -> Result<String, String> {
         log::debug!("handle_DATA before-queue");
-        self.check_data(envelope).await?;
+        self.check_data(transaction).await?;
-        if self.config.is_disabled(&envelope.mail_from) {
+        if self.config.is_disabled(&transaction.envelope.mail_from) {
-            log::warn!("Dropping mail; Sender {} is disabled.", envelope.mail_from);
+            log::warn!(
-            return Ok("250 OK".to_string());
+                "Dropping mail; Sender {} is disabled.",
+                transaction.envelope.mail_from
+            );
+            return Ok(OK_250.to_string());
         }
-        if envelope.rcpt_to.is_empty() {
+        if transaction.envelope.rcpt_to.is_empty() {
             log::warn!("Dropping mail; All recipients disabled.");
-            return Ok("250 OK".to_string());
+            return Ok(OK_250.to_string());
         }
-        self.reinject_mail(envelope).await.map_err(|e| {
+        self.reinject_mail(transaction).await.map_err(|e| {
             log::warn!("Failed to reinject mail: {e}");
             e
         })?;
-        Ok("250 OK".to_string())
+        Ok(OK_250.to_string())
     }
 }

filtermail/src/smtp_responses.rs

@@ -0,0 +1,7 @@
+pub const OK_250: &str = "250 OK";
+pub const OK_HTTPS_250: &str = "250 OK (HTTPS)";
+pub const OK_SMTP_250: &str = "250 OK (SMTP)";
+
+pub const ENCRYPTION_NEEDED_523: &str = "523 Encryption Needed: Invalid Unencrypted Mail";
+pub const LOCAL_ERROR_451: &str = "451 Local error";
+pub const WORKER_BUSY_421: &str = "421 Worker for this destination is busy";

filtermail/src/smtp_server.rs

@@ -1,8 +1,10 @@
 //! A simplified SMTP server implementation for internal communication.
 
+use crate::smtp_responses::OK_250;
 use crate::utils::{extract_address, log_eml};
 use async_trait::async_trait;
 use memchr::{Memchr, memmem};
+use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
@@ -12,9 +14,7 @@ use tokio::net::{TcpListener, TcpStream};
 #[derive(Debug, Default, Clone)]
 pub struct Envelope {
     pub mail_from: String,
-    pub origin_ip: String,
     pub rcpt_to: Vec<String>,
-
     /// Mail data as transmitted over SMTP/LMTP.
     ///
     /// Described in <https://www.rfc-editor.org/rfc/rfc5321.html#section-2.3.9>.
@@ -25,6 +25,16 @@ pub struct Envelope {
     pub data: Vec<u8>,
 }
 
+/// Represent an ongoing SMTP transaction.
+///
+/// Every new connection starts with an empty envelope and handler state.
+/// A RSET command starts a new transaction, which clears the envelope and state.
+#[derive(Debug, Default)]
+pub struct Transaction<S: Debug + Default> {
+    pub envelope: Envelope,
+    pub state: S,
+}
+
 /// Checks if mail data is valid.
 fn is_valid_data(data: &[u8]) -> bool {
     // DATA must end with <CRLF>.
@@ -59,19 +69,55 @@ fn is_valid_data(data: &[u8]) -> bool {
 /// Trait defining the SMTP handler interface.
 #[async_trait]
 pub trait SmtpHandler: Send + Sync {
-    /// Handles the MAIL FROM command.
+    /// Transaction state type associated with this handler.
-    fn handle_mail(&self, address: &str) -> Result<(), String>;
+    type State: Debug + Default + Send;
 
     /// Checks the DATA command before reinjection.
     ///
     /// Can optionally modify the envelope before reinjection.
-    async fn check_data(&self, envelope: &mut Envelope) -> Result<(), String>;
+    ///
+    /// Default implementation is no-op.
+    async fn check_data(&self, _transaction: &mut Transaction<Self::State>) -> Result<(), String> {
+        Ok(())
+    }
 
     /// Reinjects the mail back to postfix.
-    async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String>;
+    ///
+    /// Default implementation is no-op.
+    async fn reinject_mail(&self, _transaction: &Transaction<Self::State>) -> Result<(), String> {
+        Ok(())
+    }
 
-    /// Handles the DATA command.
+    /// Handles the MAIL FROM command.
-    async fn handle_data(&self, envelope: &mut Envelope) -> Result<String, String> {
+    ///
+    /// Default implementation is no-op.
+    fn handle_mail_from(&self, _address: &str) -> Result<(), String> {
+        Ok(())
+    }
+
+    /// Handles the RCPT TO command.
+    ///
+    /// Default implementation is no-op.
+    fn handle_rcpt_to(
+        &self,
+        _address: &str,
+        _transaction: &mut Transaction<Self::State>,
+    ) -> Result<(), String> {
+        Ok(())
+    }
+
+    /// Handles the DATA command. Called after receiving DATA, before receiving actual data.
+    ///
+    /// Default implementation is no-op.
+    fn handle_data_start(&self, _transaction: &Transaction<Self::State>) -> Result<(), String> {
+        Ok(())
+    }
+
+    /// Handles the end of DATA command. Called after receiving the final dot.
+    async fn handle_data_dot(
+        &self,
+        transaction: &mut Transaction<Self::State>,
+    ) -> Result<String, String> {
         log::debug!("handle_DATA before-queue");
 
         // Check if the DATA is valid
@@ -80,20 +126,20 @@ pub trait SmtpHandler: Send + Sync {
         // We are not going to normalize newlines
         // and escape the dots in the mail data.
         // If mail data turned out to be invalid, reject immediately.
-        if !is_valid_data(&envelope.data) {
+        if !is_valid_data(&transaction.envelope.data) {
             return Err("500 Invalid DATA".to_string());
         }
 
-        self.check_data(envelope).await?;
+        self.check_data(transaction).await?;
-        if envelope.rcpt_to.is_empty() {
+        if transaction.envelope.rcpt_to.is_empty() {
             log::warn!("Dropping mail; All recipients disabled.");
-            return Ok("250 OK".to_string());
+            return Ok(OK_250.to_string());
         }
-        self.reinject_mail(envelope).await.map_err(|e| {
+        self.reinject_mail(transaction).await.map_err(|e| {
             log::warn!("Failed to reinject mail: {e}");
             e
         })?;
-        Ok("250 OK".to_string())
+        Ok("OK_250".to_string())
     }
 }
 
@@ -150,7 +196,7 @@ where
     writer.write_all(b"220 filtermail SMTP\r\n").await?;
     writer.flush().await?;
 
-    let mut envelope = Envelope::default();
+    let mut transaction = Transaction::default();
 
     'connection: loop {
         line.clear();
@@ -181,28 +227,24 @@ where
             || cmd.to_uppercase().starts_with("LHLO")
         {
             writer
-                .write_all(b"250-filtermail\r\n250-XFORWARD ADDR\r\n250-8BITMIME\r\n250 OK\r\n")
+                .write_all(b"250-filtermail\r\n250-8BITMIME\r\n250 OK\r\n")
                 .await?;
             writer.flush().await?;
         } else if cmd.to_uppercase().starts_with("MAIL FROM:<>") {
             // bounce message
-            envelope.mail_from = String::new();
+            transaction.envelope.mail_from = String::new();
-            writer.write_all(b"250 OK\r\n").await?;
+            writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
             writer.flush().await?;
         } else if cmd.to_uppercase().starts_with("MAIL FROM:") {
             if let Some(from) = extract_address(cmd) {
-                match handler.handle_mail(&from) {
+                if let Err(e) = handler.handle_mail_from(&from) {
-                    Ok(_) => {
+                    writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
-                        envelope.mail_from = from;
+                    writer.flush().await?;
-                        writer.write_all(b"250 OK\r\n").await?;
+                    continue 'connection;
-                        writer.flush().await?;
-                    }
-                    Err(e) => {
-                        writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
-                        writer.flush().await?;
-                        break 'connection;
-                    }
                 }
+                transaction.envelope.mail_from = from;
+                writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
+                writer.flush().await?;
             } else {
                 log::warn!("Invalid MAIL FROM command. Can't extract address. Received: {cmd}");
                 writer
@@ -212,11 +254,21 @@ where
             }
         } else if cmd.to_uppercase().starts_with("RCPT TO:") {
             if let Some(to) = extract_address(cmd) {
-                envelope.rcpt_to.push(to);
+                if let Err(e) = handler.handle_rcpt_to(&to, &mut transaction) {
-                writer.write_all(b"250 OK\r\n").await?;
+                    writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
+                    writer.flush().await?;
+                    continue 'connection;
+                }
+                transaction.envelope.rcpt_to.push(to);
+                writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
                 writer.flush().await?;
             }
         } else if cmd.to_uppercase().starts_with("DATA") {
+            if let Err(e) = handler.handle_data_start(&transaction) {
+                writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
+                writer.flush().await?;
+                continue 'connection;
+            }
             writer
                 .write_all(b"354 End data with <CR><LF>.<CR><LF>\r\n")
                 .await?;
@@ -255,14 +307,14 @@ where
                         .write_all(b"552 Message exceeds maximum size\r\n")
                         .await?;
                     writer.flush().await?;
-                    break 'connection;
+                    continue 'connection;
                 }
             }
 
-            envelope.data = data;
+            transaction.envelope.data = data;
 
             // Process the message
-            match handler.handle_data(&mut envelope).await {
+            match handler.handle_data_dot(&mut transaction).await {
                 Ok(response) => {
                     log::debug!("Sent: {response}");
                     writer
@@ -277,29 +329,17 @@ where
                 }
             }
 
-            envelope = Envelope::default();
+            transaction = Transaction::default();
-        } else if cmd.to_uppercase().starts_with("XFORWARD") {
-            // https://www.postfix.org/XFORWARD_README.html
-            if let Some(addr_part) = cmd
-                .split_whitespace()
-                .find(|part| part.to_uppercase().starts_with("ADDR="))
-                && let Some(ip) = addr_part.strip_prefix("ADDR=")
-            {
-                let ip = ip.to_lowercase();
-                envelope.origin_ip = ip.strip_prefix("ipv6:").unwrap_or(&ip).to_string();
-                writer.write_all(b"250 OK\r\n").await?;
-                writer.flush().await?;
-            }
         } else if cmd.to_uppercase().starts_with("QUIT") {
             writer.write_all(b"221 OK\r\n").await?;
             writer.flush().await?;
             break 'connection;
         } else if cmd.to_uppercase().starts_with("RSET") {
-            envelope = Envelope::default();
+            transaction = Transaction::default();
-            writer.write_all(b"250 OK\r\n").await?;
+            writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
             writer.flush().await?;
         } else if cmd.to_uppercase().starts_with("NOOP") {
-            writer.write_all(b"250 OK\r\n").await?;
+            writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
             writer.flush().await?;
         } else {
             writer.write_all(b"500 Command not recognized\r\n").await?;

filtermail/src/transport.rs

@@ -1,375 +1,112 @@
+mod https_client;
+mod worker;
+
 use crate::config::Config;
-use crate::smtp_client::{SmtpConnectionPool, TlsConfig};
+use crate::smtp_responses::{LOCAL_ERROR_451, WORKER_BUSY_421};
-use crate::smtp_server::{Envelope, SmtpHandler};
+use crate::smtp_server::{SmtpHandler, Transaction};
-use crate::tls;
+use crate::utils::AddressDomain;
-use crate::utils::{AddressDomain, build_resolver};
 use async_trait::async_trait;
-use hickory_resolver::{TokioResolver, proto::rr::RData};
-use http_body_util::BodyExt;
-use hyper::body::Bytes;
-use hyper_rustls::HttpsConnector;
-use hyper_util::client::legacy::connect::HttpConnector;
 use std::collections::BTreeMap;
 use std::str::FromStr;
-use std::sync::Arc;
+use tokio::sync::mpsc::OwnedPermit;
-use std::time::Duration;
+use tokio::task::JoinSet;
-use tokio::task::{JoinHandle, JoinSet};
+use worker::{WorkerMessage, WorkerPool};
-use tokio_rustls::rustls;
 
 pub const HEADER_MAIL_FROM: &str = "X-MAIL-FROM";
 pub const HEADER_RCPT_TO: &str = "X-MAIL-TO";
 
-/// Cheaply clonable HTTPS client.
-///
-/// Holds regular secure variant and relaxed - without certificate verification.
-///
-/// Connection pool handled internally by [`hyper_util::client::legacy::Client`].
-#[derive(Clone)]
-struct HttpsClient {
-    pub secure: hyper_util::client::legacy::Client<
-        HttpsConnector<HttpConnector>,
-        http_body_util::Full<Bytes>,
-    >,
-    pub relaxed: hyper_util::client::legacy::Client<
-        HttpsConnector<HttpConnector>,
-        http_body_util::Full<Bytes>,
-    >,
-}
-
-impl HttpsClient {
-    /// Creates a new `[HttpsClient]`.
-    pub fn new(
-        tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
-    ) -> Result<Self, crate::error::Error> {
-        let tls_client_config = tls::configure_rustls(tls_resumption_store.clone(), false)?;
-        let https_connector = hyper_rustls::HttpsConnectorBuilder::new()
-            .with_tls_config(tls_client_config)
-            .https_only()
-            .enable_http1()
-            .enable_http2()
-            .build();
-        let https_client =
-            hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
-                .build(https_connector);
-
-        let tls_client_config_relaxed = tls::configure_rustls(tls_resumption_store, true)?;
-        let https_connector_relaxed = hyper_rustls::HttpsConnectorBuilder::new()
-            .with_tls_config(tls_client_config_relaxed)
-            .https_only()
-            .enable_http1()
-            .enable_http2()
-            .build();
-        let https_client_relaxed =
-            hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
-                .build(https_connector_relaxed);
-
-        Ok(Self {
-            secure: https_client,
-            relaxed: https_client_relaxed,
-        })
-    }
-}
-
 pub struct TransportHandler {
-    config: Config,
+    workers: WorkerPool,
-    dns_resolver: Arc<TokioResolver>,
-    tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
-    smtp_connection_pool: Arc<SmtpConnectionPool>,
-    https_client: HttpsClient,
-    mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, bool>>,
-    monitor_handle: JoinHandle<()>,
 }
 
 impl TransportHandler {
+    /// Creates a new [`TransportHandler`].
     pub fn new(config: Config) -> Result<Self, crate::error::Error> {
-        let dns_resolver = Arc::new(build_resolver()?);
+        let workers = WorkerPool::new(config)?;
-        let tls_resumption_store = Arc::new(rustls::client::ClientSessionMemoryCache::new(256));
-        let https_client = HttpsClient::new(tls_resumption_store.clone())?;
 
-        let mxdeliv_cache = Arc::new(retainer::Cache::new());
+        Ok(Self { workers })
-        let mxdeliv_cache_clone = mxdeliv_cache.clone();
-
-        let monitor_handle = tokio::spawn(async move {
-            mxdeliv_cache_clone
-                .monitor(4, 0.25, Duration::from_secs(10))
-                .await
-        });
-
-        Ok(Self {
-            config,
-            dns_resolver,
-            tls_resumption_store,
-            smtp_connection_pool: SmtpConnectionPool::new(),
-            https_client,
-            mxdeliv_unsupported_hosts: mxdeliv_cache,
-            monitor_handle,
-        })
     }
 
-    /// Handles a single email transaction for a single recipient domain.
+    /// Same as [`Self::new`], but lets you set worker queue size.
-    #[expect(clippy::too_many_arguments)]
-    async fn handle_single_domain(
-        tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
-        smtp_connection_pool: Arc<SmtpConnectionPool>,
-        mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, bool>>,
-        https_client: HttpsClient,
-        dns_resolver: Arc<TokioResolver>,
-        domain: AddressDomain,
-        envelope: Envelope,
-        client_hostname: String,
-    ) -> Result<String, String> {
-        let mut allow_invalid_cert = false;
-        let mut skip_tls = false; // only respected by smtp channel
-
-        let mx_hosts = match domain {
-            // no-DNS setup; assume the ip from email address is the destination.
-            AddressDomain::Literal(ip) => {
-                // We allow self-signed certs on IP-based relays.
-                allow_invalid_cert = true;
-                vec![(0, ip)]
-            }
-            AddressDomain::Name(mx_domain) => {
-                if mx_domain.eq_ignore_ascii_case("nauta.cu") {
-                    // Special case; We don't want to defederate nauta.cu,
-                    // which doesn't support STARTTLS at all.
-                    skip_tls = true;
-                } else if mx_domain.starts_with('_') {
-                    // We use domains starting with `_` for test deployments.
-                    // (You can't request a non-wildcard cert for such domain)
-                    allow_invalid_cert = true;
-                }
-                let query = format!("{mx_domain}.");
-
-                match dns_resolver.mx_lookup(query).await {
-                    Ok(mx_records) => {
-                        let mut hosts: Vec<(u16, String)> = Vec::new();
-                        for mx_record in mx_records.answers() {
-                            let mx = match mx_record.data {
-                                RData::MX(ref mx) => mx,
-                                _ => continue,
-                            };
-
-                            // Null MX / RFC7505
-                            if mx.exchange.is_root() {
-                                // From RFC7505 section 3:
-                                // > A domain that advertises a null MX MUST NOT
-                                // > advertise any other MX RR.
-                                // We assume this is the only record and exit early.
-                                return Err(
-                                    "556 5.1.10 Permanent failure: Recipient address has null MX"
-                                        .to_string(),
-                                );
-                            }
-
-                            let host = mx.exchange.to_string().trim_end_matches('.').to_string();
-                            hosts.push((mx.preference, host))
-                        }
-                        hosts.sort();
-                        hosts
-                    }
-                    Err(e) => {
-                        if e.is_no_records_found() {
-                            // "implicit MX" as described by section 5.1 of RFC5321
-                            // https://datatracker.ietf.org/doc/html/rfc5321#section-5.1
-                            log::debug!("No MX record found, using implicit MX: {mx_domain}");
-                            vec![(0, mx_domain)]
-                        } else if e.is_nx_domain() {
-                            return Err(format!("512 Domain {mx_domain} does not exist"));
-                        } else {
-                            return Err(format!("421 DNS resolution failed for {mx_domain}"));
-                        }
-                    }
-                }
-            }
-        };
-
-        let tls_config = match skip_tls {
-            true => None,
-            false => Some(TlsConfig {
-                allow_invalid_cert,
-                session_cache: tls_resumption_store,
-            }),
-        };
-
-        let mut last_error = None;
-
-        // we try sequentially in order of MX preference,
-        // but the IPv4 and IPv6 connections (after `smtp_client::send` resolves mx hostname)
-        // happens in parallel.
-        'try_relay: for (_, mx_host) in mx_hosts {
-            let skip_mxdeliv = mxdeliv_unsupported_hosts
-                .get(&mx_host)
-                .await
-                .map(|guard| *guard.value())
-                .unwrap_or(false);
-
-            // HTTPS channel
-            if skip_mxdeliv {
-                log::debug!("Skipping HTTP delivery to host that failed recently: {mx_host}");
-            } else {
-                match Self::https_delivery(
-                    https_client.clone(),
-                    mx_host.clone(),
-                    &envelope,
-                    allow_invalid_cert,
-                )
-                .await
-                {
-                    Ok(_) => {
-                        return Ok("250 Ok (HTTPS)".to_string());
-                    }
-                    Err(e) => {
-                        log::debug!("HTTPS delivery to {mx_host} failed: {e}");
-                    }
-                }
-            }
-
-            // SMTP channel (fallback)
-            match crate::smtp_client::send(
-                &mx_host,
-                25,
-                &envelope,
-                &client_hostname,
-                tls_config.clone(),
-                dns_resolver.clone(),
-                smtp_connection_pool.clone(),
-            )
-            .await
-            {
-                Ok(_) => {
-                    // Switches this host to SMTP for 30 minutes.
-                    // Note: this MUST happen only after a successful SMTP delivery,
-                    // or otherwise any http error will lock us out of any way to
-                    // deliver to a relay with a blocked port 25 for 30 minutes.
-                    mxdeliv_unsupported_hosts
-                        .insert(mx_host.clone(), true, Duration::from_mins(30))
-                        .await;
-                    return Ok("250 Ok (SMTP)".to_string());
-                }
-                Err(error) => {
-                    match &error {
-                        // We only want to try other MX hosts if we encounter a problem
-                        // related to connection.
-                        // (So we don't spam other servers if the message is actually rejected.)
-                        crate::error::Error::Io(_)
-                        | crate::error::Error::ConnectionFailed(_)
-                        | crate::error::Error::Tls(_) => {
-                            // Make sure we quickly retry HTTP if SMTP failed to connect
-                            mxdeliv_unsupported_hosts.remove(&mx_host).await;
-                            log::warn!(
-                                "Connection error relaying to mail server {mx_host}: {error}"
-                            );
-                            last_error = Some((error.smtp_response(), mx_host.clone()));
-                            continue 'try_relay;
-                        }
-                        crate::error::Error::MailSend { .. } => {
-                            log::warn!("Message rejected by mail server {mx_host}: {error}");
-                            return Err(error.smtp_response());
-                        }
-                        _ => {
-                            log::warn!(
-                                "Unexpected error while delivering to mail server {mx_host}: {error}"
-                            );
-                            return Err(format!(
-                                "{} (while attempting delivery to {mx_host})",
-                                error.smtp_response()
-                            ));
-                        }
-                    }
-                }
-            }
-        }
-
-        let (error, mx_host) = last_error.unwrap_or(("?".to_string(), "?".to_string()));
-        Err(format!(
-            "421 Failed to connect to any mail server; last attempt to {mx_host}: {error}"
-        ))
-    }
-
-    /// Performs mail delivery to `mx_host` over HTTPS.
     ///
-    /// Times out after 60s.
+    /// Only used for tests.
-    async fn https_delivery(
+    #[cfg(test)]
-        https_client: HttpsClient,
+    pub fn with_queue_size(config: Config, queue_size: usize) -> Result<Self, crate::error::Error> {
-        mx_host: String,
+        let workers = WorkerPool::with_queue_size(config, queue_size)?;
-        envelope: &Envelope,
-        allow_invalid_cert: bool,
-    ) -> Result<(), crate::error::Error> {
-        let request: hyper::Request<http_body_util::Full<Bytes>> = {
-            let mut builder = hyper::Request::builder()
-                .method(hyper::Method::POST)
-                .uri(format!("https://{mx_host}/mxdeliv"));
 
-            if !envelope.mail_from.is_empty() {
+        Ok(Self { workers })
-                builder = builder.header(HEADER_MAIL_FROM, &envelope.mail_from);
-            }
-
-            for rcpt_to in &envelope.rcpt_to {
-                builder = builder.header(HEADER_RCPT_TO, rcpt_to);
-            }
-
-            builder.body(http_body_util::Full::from(envelope.data.clone()))?
-        };
-
-        let client = if allow_invalid_cert {
-            https_client.relaxed
-        } else {
-            https_client.secure
-        };
-
-        let response = tokio::time::timeout(Duration::from_secs(60), client.request(request))
-            .await
-            .map_err(|_| crate::error::Error::MailSend {
-                context: "HTTPS delivery".to_string(),
-                raw_smtp_answer: "[timeout]".to_string(),
-                host: mx_host.clone(),
-            })??;
-        if response.status().is_success() {
-            Ok(())
-        } else {
-            let response_body = response.collect().await?.to_bytes();
-            Err(crate::error::Error::MailSend {
-                context: "HTTPS delivery".to_string(),
-                raw_smtp_answer: String::from_utf8_lossy(&response_body).into(),
-                host: mx_host,
-            })
-        }
     }
 }
 
-impl Drop for TransportHandler {
+#[derive(Debug, Default)]
-    fn drop(&mut self) {
+pub struct TransactionState {
-        self.monitor_handle.abort();
+    permits: BTreeMap<AddressDomain, OwnedPermit<WorkerMessage>>,
-    }
 }
 
 #[async_trait]
 impl SmtpHandler for TransportHandler {
-    /// NO-OP
+    type State = TransactionState;
-    fn handle_mail(&self, _: &str) -> Result<(), String> {
+
+    fn handle_rcpt_to(
+        &self,
+        address: &str,
+        transaction: &mut Transaction<Self::State>,
+    ) -> Result<(), String> {
+        let domain = AddressDomain::from_str(address).map_err(|e| e.smtp_response())?;
+
+        if transaction.state.permits.contains_key(&domain) {
+            // We already acquired a permit for this domain
+            return Ok(());
+        }
+
+        log::trace!(
+            "Trying to acquire a permit for {} worker...",
+            domain.as_ref()
+        );
+        if let Some(permit) = self.workers.get_permit(&domain) {
+            transaction.state.permits.insert(domain, permit);
+        }
+
         Ok(())
     }
 
-    /// NO-OP
+    fn handle_data_start(&self, transaction: &Transaction<Self::State>) -> Result<(), String> {
-    async fn check_data(&self, _: &mut Envelope) -> Result<(), String> {
+        // We want to prevent needlessly sending data from postfix to filtermail,
-        Ok(())
+        // so we fail here if we didn't get any permit.
-    }
+        //
+        // Examplary scenario:
+        // Consider destinations A and B, where A is unavailable.
+        // We are sending a message to a group of 1@A, 2@A, 1@B, 2@B.
+        // After handle_rcpt_to on every recipient, we end up with a permit for domain B (A fails).
+        // handle_data_start passes and mail data is transmitted to filtermail.
+        // Delivery to B is performed; 1@B and 2@B receive message and a message to 1@A and 2@A
+        // is deferred.
+        // After some time the message is retried, now we only try to acquire permit for A,
+        // but fail -> empty `transaction.state.permits`
+        // handle_data_start fails and mail data is not sent to filtermail.
+        // This greatly reduces RAM usage, as unavailable destination can cause large numbers of
+        // deferred mails to be constantly retried.
+
+        if transaction.state.permits.is_empty() {
+            return Err(WORKER_BUSY_421.to_string());
+        }
 
-    /// NO-OP
-    async fn reinject_mail(&self, _: &Envelope) -> Result<(), String> {
         Ok(())
     }
 
     /// Handles the DATA command and returns LMTP responses as single string.
     ///
     /// Never returns an error, as LMTP response is composite.
-    async fn handle_data(&self, envelope: &mut Envelope) -> Result<String, String> {
+    async fn handle_data_dot(
+        &self,
+        transaction: &mut Transaction<Self::State>,
+    ) -> Result<String, String> {
         let mut domain_rcpts_map = BTreeMap::new();
 
-        for rcpt in &envelope.rcpt_to {
+        for rcpt in &transaction.envelope.rcpt_to {
             let domain = AddressDomain::from_str(rcpt)
                 // Currently we cancel all transactions if any recipient address is invalid.
-                .map_err(|e| e.lmtp_response(envelope.rcpt_to.len()))?;
+                .map_err(|e| e.lmtp_response(transaction.envelope.rcpt_to.len()))?;
             domain_rcpts_map
                 .entry(domain)
                 .or_insert_with(Vec::new)
@@ -382,40 +119,50 @@ impl SmtpHandler for TransportHandler {
 
         for (rcpt_domain, rcpts) in &domain_rcpts_map {
             let domain_envelope = {
-                let mut envelope = envelope.clone();
+                let mut envelope = transaction.envelope.clone();
                 envelope.rcpt_to = rcpts.clone();
                 envelope
             };
-            let task_id = transactions
+            let receiver_task_id =
-                .spawn(Self::handle_single_domain(
+                if let Some(permit) = transaction.state.permits.remove(rcpt_domain) {
-                    self.tls_resumption_store.clone(),
+                    let (message, receiver) = WorkerMessage::new(domain_envelope);
-                    self.smtp_connection_pool.clone(),
+                    permit.send(message);
-                    self.mxdeliv_unsupported_hosts.clone(),
+                    // todo: receiver timeout?
-                    self.https_client.clone(),
+                    transactions.spawn(receiver).id()
-                    self.dns_resolver.clone(),
+                } else {
-                    rcpt_domain.clone(),
+                    transactions
-                    domain_envelope,
+                        .spawn(async move { Ok(Err(WORKER_BUSY_421.to_string())) })
-                    self.config.mail_domain.clone(),
+                        .id()
-                ))
+                };
-                .id();
+            task_id_domain_map.insert(receiver_task_id, rcpt_domain);
-            task_id_domain_map.insert(task_id, rcpt_domain);
         }
 
         let mut rcpt_response_map = BTreeMap::new();
         while let Some(result) = transactions.join_next_with_id().await {
-            let domain_response = match result {
+            let domain = match &result {
-                Ok((id, Ok(resp))) | Ok((id, Err(resp))) => {
+                Ok((id, _)) => task_id_domain_map.remove(id),
-                    task_id_domain_map.remove(&id).map(|domain| (domain, resp))
+                Err(e) => task_id_domain_map.remove(&e.id()),
+            };
+
+            let smtp_response = match result {
+                Ok((_, Ok(Ok(resp)))) | Ok((_, Ok(Err(resp)))) => resp,
+                Ok((_, Err(e))) => {
+                    log::error!(
+                        "Worker task failed while delivering to {}: {e}",
+                        domain.map(AsRef::as_ref).unwrap_or("<unknown>")
+                    );
+                    LOCAL_ERROR_451.to_string()
                 }
                 Err(e) => {
-                    log::error!("Failed to join task: {e}");
+                    log::error!(
-                    task_id_domain_map
+                        "Failed to join task while delivering to {}: {e}",
-                        .remove(&e.id())
+                        domain.map(AsRef::as_ref).unwrap_or("<unknown>")
-                        .map(|domain| (domain, "451 Local error".to_string()))
+                    );
+                    LOCAL_ERROR_451.to_string()
                 }
             };
 
-            if let Some((domain, smtp_response)) = domain_response
+            if let Some(domain) = domain
                 && let Some(rcpts) = domain_rcpts_map.get(domain)
             {
                 for rcpt in rcpts {
@@ -425,16 +172,95 @@ impl SmtpHandler for TransportHandler {
         }
 
         // compose lmtp response...
-        let ordered_responses: Vec<String> = envelope
+        let ordered_responses: Vec<String> = transaction
+            .envelope
             .rcpt_to
             .iter()
             .map(|rcpt| {
                 rcpt_response_map
                     .remove(rcpt)
-                    .unwrap_or_else(|| "451 Local error".to_string())
+                    .unwrap_or_else(|| LOCAL_ERROR_451.to_string())
             })
             .collect();
 
         Ok(ordered_responses.join("\r\n"))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rstest::{fixture, rstest};
+    use testresult::TestResult;
+
+    #[fixture]
+    fn addrs1() -> Vec<String> {
+        let mut vec = Vec::new();
+        for idx in 0..5 {
+            vec.push(format!("{idx}@one.example.org"))
+        }
+        vec
+    }
+
+    #[fixture]
+    fn addrs2() -> Vec<String> {
+        let mut vec = Vec::new();
+        for idx in 0..5 {
+            vec.push(format!("{idx}@two.example.org"))
+        }
+        vec
+    }
+
+    #[rstest]
+    #[tokio::test]
+    async fn test_rcpt_to_and_start_data(addrs1: Vec<String>, addrs2: Vec<String>) -> TestResult {
+        let transport_handler = TransportHandler::with_queue_size(Config::default(), 1)?;
+        let domain1 = AddressDomain::from_str(addrs1.first().unwrap())?;
+        let domain2 = AddressDomain::from_str(addrs2.first().unwrap())?;
+
+        {
+            let mut trans_1 = Transaction::default();
+            let mut trans_2 = Transaction::default();
+            let mut trans_3 = Transaction::default();
+
+            transport_handler.handle_rcpt_to(addrs1.first().unwrap(), &mut trans_1)?;
+            assert!(trans_1.state.permits.contains_key(&domain1));
+
+            // Within one transaction, we only use one worker permit, so queue_size=1 is enough.
+            transport_handler.handle_rcpt_to(addrs1.get(1).unwrap(), &mut trans_1)?;
+            assert!(trans_1.state.permits.contains_key(&domain1));
+
+            // However, a second transaction with the same domain won't get a permit.
+            transport_handler.handle_rcpt_to(addrs1.get(2).unwrap(), &mut trans_2)?;
+            assert!(!trans_2.state.permits.contains_key(&domain1));
+
+            // Different domain will work though, as it uses a separate worker, with its own queue.
+            transport_handler.handle_rcpt_to(addrs2.first().unwrap(), &mut trans_2)?;
+            assert!(trans_2.state.permits.contains_key(&domain2));
+
+            // Third transaction won't get any permits.
+            transport_handler.handle_rcpt_to(addrs1.get(3).unwrap(), &mut trans_3)?;
+            transport_handler.handle_rcpt_to(addrs2.get(2).unwrap(), &mut trans_3)?;
+            assert!(!trans_3.state.permits.contains_key(&domain1));
+            assert!(!trans_3.state.permits.contains_key(&domain2));
+
+            // all permits granted -> accept DATA command
+            assert_eq!(transport_handler.handle_data_start(&trans_1), Ok(()));
+
+            // some permits granted -> accept DATA command
+            assert_eq!(transport_handler.handle_data_start(&trans_2), Ok(()));
+
+            // no permits granted -> reject
+            assert!(transport_handler.handle_data_start(&trans_3).is_err());
+        }
+
+        // Transactions (and owned by them permits) going out of scope frees the queues.
+        let mut trans_4 = Transaction::default();
+        transport_handler.handle_rcpt_to(addrs1.first().unwrap(), &mut trans_4)?;
+        transport_handler.handle_rcpt_to(addrs2.first().unwrap(), &mut trans_4)?;
+        assert!(trans_4.state.permits.contains_key(&domain1));
+        assert!(trans_4.state.permits.contains_key(&domain2));
+
+        Ok(())
+    }
+}

filtermail/src/transport/https_client.rs

@@ -0,0 +1,57 @@
+use crate::tls;
+use hyper::body::Bytes;
+use hyper_rustls::HttpsConnector;
+use hyper_util::client::legacy::connect::HttpConnector;
+use std::sync::Arc;
+use tokio_rustls::rustls;
+
+/// Cheaply clonable HTTPS client.
+///
+/// Holds regular secure variant and relaxed - without certificate verification.
+///
+/// Connection pool handled internally by [`hyper_util::client::legacy::Client`].
+#[derive(Clone)]
+pub(crate) struct HttpsClient {
+    pub secure: hyper_util::client::legacy::Client<
+        HttpsConnector<HttpConnector>,
+        http_body_util::Full<Bytes>,
+    >,
+    pub relaxed: hyper_util::client::legacy::Client<
+        HttpsConnector<HttpConnector>,
+        http_body_util::Full<Bytes>,
+    >,
+}
+
+impl HttpsClient {
+    /// Creates a new `[HttpsClient]`.
+    pub fn new(
+        tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
+    ) -> Result<Self, crate::error::Error> {
+        let tls_client_config = tls::configure_rustls(tls_resumption_store.clone(), false)?;
+        let https_connector = hyper_rustls::HttpsConnectorBuilder::new()
+            .with_tls_config(tls_client_config)
+            .https_only()
+            .enable_http1()
+            .enable_http2()
+            .build();
+        let https_client =
+            hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
+                .build(https_connector);
+
+        let tls_client_config_relaxed = tls::configure_rustls(tls_resumption_store, true)?;
+        let https_connector_relaxed = hyper_rustls::HttpsConnectorBuilder::new()
+            .with_tls_config(tls_client_config_relaxed)
+            .https_only()
+            .enable_http1()
+            .enable_http2()
+            .build();
+        let https_client_relaxed =
+            hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
+                .build(https_connector_relaxed);
+
+        Ok(Self {
+            secure: https_client,
+            relaxed: https_client_relaxed,
+        })
+    }
+}

filtermail/src/transport/worker.rs

@@ -0,0 +1,447 @@
+use crate::config::Config;
+use crate::smtp_client::{SmtpConnectionPool, TlsConfig};
+use crate::smtp_responses::{OK_HTTPS_250, OK_SMTP_250};
+use crate::smtp_server::Envelope;
+use crate::transport::{HEADER_MAIL_FROM, HEADER_RCPT_TO, https_client::HttpsClient};
+use crate::utils::{AddressDomain, build_resolver};
+use hickory_resolver::TokioResolver;
+use hickory_resolver::proto::rr::RData;
+use http_body_util::BodyExt;
+use hyper::body::Bytes;
+use parking_lot::RwLock;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc::OwnedPermit;
+use tokio::sync::{mpsc, oneshot};
+use tokio::task;
+use tokio::task::JoinHandle;
+use tokio_rustls::rustls;
+
+/// Message queue size per [`Worker`].
+///
+/// If a queue to a single destination reaches this limit,
+/// all new messages will be immediately deferred.
+const PER_DESTINATION_QUEUE_SIZE: usize = 30;
+
+type SMTPResponse = Result<String, String>;
+
+pub struct WorkerPool {
+    inner: RwLock<BTreeMap<AddressDomain, Arc<Worker>>>,
+    client_hostname: String,
+    smtp_connection_pool: Arc<SmtpConnectionPool>,
+    mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
+    monitor_handle: JoinHandle<()>,
+    dns_resolver: Arc<TokioResolver>,
+    queue_size: usize,
+}
+
+impl WorkerPool {
+    pub fn new(config: Config) -> Result<Self, crate::error::Error> {
+        let dns_resolver = Arc::new(build_resolver()?);
+
+        let mxdeliv_cache = Arc::new(retainer::Cache::new());
+        let mxdeliv_cache_clone = mxdeliv_cache.clone();
+
+        let monitor_handle = tokio::spawn(async move {
+            mxdeliv_cache_clone
+                .monitor(4, 0.25, Duration::from_secs(10))
+                .await
+        });
+
+        Ok(Self {
+            inner: Default::default(),
+            client_hostname: config.mail_domain,
+            dns_resolver,
+            smtp_connection_pool: SmtpConnectionPool::new(),
+            mxdeliv_unsupported_hosts: mxdeliv_cache,
+            monitor_handle,
+            queue_size: PER_DESTINATION_QUEUE_SIZE,
+        })
+    }
+
+    /// Same as [`Self::new`], but lets you set the size of the queue.
+    ///
+    /// Used only for tests.
+    #[cfg(test)]
+    pub fn with_queue_size(config: Config, queue_size: usize) -> Result<Self, crate::error::Error> {
+        let mut this = Self::new(config)?;
+        this.queue_size = queue_size;
+        Ok(this)
+    }
+
+    fn get_or_create_worker(&self, destination: &AddressDomain) -> Arc<Worker> {
+        // NOTE: these locks are blocking, but critical section here is quite small and
+        // shouldn't cause issues in async code.
+        // NOTE: read() returns a guard that is dropped before the match statement.
+        // This must be ensured or else, the write() line would cause a deadlock.
+        let worker = {
+            let mut worker = {
+                let map = self.inner.read();
+                map.get(destination).cloned()
+            };
+            // Remove (and re-create) worker if it finished/crashed.
+            // In reality, this should never happen.
+            if let Some(w) = &worker
+                && w.handle.is_finished()
+            {
+                log::error!(
+                    "Worker for destination {} crashed! Restarting...",
+                    destination.as_ref()
+                );
+                worker = None;
+                {
+                    let mut map = self.inner.write();
+                    map.remove(destination);
+                }
+            };
+            worker
+        };
+
+        match worker {
+            Some(worker) => worker,
+            None => {
+                // Worker for this destination wasn't spawned yet.
+                let (tx, rx) = mpsc::channel(self.queue_size);
+                let handle = tokio::spawn(Worker::run(
+                    destination.clone(),
+                    rx,
+                    self.client_hostname.clone(),
+                    self.smtp_connection_pool.clone(),
+                    self.mxdeliv_unsupported_hosts.clone(),
+                    self.dns_resolver.clone(),
+                ));
+                log::trace!("Worker {} spawned", handle.id());
+                let worker = Arc::new(Worker { tx, handle });
+
+                self.inner
+                    .write()
+                    .insert(destination.clone(), worker.clone());
+                worker
+            }
+        }
+    }
+
+    /// Tries to get an [`OwnedPermit`] to the worker for specified destination.
+    ///
+    /// Returns [`None`] if the worker's queue is full.
+    pub fn get_permit(&self, destination: &AddressDomain) -> Option<OwnedPermit<WorkerMessage>> {
+        let worker = self.get_or_create_worker(destination);
+        worker.tx.clone().try_reserve_owned().ok()
+    }
+}
+
+impl Drop for WorkerPool {
+    fn drop(&mut self) {
+        self.monitor_handle.abort();
+    }
+}
+
+#[derive(Debug)]
+pub struct Worker {
+    pub tx: mpsc::Sender<WorkerMessage>,
+    handle: JoinHandle<Result<(), crate::error::Error>>,
+}
+
+impl Drop for Worker {
+    fn drop(&mut self) {
+        self.handle.abort();
+    }
+}
+
+impl Worker {
+    pub async fn run(
+        destination: AddressDomain,
+        mut rx: mpsc::Receiver<WorkerMessage>,
+        client_hostname: String,
+        smtp_connection_pool: Arc<SmtpConnectionPool>,
+        mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
+        dns_resolver: Arc<TokioResolver>,
+    ) -> Result<(), crate::error::Error> {
+        let worker_id = task::try_id()
+            .map(|id| id.to_string())
+            .unwrap_or("?".to_string());
+
+        log::info!(
+            "Starting worker {worker_id} for destination {}",
+            destination.as_ref()
+        );
+
+        let tls_resumption_store = Arc::new(rustls::client::ClientSessionMemoryCache::new(256));
+        let https_client = HttpsClient::new(tls_resumption_store.clone())?;
+
+        while let Some(message) = rx.recv().await {
+            log::trace!(
+                "Worker {worker_id} received a message from {}",
+                message.envelope.mail_from
+            );
+            let result = Self::handle_single_domain(
+                tls_resumption_store.clone(),
+                smtp_connection_pool.clone(),
+                mxdeliv_unsupported_hosts.clone(),
+                https_client.clone(),
+                dns_resolver.clone(),
+                destination.clone(),
+                message.envelope,
+                client_hostname.clone(),
+            )
+            .await;
+            if message.response_tx.send(result).is_err() {
+                log::error!(
+                    "Worker {worker_id} ({}) failed to send response to transport handler.",
+                    destination.as_ref()
+                );
+            };
+        }
+
+        Ok(())
+    }
+
+    /// Handles a single email transaction for a single recipient domain.
+    #[expect(clippy::too_many_arguments)]
+    async fn handle_single_domain(
+        tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
+        smtp_connection_pool: Arc<SmtpConnectionPool>,
+        mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
+        https_client: HttpsClient,
+        dns_resolver: Arc<TokioResolver>,
+        domain: AddressDomain,
+        envelope: Envelope,
+        client_hostname: String,
+    ) -> Result<String, String> {
+        let mut allow_invalid_cert = false;
+        let mut skip_tls = false; // only respected by smtp channel
+
+        let mx_hosts = match domain {
+            // no-DNS setup; assume the ip from email address is the destination.
+            AddressDomain::Literal(ip) => {
+                // We allow self-signed certs on IP-based relays.
+                allow_invalid_cert = true;
+                vec![(0, ip)]
+            }
+            AddressDomain::Name(mx_domain) => {
+                if mx_domain.eq_ignore_ascii_case("nauta.cu") {
+                    // Special case; We don't want to defederate nauta.cu,
+                    // which doesn't support STARTTLS at all.
+                    skip_tls = true;
+                } else if mx_domain.starts_with('_') {
+                    // We use domains starting with `_` for test deployments.
+                    // (You can't request a non-wildcard cert for such domain)
+                    allow_invalid_cert = true;
+                }
+                let query = format!("{mx_domain}.");
+
+                match dns_resolver.mx_lookup(query).await {
+                    Ok(mx_records) => {
+                        let mut hosts: Vec<(u16, String)> = Vec::new();
+                        for mx_record in mx_records.answers() {
+                            let mx = match mx_record.data {
+                                RData::MX(ref mx) => mx,
+                                _ => continue,
+                            };
+
+                            // Null MX / RFC7505
+                            if mx.exchange.is_root() {
+                                // From RFC7505 section 3:
+                                // > A domain that advertises a null MX MUST NOT
+                                // > advertise any other MX RR.
+                                // We assume this is the only record and exit early.
+                                return Err(
+                                    "556 5.1.10 Permanent failure: Recipient address has null MX"
+                                        .to_string(),
+                                );
+                            }
+
+                            let host = mx.exchange.to_string().trim_end_matches('.').to_string();
+                            hosts.push((mx.preference, host))
+                        }
+                        hosts.sort();
+                        hosts
+                    }
+                    Err(e) => {
+                        if e.is_no_records_found() {
+                            // "implicit MX" as described by section 5.1 of RFC5321
+                            // https://datatracker.ietf.org/doc/html/rfc5321#section-5.1
+                            log::debug!("No MX record found, using implicit MX: {mx_domain}");
+                            vec![(0, mx_domain)]
+                        } else if e.is_nx_domain() {
+                            return Err(format!("512 Domain {mx_domain} does not exist"));
+                        } else {
+                            return Err(format!("421 DNS resolution failed for {mx_domain}"));
+                        }
+                    }
+                }
+            }
+        };
+
+        let tls_config = match skip_tls {
+            true => None,
+            false => Some(TlsConfig {
+                allow_invalid_cert,
+                session_cache: tls_resumption_store,
+            }),
+        };
+
+        let mut last_error = None;
+
+        // we try sequentially in order of MX preference,
+        // but the IPv4 and IPv6 connections (after `smtp_client::send` resolves mx hostname)
+        // happens in parallel.
+        'try_relay: for (_, mx_host) in mx_hosts {
+            let skip_mxdeliv = mxdeliv_unsupported_hosts
+                .get(&mx_host)
+                .await
+                .map(|guard| *guard.value())
+                .is_some();
+
+            // HTTPS channel
+            if skip_mxdeliv {
+                log::debug!("Skipping HTTP delivery to host that failed recently: {mx_host}");
+            } else {
+                match Self::https_delivery(
+                    https_client.clone(),
+                    mx_host.clone(),
+                    &envelope,
+                    allow_invalid_cert,
+                )
+                .await
+                {
+                    Ok(_) => {
+                        return Ok(OK_HTTPS_250.to_string());
+                    }
+                    Err(e) => {
+                        log::debug!("HTTPS delivery to {mx_host} failed: {e}");
+                    }
+                }
+            }
+
+            // SMTP channel (fallback)
+            match crate::smtp_client::send(
+                &mx_host,
+                25,
+                &envelope,
+                &client_hostname,
+                tls_config.clone(),
+                dns_resolver.clone(),
+                smtp_connection_pool.clone(),
+            )
+            .await
+            {
+                Ok(_) => {
+                    // Switches this host to SMTP for 30 minutes.
+                    // Note: this MUST happen only after a successful SMTP delivery,
+                    // or otherwise any http error will lock us out of any way to
+                    // deliver to a relay with a blocked port 25 for 30 minutes.
+                    mxdeliv_unsupported_hosts
+                        .insert(mx_host.clone(), (), Duration::from_mins(30))
+                        .await;
+                    return Ok(OK_SMTP_250.to_string());
+                }
+                Err(error) => {
+                    match &error {
+                        // We only want to try other MX hosts if we encounter a problem
+                        // related to connection.
+                        // (So we don't spam other servers if the message is actually rejected.)
+                        crate::error::Error::Io(_)
+                        | crate::error::Error::ConnectionFailed(_)
+                        | crate::error::Error::Tls(_) => {
+                            // Make sure we quickly retry HTTP if SMTP failed to connect
+                            mxdeliv_unsupported_hosts.remove(&mx_host).await;
+                            log::warn!(
+                                "Connection error relaying to mail server {mx_host}: {error}"
+                            );
+                            last_error = Some((error.smtp_response(), mx_host.clone()));
+                            continue 'try_relay;
+                        }
+                        crate::error::Error::MailSend { .. } => {
+                            log::warn!("Message rejected by mail server {mx_host}: {error}");
+                            return Err(error.smtp_response());
+                        }
+                        _ => {
+                            log::warn!(
+                                "Unexpected error while delivering to mail server {mx_host}: {error}"
+                            );
+                            return Err(format!(
+                                "{} (while attempting delivery to {mx_host})",
+                                error.smtp_response()
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+
+        let (error, mx_host) = last_error.unwrap_or(("?".to_string(), "?".to_string()));
+        Err(format!(
+            "421 Failed to connect to any mail server; last attempt to {mx_host}: {error}"
+        ))
+    }
+
+    /// Performs mail delivery to `mx_host` over HTTPS.
+    ///
+    /// Times out after 60s.
+    async fn https_delivery(
+        https_client: HttpsClient,
+        mx_host: String,
+        envelope: &Envelope,
+        allow_invalid_cert: bool,
+    ) -> Result<(), crate::error::Error> {
+        let request: hyper::Request<http_body_util::Full<Bytes>> = {
+            let mut builder = hyper::Request::builder()
+                .method(hyper::Method::POST)
+                .uri(format!("https://{mx_host}/mxdeliv"));
+
+            if !envelope.mail_from.is_empty() {
+                builder = builder.header(HEADER_MAIL_FROM, &envelope.mail_from);
+            }
+
+            for rcpt_to in &envelope.rcpt_to {
+                builder = builder.header(HEADER_RCPT_TO, rcpt_to);
+            }
+
+            builder.body(http_body_util::Full::from(envelope.data.clone()))?
+        };
+
+        let client = if allow_invalid_cert {
+            https_client.relaxed
+        } else {
+            https_client.secure
+        };
+
+        let response = tokio::time::timeout(Duration::from_secs(60), client.request(request))
+            .await
+            .map_err(|_| crate::error::Error::MailSend {
+                context: "HTTPS delivery".to_string(),
+                raw_smtp_answer: "[timeout]".to_string(),
+                host: mx_host.clone(),
+            })??;
+        if response.status().is_success() {
+            Ok(())
+        } else {
+            let response_body = response.collect().await?.to_bytes();
+            Err(crate::error::Error::MailSend {
+                context: "HTTPS delivery".to_string(),
+                raw_smtp_answer: String::from_utf8_lossy(&response_body).into(),
+                host: mx_host,
+            })
+        }
+    }
+}
+
+pub struct WorkerMessage {
+    pub envelope: Envelope,
+    pub response_tx: oneshot::Sender<SMTPResponse>,
+}
+
+impl WorkerMessage {
+    pub fn new(envelope: Envelope) -> (Self, oneshot::Receiver<SMTPResponse>) {
+        let (response_tx, response_rx) = oneshot::channel();
+        (
+            Self {
+                envelope,
+                response_tx,
+            },
+            response_rx,
+        )
+    }
+}