feat: switch to another SMTP transport when current one returns an error

Classic email servers sometimes start rejecting all messages and ask the user to solve web CAPTCHA, but still allow to connect.
In this case we want to switch to another SMTP server by moving the transport to the end of the list returned by smtp::sorted_transports.

We disconnect after any message sending failure, so next message or resent same message will be sent over another other transport.

Permanent SMTP error does not necessarily mean that switching to another server will help.
For example if the user tries to send a large message, any server will reject it, and it is not a failure of the server.
We do not try to distinguish between error messages, so it is expected that we may
sometimes switch to another server when it is unnecessary.

.github/workflows/ci.yml

@@ -25,8 +25,6 @@ env:
   # Minimum Supported Rust Version
   MSRV: 1.89.0
 
-  PYTHON_VERSION: 3.15
-
 jobs:
   lint_rust:
     name: Lint Rust
@@ -270,11 +268,11 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          # Currently used Python version.
+          # Currently used Rust version.
           - os: ubuntu-latest
-            python: latest
+            python: 3.14
           - os: macos-latest
-            python: latest
+            python: 3.14
 
           # PyPy tests
           - os: ubuntu-latest
@@ -305,7 +303,7 @@ jobs:
       - name: Install python
         uses: actions/setup-python@v7.0.0
         with:
-          python-version: ${{ matrix.python == 'latest' && env.PYTHON_VERSION || matrix.python }}
+          python-version: ${{ matrix.python }}
 
       - name: Install tox
         run: pip install tox
@@ -326,11 +324,11 @@ jobs:
       matrix:
         include:
           - os: ubuntu-latest
-            python: latest
+            python: 3.14
           - os: macos-latest
-            python: latest
+            python: 3.14
           - os: windows-latest
-            python: latest
+            python: 3.14
 
           # PyPy tests
           - os: ubuntu-latest
@@ -353,7 +351,7 @@ jobs:
       - name: Install python
         uses: actions/setup-python@v7.0.0
         with:
-          python-version: ${{ matrix.python == 'latest' && env.PYTHON_VERSION || matrix.python }}
+          python-version: ${{ matrix.python }}
 
       - name: Install tox
         run: pip install tox

docs/schema.sql

@@ -471,6 +471,19 @@ CREATE TABLE smtp_success (
     FOREIGN KEY(transport_id) REFERENCES transports(id) ON DELETE CASCADE
 ) STRICT;
 
