Merge pull request #9961 from Turbo87/async-emails

Use async `lettre` transports for sending emails

Author: Turbo87

Cargo.lock

@@ -84,7 +84,7 @@ indicatif = "=0.17.9"
 ipnetwork = "=0.20.0"
 json-subscriber = "=0.2.2"
 tikv-jemallocator = { version = "=0.6.0", features = ['unprefixed_malloc_on_supported_platforms', 'profiling'] }
-lettre = { version = "=0.11.10", default-features = false, features = ["file-transport", "smtp-transport", "native-tls", "hostname", "builder"] }
+lettre = { version = "=0.11.10", default-features = false, features = ["file-transport", "smtp-transport", "hostname", "builder", "tokio1", "tokio1-native-tls"] }
 minijinja = "=2.5.0"
 mockall = "=0.13.0"
 native-tls = "=0.2.12"

Cargo.toml

@@ -3,7 +3,6 @@ use crate::app::AppState;
 use crate::auth::AuthCheck;
 use crate::controllers::helpers::ok_true;
 use crate::models::Email;
-use crate::tasks::spawn_blocking;
 use crate::util::errors::AppResult;
 use crate::util::errors::{bad_request, BoxedAppError};
 use axum::extract::Path;
@@ -38,19 +37,17 @@ pub async fn regenerate_token_and_send(
                 .optional()?
                 .ok_or_else(|| bad_request("Email could not be found"))?;
 
-            spawn_blocking(move || {
-                let email1 = UserConfirmEmail {
-                    user_name: &auth.user().gh_login,
-                    domain: &state.emails.domain,
-                    token: email.token,
-                };
-
-                state
-                    .emails
-                    .send(&email.email, email1)
-                    .map_err(BoxedAppError::from)
-            })
-            .await
+            let email1 = UserConfirmEmail {
+                user_name: &auth.user().gh_login,
+                domain: &state.emails.domain,
+                token: email.token,
+            };
+
+            state
+                .emails
+                .async_send(&email.email, email1)
+                .await
+                .map_err(BoxedAppError::from)
         }
         .scope_boxed()
     })
@@ -74,7 +71,7 @@ mod tests {
         assert_eq!(response.status(), StatusCode::FORBIDDEN);
         assert_snapshot!(response.text(), @r#"{"errors":[{"detail":"this action requires authentication"}]}"#);
 
-        assert_eq!(app.emails().len(), 0);
+        assert_eq!(app.emails().await.len(), 0);
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -87,7 +84,7 @@ mod tests {
         assert_eq!(response.status(), StatusCode::BAD_REQUEST);
         assert_snapshot!(response.text(), @r#"{"errors":[{"detail":"current user does not match requested user"}]}"#);
 
-        assert_eq!(app.emails().len(), 0);
+        assert_eq!(app.emails().await.len(), 0);
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -99,6 +96,6 @@ mod tests {
         assert_eq!(response.status(), StatusCode::OK);
         assert_snapshot!(response.text(), @r#"{"ok":true}"#);
 
-        assert_snapshot!(app.emails_snapshot());
+        assert_snapshot!(app.emails_snapshot().await);
     }
 }

src/controllers/user/resend.rs

@@ -3,12 +3,14 @@ use crate::Env;
 use lettre::address::Envelope;
 use lettre::message::header::ContentType;
 use lettre::message::Mailbox;
-use lettre::transport::file::FileTransport;
+use lettre::transport::file::AsyncFileTransport;
 use lettre::transport::smtp::authentication::{Credentials, Mechanism};
-use lettre::transport::smtp::SmtpTransport;
-use lettre::transport::stub::StubTransport;
-use lettre::{Address, Message, Transport};
+use lettre::transport::smtp::AsyncSmtpTransport;
+use lettre::transport::stub::AsyncStubTransport;
+use lettre::{Address, AsyncTransport, Message, Tokio1Executor};
 use rand::distributions::{Alphanumeric, DistString};
+use std::sync::Arc;
+use tokio::runtime::Handle;
 
 pub trait Email {
     fn subject(&self) -> String;
@@ -36,7 +38,7 @@ impl Emails {
 
         let backend = match (login, password, server) {
             (Ok(login), Ok(password), Ok(server)) => {
-                let transport = SmtpTransport::relay(&server)
+                let transport = AsyncSmtpTransport::<Tokio1Executor>::relay(&server)
                     .unwrap()
                     .credentials(Credentials::new(login, password))
                     .authentication(vec![Mechanism::Plain])
@@ -45,8 +47,8 @@ impl Emails {
                 EmailBackend::Smtp(Box::new(transport))
             }
             _ => {
-                let transport = FileTransport::new("/tmp");
-                EmailBackend::FileSystem(transport)
+                let transport = AsyncFileTransport::new("/tmp");
+                EmailBackend::FileSystem(Arc::new(transport))
             }
         };
 
@@ -67,23 +69,23 @@ impl Emails {
     /// to later assert the mails were sent.
     pub fn new_in_memory() -> Self {
         Self {
-            backend: EmailBackend::Memory(StubTransport::new_ok()),
+            backend: EmailBackend::Memory(AsyncStubTransport::new_ok()),
             domain: "crates.io".into(),
             from: DEFAULT_FROM.parse().unwrap(),
         }
     }
 
     /// This is supposed to be used only during tests, to retrieve the messages stored in the
     /// "memory" backend. It's not cfg'd away because our integration tests need to access this.
-    pub fn mails_in_memory(&self) -> Option<Vec<(Envelope, String)>> {
+    pub async fn mails_in_memory(&self) -> Option<Vec<(Envelope, String)>> {
         if let EmailBackend::Memory(transport) = &self.backend {
-            Some(transport.messages())
+            Some(transport.messages().await)
         } else {
             None
         }
     }
 
-    pub fn send<E: Email>(&self, recipient: &str, email: E) -> Result<(), EmailError> {
+    fn build_message<E: Email>(&self, recipient: &str, email: E) -> Result<Message, EmailError> {
         // The message ID is normally generated by the SMTP server, but if we let it generate the
         // ID there will be no way for the crates.io application to know the ID of the message it
         // just sent, as it's not included in the SMTP response.
@@ -102,15 +104,32 @@ impl Emails {
         let subject = email.subject();
         let body = email.body();
 
-        let email = Message::builder()
+        let message = Message::builder()
             .message_id(Some(message_id.clone()))
             .to(recipient.parse()?)
             .from(from)
             .subject(subject)
             .header(ContentType::TEXT_PLAIN)
             .body(body)?;
 
-        self.backend.send(email).map_err(EmailError::TransportError)
+        Ok(message)
+    }
+
+    pub fn send<E: Email>(&self, recipient: &str, email: E) -> Result<(), EmailError> {
+        let email = self.build_message(recipient, email)?;
+
+        Handle::current()
+            .block_on(self.backend.send(email))
+            .map_err(EmailError::TransportError)
+    }
+
+    pub async fn async_send<E: Email>(&self, recipient: &str, email: E) -> Result<(), EmailError> {
+        let email = self.build_message(recipient, email)?;
+
+        self.backend
+            .send(email)
+            .await
+            .map_err(EmailError::TransportError)
     }
 }
 
@@ -129,19 +148,19 @@ enum EmailBackend {
     /// Backend used in production to send mails using SMTP.
     ///
     /// This is using `Box` to avoid a large size difference between variants.
-    Smtp(Box<SmtpTransport>),
+    Smtp(Box<AsyncSmtpTransport<Tokio1Executor>>),
     /// Backend used locally during development, will store the emails in the provided directory.
-    FileSystem(FileTransport),
+    FileSystem(Arc<AsyncFileTransport<Tokio1Executor>>),
     /// Backend used during tests, will keep messages in memory to allow tests to retrieve them.
-    Memory(StubTransport),
+    Memory(AsyncStubTransport),
 }
 
 impl EmailBackend {
-    fn send(&self, message: Message) -> anyhow::Result<()> {
+    async fn send(&self, message: Message) -> anyhow::Result<()> {
         match self {
-            EmailBackend::Smtp(transport) => transport.send(&message).map(|_| ())?,
-            EmailBackend::FileSystem(transport) => transport.send(&message).map(|_| ())?,
-            EmailBackend::Memory(transport) => transport.send(&message).map(|_| ())?,
+            EmailBackend::Smtp(transport) => transport.send(message).await.map(|_| ())?,
+            EmailBackend::FileSystem(transport) => transport.send(message).await.map(|_| ())?,
+            EmailBackend::Memory(transport) => transport.send(message).await.map(|_| ())?,
         }
 
         Ok(())
@@ -171,20 +190,19 @@ mod tests {
         }
     }
 
-    #[test]
-    fn sending_to_invalid_email_fails() {
+    #[tokio::test]
+    async fn sending_to_invalid_email_fails() {
         let emails = Emails::new_in_memory();
 
-        assert_err!(emails.send(
-            "String.Format(\"{0}.{1}@live.com\", FirstName, LastName)",
-            TestEmail
-        ));
+        let address = "String.Format(\"{0}.{1}@live.com\", FirstName, LastName)";
+        assert_err!(emails.async_send(address, TestEmail).await);
     }
 
-    #[test]
-    fn sending_to_valid_email_succeeds() {
+    #[tokio::test]
+    async fn sending_to_valid_email_succeeds() {
         let emails = Emails::new_in_memory();
 
-        assert_ok!(emails.send("[email protected]", TestEmail));
+        let address = "[email protected]";
+        assert_ok!(emails.async_send(address, TestEmail).await);
     }
 }

src/email.rs

@@ -45,7 +45,7 @@ async fn github_secret_alert_revokes_token() {
     let mut conn = app.async_db_conn().await;
 
     // Ensure no emails were sent up to this point
-    assert_eq!(app.emails().len(), 0);
+    assert_eq!(app.emails().await.len(), 0);
 
     // Ensure that the token currently exists in the database
     let tokens: Vec<ApiToken> = assert_ok!(
@@ -94,7 +94,7 @@ async fn github_secret_alert_revokes_token() {
     assert_that!(tokens, len(eq(1)));
 
     // Ensure exactly one email was sent
-    assert_snapshot!(app.emails_snapshot());
+    assert_snapshot!(app.emails_snapshot().await);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -103,7 +103,7 @@ async fn github_secret_alert_for_revoked_token() {
     let mut conn = app.async_db_conn().await;
 
     // Ensure no emails were sent up to this point
-    assert_eq!(app.emails().len(), 0);
+    assert_eq!(app.emails().await.len(), 0);
 
     // Ensure that the token currently exists in the database
     let tokens: Vec<ApiToken> = assert_ok!(
@@ -155,7 +155,7 @@ async fn github_secret_alert_for_revoked_token() {
     assert_that!(tokens, len(eq(1)));
 
     // Ensure still no emails were sent
-    assert_eq!(app.emails().len(), 0);
+    assert_eq!(app.emails().await.len(), 0);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -164,7 +164,7 @@ async fn github_secret_alert_for_unknown_token() {
     let mut conn = app.async_db_conn().await;
 
     // Ensure no emails were sent up to this point
-    assert_eq!(app.emails().len(), 0);
+    assert_eq!(app.emails().await.len(), 0);
 
     // Ensure that the token currently exists in the database
     let tokens: Vec<ApiToken> = assert_ok!(
@@ -197,7 +197,7 @@ async fn github_secret_alert_for_unknown_token() {
     assert_eq!(tokens[0].name, token.as_model().name);
 
     // Ensure still no emails were sent
-    assert_eq!(app.emails().len(), 0);
+    assert_eq!(app.emails().await.len(), 0);
 }
 
 #[tokio::test(flavor = "multi_thread")]

src/tests/github_secret_scanning.rs

@@ -30,7 +30,7 @@ async fn new_wrong_token() {
     assert_eq!(response.status(), StatusCode::FORBIDDEN);
     assert_snapshot!(response.text(), @r#"{"errors":[{"detail":"authentication failed"}]}"#);
     assert_that!(app.stored_files().await, empty());
-    assert_that!(app.emails(), empty());
+    assert_that!(app.emails().await, empty());
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -50,5 +50,5 @@ async fn new_krate_wrong_user() {
     assert_snapshot!(response.text(), @r#"{"errors":[{"detail":"this crate exists but you don't seem to be an owner. If you believe this is a mistake, perhaps you need to accept an invitation to be an owner before publishing."}]}"#);
 
     assert_that!(app.stored_files().await, empty());
-    assert_that!(app.emails(), empty());
+    assert_that!(app.emails().await, empty());
 }

src/tests/krate/publish/auth.rs

@@ -38,7 +38,7 @@ async fn new_krate() {
         .unwrap();
     assert_eq!(email, "[email protected]");
 
-    assert_snapshot!(app.emails_snapshot());
+    assert_snapshot!(app.emails_snapshot().await);
 }
 
 #[tokio::test(flavor = "multi_thread")]

src/tests/krate/publish/basics.rs

@@ -21,7 +21,7 @@ async fn new_krate_without_any_email_fails() {
     assert_eq!(response.status(), StatusCode::BAD_REQUEST);
     assert_snapshot!(response.text(), @r#"{"errors":[{"detail":"A verified email address is required to publish crates to crates.io. Visit https://crates.io/settings/profile to set and verify your email address."}]}"#);
     assert_that!(app.stored_files().await, empty());
-    assert_that!(app.emails(), empty());
+    assert_that!(app.emails().await, empty());
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -41,5 +41,5 @@ async fn new_krate_with_unverified_email_fails() {
     assert_eq!(response.status(), StatusCode::BAD_REQUEST);
     assert_snapshot!(response.text(), @r#"{"errors":[{"detail":"A verified email address is required to publish crates to crates.io. Visit https://crates.io/settings/profile to set and verify your email address."}]}"#);
     assert_that!(app.stored_files().await, empty());
-    assert_that!(app.emails(), empty());
+    assert_that!(app.emails().await, empty());
 }