+-- Similar to smtp_success, but records failures.
+CREATE TABLE smtp_failure (
+    -- Sequentially increasing ID of the failure
+    -- Transport with the highest ID is to be used last.
+    id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+
+    -- ID of the transport that was used to attempt sending a message.
+    transport_id INTEGER UNIQUE NOT NULL,
+
+    -- Delete `smtp_failure` rows when the transport is deleted.
+    FOREIGN KEY(transport_id) REFERENCES transports(id) ON DELETE CASCADE
+) STRICT;
+
 -- Table of "sync items" to be grouped into sync messages
 -- and sent to own devices.
 CREATE TABLE multi_device_sync (

src/configure.rs

@@ -513,6 +513,8 @@ pub(crate) async fn configure(
         let smtp_password = configured_param.smtp_password.clone();
         let smtp_addr = configured_param.addr.clone();
 
+        let transport_id = 0;
+
         let proxy_config2 = proxy_config.clone();
         let smtp_config_task = task::spawn(async move {
             let mut smtp = Smtp::new();
@@ -522,6 +524,7 @@ pub(crate) async fn configure(
                 &smtp_password,
                 &proxy_config2,
                 &smtp_addr,
+                transport_id,
                 strict_tls,
             )
             .await?;
@@ -533,7 +536,6 @@ pub(crate) async fn configure(
 
         // Configure IMAP
 
-        let transport_id = 0;
         let (_s, r) = async_channel::bounded(1);
         let mut imap = Imap::new(ctx, transport_id, configured_param.clone(), r).await?;
         let imap_session = match imap.connect(ctx).await {

src/smtp.rs

@@ -32,21 +32,26 @@ use crate::transport::{
     ConfiguredLoginParam, ConfiguredServerLoginParam, prioritize_server_login_params,
 };
 
+/// SMTP connection with information about it.
+#[derive(Debug)]
+struct Connection {
+    transport: SmtpTransport<Box<dyn SessionBufStream>>,
+
+    /// Email address that should be put into the From field when sending over this connection.
+    from: EmailAddress,
+
+    /// ID of the transport from which SMTP connection configuration comes from.
+    transport_id: u32,
+
+    /// Timestamp of last successful send/receive network interaction
+    /// (eg connect or send succeeded).
+    last_success: tools::Time,
+}
+
 #[derive(Default)]
 pub(crate) struct Smtp {
     /// SMTP connection.
-    transport: Option<SmtpTransport<Box<dyn SessionBufStream>>>,
-
-    /// Email address we are sending from.
-    from: Option<EmailAddress>,
-
-    /// Transport we are connected to.
-    transport_id: Option<u32>,
-
-    /// Timestamp of last successful send/receive network interaction
-    /// (eg connect or send succeeded). On initialization and disconnect
-    /// it is set to None.
-    last_success: Option<tools::Time>,
+    connection: Option<Connection>,
 
     pub(crate) connectivity: ConnectivityStore,
 
@@ -61,7 +66,10 @@ async fn sorted_transports(context: &Context) -> Result<Vec<(u32, ConfiguredLogi
         .query_map_vec(
             "SELECT transports.id, configured_param FROM transports
              LEFT JOIN smtp_success ON smtp_success.transport_id=transports.id
-             ORDER BY IFNULL(smtp_success.id, 0) DESC, transports.id ASC",
+             LEFT JOIN smtp_failure ON smtp_failure.transport_id=transports.id
+             ORDER BY IFNULL(smtp_success.id, 0) DESC,
+                      IFNULL(smtp_failure.id, 0) ASC,
+                      transports.id ASC",
             (),
             |row| {
                 let id: u32 = row.get(0)?;
@@ -79,10 +87,36 @@ async fn record_success(context: &Context, transport_id: u32) -> Result<()> {
     // if the row exists already, so it becomes the highest rowid in the table.
     context
         .sql
-        .execute(
-            "INSERT OR REPLACE INTO smtp_success (transport_id) VALUES (?)",
-            (transport_id,),
-        )
+        .transaction(|t| {
+            t.execute(
+                "INSERT OR REPLACE INTO smtp_success (transport_id) VALUES (?)",
+                (transport_id,),
+            )?;
+            t.execute(
+                "DELETE FROM smtp_failure WHERE transport_id = ?",
+                (transport_id,),
+            )?;
+            Ok(())
+        })
+        .await?;
+    Ok(())
+}
+
+/// Removes last remembered success for SMTP transport so another transport is tried next time.
+async fn record_failure(context: &Context, transport_id: u32) -> Result<()> {
+    context
+        .sql
+        .transaction(|t| {
+            t.execute(
+                "INSERT OR REPLACE INTO smtp_failure (transport_id) VALUES (?)",
+                (transport_id,),
+            )?;
+            t.execute(
+                "DELETE FROM smtp_success WHERE transport_id = ?",
+                (transport_id,),
+            )?;
+            Ok(())
+        })
         .await?;
     Ok(())
 }
@@ -95,30 +129,25 @@ impl Smtp {
 
     /// Disconnect the SMTP transport and drop it entirely.
     pub fn disconnect(&mut self) {
-        if let Some(mut transport) = self.transport.take() {
+        if let Some(mut connection) = self.connection.take() {
             // Closing connection with a QUIT command may take some time, especially if it's a
             // stale connection and an attempt to send the command times out. Send a command in a
             // separate task to avoid waiting for reply or timeout.
-            task::spawn(async move { transport.quit().await });
+            task::spawn(async move { connection.transport.quit().await });
         }
-        self.transport_id = None;
-        self.from = None;
-        self.last_success = None;
     }
 
     /// Return true if smtp was connected but is not known to
     /// have been successfully used the last 60 seconds
     pub fn has_maybe_stale_connection(&self) -> bool {
-        if let Some(last_success) = self.last_success {
-            time_elapsed(&last_success).as_secs() > 60
-        } else {
-            false
-        }
+        self.connection
+            .as_ref()
+            .is_some_and(|connection| time_elapsed(&connection.last_success).as_secs() > 60)
     }
 
     /// Check whether we are connected.
     pub fn is_connected(&self) -> bool {
-        self.transport.is_some()
+        self.connection.is_some()
     }
 
     /// Connect using configured parameters.
@@ -143,14 +172,12 @@ impl Smtp {
                     &lp.smtp_password,
                     &proxy_config,
                     &lp.addr,
+                    transport_id,
                     lp.strict_tls(proxy_config.is_some())?,
                 )
                 .await
             {
-                Ok(()) => {
-                    self.transport_id = Some(transport_id);
-                    return Ok(());
-                }
+                Ok(()) => return Ok(()),
                 Err(err) => {
                     warn!(
                         context,
@@ -163,6 +190,7 @@ impl Smtp {
     }
 
     /// Connect using the provided login params.
+    #[expect(clippy::too_many_arguments)]
     pub async fn connect(
         &mut self,
         context: &Context,
@@ -170,6 +198,7 @@ impl Smtp {
         password: &str,
         proxy_config: &Option<ProxyConfig>,
         addr: &str,
+        transport_id: u32,
         strict_tls: bool,
     ) -> Result<()> {
         if self.is_connected() {
@@ -179,7 +208,6 @@ impl Smtp {
 
         let from = EmailAddress::new(addr.to_string())
             .with_context(|| format!("Invalid address {addr:?}"))?;
-        self.from = Some(from);
 
         let login_params =
             prioritize_server_login_params(&context.sql, login_params, "smtp").await?;
@@ -204,8 +232,12 @@ impl Smtp {
                 }
             };
 
-            self.transport = Some(transport);
-            self.last_success = Some(tools::Time::now());
+            self.connection = Some(Connection {
+                transport,
+                from,
+                transport_id,
+                last_success: tools::Time::now(),
+            });
 
             context.emit_event(EventType::SmtpConnected(format!(
                 "SMTP-LOGIN as {} ok",
@@ -230,7 +262,7 @@ pub(crate) enum SendResult {
 }
 
 /// Tries to send a message.
-pub(crate) async fn smtp_send(
+async fn smtp_send(
     context: &Context,
     recipients: &[async_smtp::EmailAddress],
     message: &str,
@@ -246,14 +278,37 @@ pub(crate) async fn smtp_send(
 
     smtp.connectivity.set_working(context);
 
-    let send_result = smtp.send(context, recipients, message.as_bytes()).await;
+    let Some(ref mut connection) = smtp.connection else {
+        warn!(
+            context,
+            "Failed to send a message because SMTP client is not connected."
+        );
+        return SendResult::Retry;
+    };
+    let transport_id = connection.transport_id;
+
+    let send_result = connection
+        .send(context, recipients, message.as_bytes())
+        .await;
     smtp.last_send_error = send_result.as_ref().err().map(|e| e.to_string());
 
     let status = match send_result {
-        Err(crate::smtp::send::Error::SmtpSend(err)) => {
+        Err(send::Error::SmtpSend(err)) => {
             // Remote error, retry later.
             info!(context, "SMTP failed to send: {:?}.", &err);
 
+            // Remove recorded SMTP success on any error returned by the server, but not network errors.
+            if matches!(
+                err,
+                async_smtp::error::Error::Permanent(_) | async_smtp::error::Error::Transient(_)
+            ) && let Err(err) = record_failure(context, transport_id).await
+            {
+                warn!(
+                    context,
+                    "Failed to remove success for {transport_id}: {err:#}."
+                )
+            }
+
             let res = match err {
                 async_smtp::error::Error::Permanent(ref response) => {
                     // Workaround for incorrectly configured servers returning permanent errors
@@ -286,6 +341,7 @@ pub(crate) async fn smtp_send(
                         SendResult::Retry
                     } else {
                         info!(context, "Permanent error, message sending failed.");
+
                         // If we do not retry, add an info message to the chat.
                         // Yandex error "554 5.7.1 [2] Message rejected under suspicion of SPAM; https://ya.cc/..."
                         // should definitely go here, because user has to open the link to
@@ -316,6 +372,7 @@ pub(crate) async fn smtp_send(
                         context,
                         "Transient error {response:?}, postponing retry for later."
                     );
+
                     SendResult::Retry
                 }
                 _ => {
@@ -327,44 +384,27 @@ pub(crate) async fn smtp_send(
                 }
             };
 
-            // this clears last_success info
             info!(context, "Failed to send message over SMTP, disconnecting.");
             smtp.disconnect();
 
             res
         }
-        Err(crate::smtp::send::Error::Envelope(err)) => {
-            // Local error, job is invalid, do not retry.
-            smtp.disconnect();
-            warn!(context, "SMTP job is invalid: {err:#}.");
-            SendResult::Failure(err)
-        }
-        Err(crate::smtp::send::Error::NoTransport) => {
-            // Should never happen.
-            // It does not even make sense to disconnect here.
-            error!(context, "SMTP job failed because SMTP has no transport.");
-            SendResult::Failure(format_err!("SMTP has not transport"))
-        }
-        Err(crate::smtp::send::Error::Other(err)) => {
+        Err(send::Error::Other(err)) => {
             // Local error, job is invalid, do not retry.
             smtp.disconnect();
             warn!(context, "Unable to load SMTP job: {err:#}.");
             SendResult::Failure(err)
         }
-        Ok(()) => SendResult::Success,
-    };
-
-    if matches!(status, SendResult::Success) {
-        debug_assert!(smtp.transport_id.is_some());
-        if let Some(transport_id) = smtp.transport_id
-            && let Err(err) = record_success(context, transport_id).await
-        {
-            warn!(
-                context,
-                "Failed to record successful use of transport {transport_id} in smtp_success table: {err:#}."
-            );
+        Ok(()) => {
+            if let Err(err) = record_success(context, transport_id).await {
+                warn!(
+                    context,
+                    "Failed to record successful use of transport {transport_id} in smtp_success table: {err:#}."
+                );
+            }
+            SendResult::Success
         }
-    }
+    };
 
     if let SendResult::Failure(err) = &status
         && let Some(msg_id) = msg_id
@@ -407,11 +447,7 @@ pub(crate) async fn insert_into_smtp(
 /// Sends message identified by `smtp` table rowid over SMTP connection.
 ///
 /// Removes row if the message should not be retried, otherwise increments retry count.
-pub(crate) async fn send_msg_to_smtp(
-    context: &Context,
-    smtp: &mut Smtp,
-    rowid: i64,
-) -> anyhow::Result<()> {
+pub(crate) async fn send_msg_to_smtp(context: &Context, smtp: &mut Smtp, rowid: i64) -> Result<()> {
     if let Err(err) = smtp
         .connect_configured(context)
         .await
@@ -472,9 +508,10 @@ pub(crate) async fn send_msg_to_smtp(
     let mut recipients = queued_mail.recipients.clone();
     if queued_mail.bcc_self {
         let from_addr = smtp
-            .from
+            .connection
             .as_ref()
-            .context("No From address available, likely not connected")?
+            .context("Not connected")?
+            .from
             .to_string();
         add_self_recipients(
             context,
@@ -504,12 +541,17 @@ pub(crate) async fn send_msg_to_smtp(
     let public_key = key::load_self_public_key(context).await?;
     let secret_key = key::load_self_secret_key(context).await?;
     let from_addr = smtp
-        .from
+        .connection
         .as_ref()
-        .context("No From address available, likely not connected")?
+        .context("Not connected")?
+        .from
         .to_string();
 
-    let transport_id = smtp.transport_id.context("Not connected")?;
+    let transport_id = smtp
+        .connection
+        .as_ref()
+        .context("Not connected")?
+        .transport_id;
     let chunk_size = context
         .get_max_smtp_rcpt_to(transport_id, &from_addr)
         .await?
@@ -760,9 +802,10 @@ async fn send_mdn_rfc724_mid(
     let public_key = key::load_self_public_key(context).await?;
     let secret_key = key::load_self_secret_key(context).await?;
     let from = smtp
-        .from
+        .connection
         .as_ref()
-        .context("No From address, not connected")?
+        .context("Not connected")?
+        .from
         .to_string();
     let rendered_msg =
         mimefactory::render_queued_mail(queued_mdn, &public_key, &secret_key, from.clone())?;

src/smtp/send.rs

@@ -1,31 +1,27 @@
 //! # SMTP message sending
 
+use anyhow::Context as _;
 use async_smtp::{EmailAddress, Envelope, SendableEmail};
 
-use super::Smtp;
+use super::Connection;
 use crate::config::Config;
 use crate::context::Context;
-use crate::log::warn;
 use crate::tools;
 
 pub type Result<T> = std::result::Result<T, Error>;
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    #[error("Envelope error: {}", _0)]
-    Envelope(anyhow::Error),
     #[error("Send error: {}", _0)]
     SmtpSend(async_smtp::error::Error),
-    #[error("SMTP has no transport")]
-    NoTransport,
     #[error("{}", _0)]
     Other(#[from] anyhow::Error),
 }
 
-impl Smtp {
+impl Connection {
     /// Send a prepared mail to recipients.
     /// On successful send out Ok() is returned.
-    pub async fn send(
+    pub(super) async fn send(
         &mut self,
         context: &Context,
         recipients: &[EmailAddress],
@@ -38,21 +34,13 @@ impl Smtp {
             context.ratelimit.write().await.send();
         }
 
-        let envelope =
-            Envelope::new(self.from.clone(), recipients.to_vec()).map_err(Error::Envelope)?;
+        let envelope = Envelope::new(Some(self.from.clone()), recipients.to_vec())
+            .context("Envelope error")?;
         let mail = SendableEmail::new(envelope, message);
 
-        let Some(ref mut transport) = self.transport else {
-            warn!(
-                context,
-                "Failed to send a message because SMTP client has no SmtpTransport."
-            );
-            return Err(Error::NoTransport);
-        };
+        self.transport.send(mail).await.map_err(Error::SmtpSend)?;
 
-        transport.send(mail).await.map_err(Error::SmtpSend)?;
-
-        self.last_success = Some(tools::Time::now());
+        self.last_success = tools::Time::now();
         Ok(())
     }
 }

src/smtp/smtp_tests.rs

@@ -1,8 +1,25 @@
 use anyhow::Result;
 
-use crate::test_utils::TestContextManager;
+use crate::test_utils::{TestContext, TestContextManager};
 use crate::transport;
 
+async fn sorted_transport_addrs(t: &TestContext) -> Result<String> {
+    let transports = super::sorted_transports(t).await?;
+    let addrs: Vec<String> = transports
+        .into_iter()
+        .map(|(_transport_id, transport)| transport.addr)
+        .collect();
+    Ok(addrs.join(" "))
+}
+
+async fn sorted_transport_ids(t: &TestContext) -> Result<Vec<u32>> {
+    let transports = super::sorted_transports(t).await?;
+    Ok(transports
+        .into_iter()
+        .map(|(transport_id, _transport)| transport_id)
+        .collect())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_smtp_candidates() -> Result<()> {
     let mut tcm = TestContextManager::new();
@@ -12,38 +29,42 @@ async fn test_smtp_candidates() -> Result<()> {
     transport::add_pseudo_transport(t, "bar@example.net").await?;
     transport::add_pseudo_transport(t, "baz@example.net").await?;
 
-    let transports = super::sorted_transports(t).await?;
-    let [
-        (transport_id1, ref transport1),
-        (transport_id2, ref transport2),
-        (transport_id3, ref transport3),
-    ] = transports[..]
-    else {
-        panic!("Unexpected number of transports");
-    };
-
     // By default first added transport is used first.
-    assert_eq!(transport1.addr, "foo@example.net");
-    assert_eq!(transport2.addr, "bar@example.net");
-    assert_eq!(transport3.addr, "baz@example.net");
+    assert_eq!(
+        sorted_transport_addrs(t).await?,
+        "foo@example.net bar@example.net baz@example.net"
+    );
+    assert_eq!(sorted_transport_ids(t).await?, [1, 2, 3]);
 
-    super::record_success(t, transport_id3).await?;
-    let transports2 = super::sorted_transports(t).await?;
-    assert_eq!(transports2[0].0, transport_id3);
-    assert_eq!(transports2[1].0, transport_id1);
-    assert_eq!(transports2[2].0, transport_id2);
+    super::record_success(t, 3).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [3, 1, 2]);
 
-    super::record_success(t, transport_id2).await?;
-    let transports3 = super::sorted_transports(t).await?;
-    assert_eq!(transports3[0].0, transport_id2);
-    assert_eq!(transports3[1].0, transport_id3);
-    assert_eq!(transports3[2].0, transport_id1);
+    super::record_success(t, 2).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [2, 3, 1]);
 
-    super::record_success(t, transport_id3).await?;
-    let transports4 = super::sorted_transports(t).await?;
-    assert_eq!(transports4[0].0, transport_id3);
-    assert_eq!(transports4[1].0, transport_id2);
-    assert_eq!(transports4[2].0, transport_id1);
+    super::record_success(t, 3).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [3, 2, 1]);
+
+    super::record_failure(t, 3).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [2, 1, 3]);
+
+    // New transport is added before all failing transports.
+    transport::add_pseudo_transport(t, "qux@example.net").await?;
+    assert_eq!(
+        sorted_transport_addrs(t).await?,
+        "bar@example.net foo@example.net qux@example.net baz@example.net"
+    );
+    assert_eq!(sorted_transport_ids(t).await?, [2, 1, 4, 3]);
+
+    super::record_success(t, 4).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [4, 2, 1, 3]);
+
+    super::record_failure(t, 4).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [2, 1, 3, 4]);
+    super::record_failure(t, 3).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [2, 1, 4, 3]);
+    super::record_failure(t, 2).await?;
+    assert_eq!(sorted_transport_ids(t).await?, [1, 4, 3, 2]);
 
     Ok(())
 }

src/sql/migrations.rs

@@ -2687,6 +2687,21 @@ CREATE TABLE smtp_success (
         .await?;
     }
 
+    inc_and_check(&mut migration_version, 169)?;
+    if dbversion < migration_version {
+        sql.execute_migration(
+            "
+CREATE TABLE smtp_failure (
+    id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+    transport_id INTEGER UNIQUE NOT NULL,
+    FOREIGN KEY(transport_id) REFERENCES transports(id) ON DELETE CASCADE
+) STRICT;
+",
+            migration_version,
+        )
+        .await?;
+    }
+
     let new_version = sql
         .get_raw_config_int(VERSION_CFG)
         .await?