move smtp-send chunking from chat.rs to smtp loop and use max-recipients value for the smtp transport in use, add chunking tests

src/chat.rs

@@ -49,8 +49,8 @@ use crate::stock_str;
 use crate::sync::{self, Sync::*, SyncData};
 use crate::tools::{
     IsNoneOrEmpty, SystemTime, buf_compress, create_broadcast_secret, create_id,
-    create_outgoing_rfc724_mid, get_abs_path, gm2local_offset, normalize_text, time,
-    truncate_msg_text,
+    create_outgoing_rfc724_mid, create_smeared_timestamp, create_smeared_timestamps, get_abs_path,
+    gm2local_offset, normalize_text, smeared_time, time, truncate_msg_text,
 };
 use crate::webxdc::StatusUpdateSerial;
 
@@ -291,7 +291,7 @@ impl ChatId {
         timestamp: i64,
     ) -> Result<Self> {
         let grpname = sanitize_single_line(grpname);
-        let timestamp = cmp::min(timestamp, time());
+        let timestamp = cmp::min(timestamp, smeared_time(context));
         let row_id =
             context.sql.insert(
                 "INSERT INTO chats (type, name, name_normalized, grpid, blocked, created_timestamp, protected, param) VALUES(?, ?, ?, ?, ?, ?, 0, ?)",
@@ -1255,7 +1255,7 @@ SELECT id, rfc724_mid, pre_rfc724_mid, timestamp, ?, 1 FROM msgs WHERE chat_id=?
         message_timestamp: i64,
         always_sort_to_bottom: bool,
     ) -> Result<i64> {
-        let mut sort_timestamp = cmp::min(message_timestamp, time());
+        let mut sort_timestamp = cmp::min(message_timestamp, smeared_time(context));
 
         let last_msg_time: Option<i64> = if always_sort_to_bottom {
             // get newest message for this chat
@@ -2405,7 +2405,7 @@ impl ChatIdBlocked {
             _ => (),
         }
 
-        let now = time();
+        let smeared_time = create_smeared_timestamp(context);
 
         let chat_id = context
             .sql
@@ -2420,7 +2420,7 @@ impl ChatIdBlocked {
                         normalize_text(&chat_name),
                         params.to_string(),
                         create_blocked as u8,
-                        now,
+                        smeared_time,
                     ),
                 )?;
                 let chat_id = ChatId::new(
@@ -2446,7 +2446,7 @@ impl ChatIdBlocked {
             && !chat.param.exists(Param::Devicetalk)
             && !chat.param.exists(Param::Selftalk)
         {
-            chat_id.add_e2ee_notice(context, now).await?;
+            chat_id.add_e2ee_notice(context, smeared_time).await?;
         }
 
         Ok(Self {
@@ -2733,7 +2733,7 @@ async fn prepare_send_msg(
     }
     msg.state = MessageState::OutPending;
 
-    msg.timestamp_sort = time();
+    msg.timestamp_sort = create_smeared_timestamp(context);
     prepare_msg_blob(context, msg).await?;
     if !msg.hidden {
         chat_id.unarchive_if_not_muted(context, msg.state).await?;
@@ -2907,7 +2907,7 @@ pub(crate) async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -
         );
     }
 
-    let now = time();
+    let now = smeared_time(context);
 
     if rendered_msg.last_added_location_id.is_some()
         && let Err(err) = location::set_kml_sent_timestamp(context, msg.chat_id, now).await
@@ -2954,7 +2954,6 @@ WHERE id=?
         )
         .await?;
 
-    let chunk_size = context.get_max_smtp_rcpt_to().await?;
     let trans_fn = |t: &mut rusqlite::Transaction| {
         let mut row_ids = Vec::<i64>::new();
 
@@ -2968,12 +2967,12 @@ WHERE id=?
             "INSERT INTO smtp (rfc724_mid, recipients, mime, msg_id)
             VALUES            (?1,         ?2,         ?3,   ?4)",
         )?;
-        for recipients_chunk in recipients.chunks(chunk_size) {
-            let recipients_chunk = recipients_chunk.join(" ");
+        if !recipients.is_empty() {
+            let all_recipients = recipients.join(" ");
             if let Some(pre_msg) = &rendered_pre_msg {
                 let row_id = stmt.execute((
                     &pre_msg.rfc724_mid,
-                    &recipients_chunk,
+                    &all_recipients,
                     &pre_msg.message,
                     msg.id,
                 ))?;
@@ -2981,7 +2980,7 @@ WHERE id=?
             }
             let row_id = stmt.execute((
                 &rendered_msg.rfc724_mid,
-                &recipients_chunk,
+                &all_recipients,
                 &rendered_msg.message,
                 msg.id,
             ))?;
@@ -3549,7 +3548,7 @@ pub(crate) async fn create_group_ex(
         chat_name = "…".to_string();
     }
 
-    let timestamp = time();
+    let timestamp = create_smeared_timestamp(context);
     let row_id = context
         .sql
         .insert(
@@ -3635,7 +3634,7 @@ pub(crate) async fn create_out_broadcast_ex(
         bail!("Invalid broadcast channel name: {chat_name}.");
     }
 
-    let timestamp = time();
+    let timestamp = create_smeared_timestamp(context);
     let trans_fn = |t: &mut rusqlite::Transaction| -> Result<ChatId> {
         let cnt: u32 = t.query_row(
             "SELECT COUNT(*) FROM chats WHERE grpid=?",
@@ -3885,11 +3884,11 @@ pub(crate) async fn add_contact_to_chat_ex(
         return Ok(false);
     }
     if from_handshake && chat.param.get_int(Param::Unpromoted).unwrap_or_default() == 1 {
-        let now = time();
+        let smeared_time = smeared_time(context);
         chat.param
             .remove(Param::Unpromoted)
-            .set_i64(Param::GroupNameTimestamp, now)
-            .set_i64(Param::GroupDescriptionTimestamp, now);
+            .set_i64(Param::GroupNameTimestamp, smeared_time)
+            .set_i64(Param::GroupDescriptionTimestamp, smeared_time);
         chat.update_param(context).await?;
     }
     if context.is_self_addr(contact.get_addr()).await? {
@@ -4452,6 +4451,7 @@ pub async fn forward_msgs(context: &Context, msg_ids: &[MsgId], chat_id: ChatId)
 }
 
 /// Forwards multiple messages to a chat in another context.
+#[expect(clippy::arithmetic_side_effects)]
 pub async fn forward_msgs_2ctx(
     ctx_src: &Context,
     msg_ids: &[MsgId],
@@ -4462,6 +4462,7 @@ pub async fn forward_msgs_2ctx(
     ensure!(!chat_id.is_special(), "can not forward to special chat");
 
     let mut created_msgs: Vec<MsgId> = Vec::new();
+    let mut curr_timestamp: i64;
 
     chat_id
         .unarchive_if_not_muted(ctx_dst, MessageState::Undefined)
@@ -4470,7 +4471,7 @@ pub async fn forward_msgs_2ctx(
     if let Some(reason) = chat.why_cant_send(ctx_dst).await? {
         bail!("cannot send to {chat_id}: {reason}");
     }
-    let now = time();
+    curr_timestamp = create_smeared_timestamps(ctx_dst, msg_ids.len());
     let mut msgs = Vec::with_capacity(msg_ids.len());
     for id in msg_ids {
         let ts: i64 = ctx_src
@@ -4535,9 +4536,10 @@ pub async fn forward_msgs_2ctx(
         msg.state = MessageState::OutPending;
         msg.rfc724_mid = create_outgoing_rfc724_mid();
         msg.pre_rfc724_mid.clear();
-        msg.timestamp_sort = now;
+        msg.timestamp_sort = curr_timestamp;
         chat.prepare_msg_raw(ctx_dst, &mut msg, None).await?;
 
+        curr_timestamp += 1;
         if !create_send_msg_jobs(ctx_dst, &mut msg).await?.is_empty() {
             ctx_dst.scheduler.interrupt_smtp().await;
         }
@@ -4630,7 +4632,7 @@ pub(crate) async fn save_copy_in_self_talk(
                 } else {
                     MessageState::InSeen
                 },
-                time(),
+                create_smeared_timestamp(context),
                 msg.param.to_string(),
                 src_msg_id,
                 src_msg_id,
@@ -4807,7 +4809,7 @@ pub async fn add_device_msg_with_importance(
         chat_id = ChatId::get_for_contact(context, ContactId::DEVICE).await?;
 
         let rfc724_mid = create_outgoing_rfc724_mid();
-        let timestamp_sent = time();
+        let timestamp_sent = create_smeared_timestamp(context);
 
         // makes sure, the added message is the last one,
         // even if the date is wrong (useful esp. when warning about bad dates)
@@ -4954,7 +4956,7 @@ pub(crate) async fn add_info_msg_with_cmd(
     } else {
         let sort_to_bottom = true;
         chat_id
-            .calc_sort_timestamp(context, time(), sort_to_bottom)
+            .calc_sort_timestamp(context, smeared_time(context), sort_to_bottom)
             .await?
     };
 
@@ -5117,7 +5119,7 @@ async fn set_contacts_by_fingerprints(
             Ok(broadcast_contacts_added)
         })
         .await?;
-    let timestamp = time();
+    let timestamp = smeared_time(context);
     for added_id in broadcast_contacts_added {
         let msg = stock_str::msg_add_member_local(context, added_id, ContactId::UNDEFINED).await;
         add_info_msg_with_cmd(

src/chat/chat_tests.rs

@@ -1625,7 +1625,6 @@ async fn test_set_chat_name() {
         "another name",
         "something different",
     ] {
-        SystemTime::shift(Duration::from_secs(1));
         set_chat_name(alice, chat_id, new_name).await.unwrap();
         let sent_msg = alice.pop_sent_msg().await;
         let received_msg = bob.recv_msg(&sent_msg).await;
@@ -3438,9 +3437,8 @@ async fn test_chat_description(
         "",
         "ä ẟ 😂",
     ] {
-        SystemTime::shift(Duration::from_secs(1));
         tcm.section(&format!(
-            "Alice sets the chat description to {description:?}"
+            "Alice sets the chat description to '{description}'"
         ));
         set_chat_description(alice, alice_chat_id, description).await?;
         let sent = alice.pop_sent_msg().await;
@@ -4461,9 +4459,7 @@ async fn test_get_chat_media_webxdc_order() -> Result<()> {
     assert_eq!(media.first().unwrap(), &instance1_id);
     assert_eq!(media.get(1).unwrap(), &instance2_id);
 
-    SystemTime::shift(Duration::from_secs(1));
-
-    // add a status update for the other instance; that resorts the list
+    // add a status update for the oder instance; that resorts the list
     alice
         .send_webxdc_status_update(instance1_id, r#"{"payload": {"foo": "bar"}}"#)
         .await?;
@@ -4877,6 +4873,10 @@ async fn test_sync_broadcast_and_send_message() -> Result<()> {
         vec![a2b_contact_id]
     );
 
+    // alice2's smeared clock may be behind alice1's one, so we need to work around "hi" appearing
+    // before "You joined the channel." for bob. alice1 makes 3 more calls of
+    // create_smeared_timestamp() than alice2 does as of 2026-03-10.
+    SystemTime::shift(Duration::from_secs(3));
     tcm.section("Alice's second device sends a message to the channel");
     let sent_msg = alice2.send_text(a2_broadcast_id, "hi").await;
     let msg = bob.recv_msg(&sent_msg).await;

src/constants.rs

@@ -204,9 +204,6 @@ pub const MAX_RCVD_IMAGE_PIXELS: u32 = 50_000_000;
 // `max_smtp_rcpt_to` in the provider db.
 pub(crate) const DEFAULT_MAX_SMTP_RCPT_TO: usize = 50;
 
-/// Same as `DEFAULT_MAX_SMTP_RCPT_TO`, but for chatmail relays.
-pub(crate) const DEFAULT_CHATMAIL_MAX_SMTP_RCPT_TO: usize = 999;
-
 /// How far the last quota check needs to be in the past to be checked by the background function (in seconds).
 pub(crate) const DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT: u64 = 12 * 60 * 60; // 12 hours
 

src/context.rs

@@ -16,7 +16,7 @@ use tokio::sync::{Mutex, Notify, RwLock};
 
 use crate::chat::{ChatId, get_chat_cnt};
 use crate::config::Config;
-use crate::constants::{self, DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT, DC_VERSION_STR};
+use crate::constants::{DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT, DC_VERSION_STR};
 use crate::contact::{Contact, ContactId};
 use crate::debug_logging::DebugLogging;
 use crate::events::{Event, EventEmitter, EventType, Events};
@@ -32,6 +32,7 @@ use crate::quota::QuotaInfo;
 use crate::scheduler::{ConnectivityStore, SchedulerState};
 use crate::sql::Sql;
 use crate::stock_str::StockStrings;
+use crate::timesmearing::SmearedTimestamp;
 use crate::tools::{self, duration_to_str, time, time_elapsed};
 use crate::transport::ConfiguredLoginParam;
 use crate::{chatlist_events, stats};
@@ -227,6 +228,7 @@ pub struct InnerContext {
     /// Blob directory path
     pub(crate) blobdir: PathBuf,
     pub(crate) sql: Sql,
+    pub(crate) smeared_timestamp: SmearedTimestamp,
     /// The global "ongoing" process state.
     ///
     /// This is a global mutex-like state for operations which should be modal in the
@@ -496,6 +498,7 @@ impl Context {
             blobdir,
             running_state: RwLock::new(Default::default()),
             sql: Sql::new(dbfile),
+            smeared_timestamp: SmearedTimestamp::new(),
             oauth2_mutex: Mutex::new(()),
             wrong_pw_warning_mutex: Mutex::new(()),
             housekeeping_mutex: Mutex::new(()),
@@ -584,23 +587,6 @@ impl Context {
         self.get_config_bool(Config::IsChatmail).await
     }
 
-    /// Returns maximum number of recipients the provider allows to send a single email to.
-    pub(crate) async fn get_max_smtp_rcpt_to(&self) -> Result<usize> {
-        let is_chatmail = self.is_chatmail().await?;
-        let val = self
-            .get_configured_provider()
-            .await?
-            .and_then(|provider| provider.opt.max_smtp_rcpt_to)
-            .map_or_else(
-                || match is_chatmail {
-                    true => constants::DEFAULT_CHATMAIL_MAX_SMTP_RCPT_TO,
-                    false => constants::DEFAULT_MAX_SMTP_RCPT_TO,
-                },
-                usize::from,
-            );
-        Ok(val)
-    }
-
     /// Does a single round of fetching from IMAP and returns.
     ///
     /// Can be used even if I/O is currently stopped.

src/ephemeral/ephemeral_tests.rs

@@ -10,6 +10,7 @@ use crate::location;
 use crate::message::markseen_msgs;
 use crate::receive_imf::receive_imf;
 use crate::test_utils::{TestContext, TestContextManager};
+use crate::timesmearing::MAX_SECONDS_TO_LEND_FROM_FUTURE;
 use crate::{
     chat::{self, Chat, ChatItem, create_group, send_text_msg},
     tools::IsNoneOrEmpty,
@@ -351,9 +352,17 @@ async fn test_ephemeral_delete_msgs() -> Result<()> {
     let now = time();
     let msg = t.send_text(bob_chat.id, "Message text").await;
 
-    check_msg_will_be_deleted(&t, msg.sender_msg_id, &bob_chat, now + 1799, time() + 1801)
-        .await
-        .unwrap();
+    check_msg_will_be_deleted(
+        &t,
+        msg.sender_msg_id,
+        &bob_chat,
+        now + 1799,
+        // The message may appear to be sent MAX_SECONDS_TO_LEND_FROM_FUTURE later and
+        // therefore be deleted MAX_SECONDS_TO_LEND_FROM_FUTURE later.
+        time() + 1801 + MAX_SECONDS_TO_LEND_FROM_FUTURE,
+    )
+    .await
+    .unwrap();
 
     // Enable ephemeral messages with Bob -> message will be deleted after 60s.
     // This tests that the message is deleted at min(ephemeral deletion time, DeleteDeviceAfter deletion time).

src/imap.rs

@@ -1503,7 +1503,7 @@ impl Session {
             .get_metadata(
                 mailbox,
                 options,
-                "(/shared/comment /shared/admin /shared/vendor/deltachat/irohrelay /shared/vendor/deltachat/turn)",
+                "(/shared/comment /shared/admin /shared/vendor/deltachat/irohrelay /shared/vendor/deltachat/turn /shared/vendor/deltachat/maxsmtprecipients)",
             )
             .await?;
         for m in metadata {
@@ -1539,6 +1539,21 @@ impl Session {
                         }
                     }
                 }
+                "/shared/vendor/deltachat/maxsmtprecipients" => {
+                    if let Some(value) = m.value.and_then(|v| v.parse::<u32>().ok()) {
+                        let transport_id = self.transport_id();
+                        context
+                            .sql
+                            .execute(
+                                "UPDATE transports \
+                                 SET max_smtp_rcpt_to=? WHERE id=?",
+                                (value, transport_id),
+                            )
+                            .await
+                            .log_err(context)
+                            .ok();
+                    }
+                }
                 _ => {}
             }
         }

src/lib.rs

@@ -94,6 +94,7 @@ mod smtp;
 pub mod stock_str;
 pub mod storage_usage;
 mod sync;
+mod timesmearing;
 mod token;
 mod transport;
 mod update_helper;

src/mimefactory.rs

@@ -35,7 +35,10 @@ use crate::peer_channels::{create_iroh_header, get_iroh_topic_for_msg};
 use crate::pgp::{SeipdVersion, addresses_from_public_key, pubkey_supports_seipdv2};
 use crate::simplify::escape_message_footer_marks;
 use crate::stock_str;
-use crate::tools::{IsNoneOrEmpty, create_outgoing_rfc724_mid, remove_subject_prefix, time};
+use crate::tools::{
+    IsNoneOrEmpty, create_outgoing_rfc724_mid, create_smeared_timestamp, remove_subject_prefix,
+    time,
+};
 use crate::webxdc::StatusUpdateSerial;
 
 // attachments of 25 mb brutto should work on the majority of providers
@@ -577,7 +580,7 @@ impl MimeFactory {
     ) -> Result<MimeFactory> {
         let contact = Contact::get_by_id(context, from_id).await?;
         let from_addr = context.get_primary_self_addr().await?;
-        let timestamp = time();
+        let timestamp = create_smeared_timestamp(context);
 
         let addr = contact.get_addr().to_string();
         let encryption_pubkeys = if from_id == ContactId::SELF {
@@ -2298,7 +2301,7 @@ pub(crate) async fn render_symm_encrypted_securejoin_message(
         mail_builder::headers::text::Text::new("Secure-Join".to_string()).into(),
     ));
 
-    let timestamp = time();
+    let timestamp = create_smeared_timestamp(context);
     let date = chrono::DateTime::<chrono::Utc>::from_timestamp(timestamp, 0)
         .unwrap()
         .to_rfc2822();

src/mimeparser.rs

@@ -31,7 +31,9 @@ use crate::message::{self, Message, MsgId, Viewtype, get_vcard_summary, set_msg_
 use crate::param::{Param, Params};
 use crate::simplify::{SimplifiedText, simplify};
 use crate::sync::SyncItems;
-use crate::tools::{get_filemeta, parse_receive_headers, time, truncate_msg_text, validate_id};
+use crate::tools::{
+    get_filemeta, parse_receive_headers, smeared_time, time, truncate_msg_text, validate_id,
+};
 use crate::{chatlist_events, location, tools};
 
 /// Public key extracted from `Autocrypt-Gossip`
@@ -269,7 +271,7 @@ impl MimeMessage {
     pub(crate) async fn from_bytes(context: &Context, body: &[u8]) -> Result<Self> {
         let mail = mailparse::parse_mail(body)?;
 
-        let timestamp_rcvd = time();
+        let timestamp_rcvd = smeared_time(context);
         let mut timestamp_sent =
             Self::get_timestamp_sent(&mail.headers, timestamp_rcvd, timestamp_rcvd);
         let hop_info = parse_receive_headers(&mail.get_headers());

src/receive_imf/receive_imf_tests.rs

@@ -3975,8 +3975,6 @@ async fn test_delayed_removal_is_ignored() -> Result<()> {
     remove_contact_from_chat(bob, bob_chat_id, bob_contact_fiona).await?;
     let remove_msg = bob.pop_sent_msg().await;
 
-    SystemTime::shift(Duration::from_secs(1));
-
     // Bob adds new members Dom and Elena, but first addition message is lost.
     let dom = &tcm.dom().await;
     let elena = &tcm.elena().await;
@@ -3993,8 +3991,6 @@ async fn test_delayed_removal_is_ignored() -> Result<()> {
     alice.recv_msg(&add_msg).await;
     assert_eq!(get_chat_contacts(alice, chat_id).await?.len(), 4);
 
-    SystemTime::shift(Duration::from_secs(1));
-
     // Alice re-adds Fiona.
     add_contact_to_chat(alice, chat_id, alice_fiona).await?;
     assert_eq!(get_chat_contacts(alice, chat_id).await?.len(), 5);

src/securejoin/bob.rs

@@ -19,7 +19,7 @@ use crate::securejoin::{
 };
 use crate::stock_str;
 use crate::sync::Sync::*;
-use crate::tools::{create_outgoing_rfc724_mid, time};
+use crate::tools::{create_outgoing_rfc724_mid, smeared_time, time};
 use crate::{chatlist_events, mimefactory};
 
 /// Starts the securejoin protocol with the QR `invite`.
@@ -465,7 +465,7 @@ async fn joining_chat_id(
                         name,
                         Blocked::Not,
                         None,
-                        time(),
+                        smeared_time(context),
                     )
                     .await?
                 }

src/smtp.rs

@@ -2,6 +2,8 @@
 
 mod connect;
 pub mod send;
+#[cfg(test)]
+mod chunking_tests;
 
 use anyhow::{Context as _, Error, Result, bail, format_err};
 use async_smtp::response::{Category, Code, Detail};
@@ -10,6 +12,7 @@ use tokio::task;
 
 use crate::chat::{ChatId, add_info_msg_with_cmd};
 use crate::config::Config;
+use crate::constants;
 use crate::contact::{Contact, ContactId};
 use crate::context::Context;
 use crate::events::EventType;
@@ -34,6 +37,9 @@ pub(crate) struct Smtp {
     /// Email address we are sending from.
     from: Option<EmailAddress>,
 
+    /// Transport used for the current connection.
+    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.
@@ -60,6 +66,7 @@ impl Smtp {
             task::spawn(async move { transport.quit().await });
         }
         self.last_success = None;
+        self.transport_id = None;
     }
 
     /// Return true if smtp was connected but is not known to
@@ -89,9 +96,10 @@ impl Smtp {
         }
 
         self.connectivity.set_connecting(context);
-        let (_transport_id, lp) = ConfiguredLoginParam::load(context)
+        let (transport_id, lp) = ConfiguredLoginParam::load(context)
             .await?
             .context("Not configured")?;
+        self.transport_id = Some(transport_id);
         let proxy_config = ProxyConfig::load(context).await?;
         self.connect(
             context,
@@ -165,6 +173,7 @@ impl Smtp {
     }
 }
 
+#[derive(Debug)]
 pub(crate) enum SendResult {
     /// Message was sent successfully.
     Success,
@@ -176,13 +185,36 @@ pub(crate) enum SendResult {
     Retry,
 }
 
+pub(crate) trait SmtpSender: Send {
+    fn send_chunk<'a>(
+        &'a mut self,
+        context: &'a Context,
+        recipients: &'a [async_smtp::EmailAddress],
+        body: &'a str,
+    ) -> futures::future::BoxFuture<'a, SendResult>;
+}
+
+struct RealSmtpSender<'a> {
+    smtp: &'a mut Smtp,
+}
+
+impl SmtpSender for RealSmtpSender<'_> {
+    fn send_chunk<'a>(
+        &'a mut self,
+        context: &'a Context,
+        recipients: &'a [async_smtp::EmailAddress],
+        body: &'a str,
+    ) -> futures::future::BoxFuture<'a, SendResult> {
+        Box::pin(smtp_send(context, recipients, body, self.smtp))
+    }
+}
+
 /// Tries to send a message.
 pub(crate) async fn smtp_send(
     context: &Context,
     recipients: &[async_smtp::EmailAddress],
     message: &str,
     smtp: &mut Smtp,
-    msg_id: Option<MsgId>,
 ) -> SendResult {
     if recipients.is_empty() {
         return SendResult::Success;
@@ -310,25 +342,6 @@ pub(crate) async fn smtp_send(
         Ok(()) => SendResult::Success,
     };
 
-    if let SendResult::Failure(err) = &status
-        && let Some(msg_id) = msg_id
-    {
-        // We couldn't send the message, so mark it as failed
-        match Message::load_from_db(context, msg_id).await {
-            Ok(mut msg) => {
-                if let Err(err) = message::set_msg_failed(context, &mut msg, &err.to_string()).await
-                {
-                    error!(context, "Failed to mark {msg_id} as failed: {err:#}.");
-                }
-            }
-            Err(err) => {
-                error!(
-                    context,
-                    "Failed to load {msg_id} to mark it as failed: {err:#}."
-                );
-            }
-        }
-    }
     status
 }
 
@@ -406,7 +419,40 @@ pub(crate) async fn send_msg_to_smtp(
         )
         .collect::<Vec<_>>();
 
-    let status = smtp_send(context, &recipients_list, body.as_str(), smtp, Some(msg_id)).await;
+    let transport_id = smtp
+        .transport_id
+        .context("SMTP not connected to a transport")?;
+    let chunk_size = max_smtp_rcpt_to(context, transport_id).await?;
+
+    let mut sender = RealSmtpSender { smtp };
+    let (status, start_idx) = send_smtp_chunks(
+        context,
+        &recipients_list,
+        body.as_str(),
+        chunk_size,
+        &mut sender,
+    )
+    .await;
+
+    let unsent_saved = start_idx < recipients_list.len();
+    if let Some(unsent) = recipients_list.get(start_idx..)
+        && !unsent.is_empty()
+    {
+        let unsent_str: String = unsent
+            .iter()
+            .map(|a| a.as_ref())
+            .collect::<Vec<&str>>()
+            .join(" ");
+        context
+            .sql
+            .execute(
+                "UPDATE smtp SET recipients=? WHERE id=?",
+                (unsent_str, rowid),
+            )
+            .await
+            .log_err(context)
+            .ok();
+    }
 
     match status {
         SendResult::Retry => {}
@@ -455,10 +501,15 @@ pub(crate) async fn send_msg_to_smtp(
                     .await?;
                 };
             }
-            context
-                .sql
-                .execute("DELETE FROM smtp WHERE id=?", (rowid,))
-                .await?;
+            if let Some(mut msg) = Message::load_from_db_optional(context, msg_id).await? {
+                message::set_msg_failed(context, &mut msg, &err.to_string()).await?;
+            }
+            if !unsent_saved {
+                context
+                    .sql
+                    .execute("DELETE FROM smtp WHERE id=?", (rowid,))
+                    .await?;
+            }
         }
     };
 
@@ -470,10 +521,39 @@ pub(crate) async fn send_msg_to_smtp(
             }
             Ok(())
         }
-        SendResult::Failure(err) => Err(format_err!("{err}")),
+        SendResult::Failure(err) => {
+            if unsent_saved {
+                Err(format_err!("Retry"))
+            } else {
+                Err(format_err!("{err}"))
+            }
+        }
     }
 }
 
+async fn max_smtp_rcpt_to(context: &Context, transport_id: u32) -> Result<usize> {
+    let limit = context
+        .sql
+        .query_row_optional(
+            "SELECT max_smtp_rcpt_to FROM transports WHERE id=?",
+            (transport_id,),
+            |row| row.get::<_, u32>(0),
+        )
+        .await?
+        .unwrap_or(0);
+
+    if limit > 0 {
+        return Ok(limit as usize);
+    }
+
+    let val = context
+        .get_configured_provider()
+        .await?
+        .and_then(|provider| provider.opt.max_smtp_rcpt_to)
+        .map_or(constants::DEFAULT_MAX_SMTP_RCPT_TO, usize::from);
+    Ok(val)
+}
+
 pub(crate) async fn msg_has_pending_smtp_job(
     context: &Context,
     msg_id: MsgId,
@@ -600,7 +680,7 @@ async fn send_mdn_rfc724_mid(
         })
         .collect();
 
-    match smtp_send(context, &recipients, &body, smtp, None).await {
+    match smtp_send(context, &recipients, &body, smtp).await {
         SendResult::Success => {
             if !recipients.is_empty() {
                 info!(context, "Successfully sent MDN for {rfc724_mid}.");
@@ -722,3 +802,22 @@ pub(crate) async fn add_self_recipients(
     }
     Ok(())
 }
+
+#[allow(clippy::arithmetic_side_effects)]
+pub(crate) async fn send_smtp_chunks(
+    context: &Context,
+    recipients: &[async_smtp::EmailAddress],
+    body: &str,
+    chunk_size: usize,
+    sender: &mut (dyn SmtpSender + Send),
+) -> (SendResult, usize) {
+    for (i, chunk) in recipients.chunks(chunk_size).enumerate() {
+        let status = sender.send_chunk(context, chunk, body).await;
+        match status {
+            SendResult::Success => continue,
+            SendResult::Failure(_) => return (status, (i + 1) * chunk_size),
+            SendResult::Retry => return (status, i * chunk_size),
+        }
+    }
+    (SendResult::Success, recipients.len())
+}

src/smtp/chunking_tests.rs

@@ -0,0 +1,102 @@
+use crate::smtp::{send_smtp_chunks, SendResult, SmtpSender};
+use crate::test_utils::TestContextManager;
+use crate::context::Context;
+use anyhow::Result;
+use futures::future::{BoxFuture, FutureExt};
+
+/// Result the mock should return on the designated call.
+enum MockFailure {
+    Transient,
+    Permanent,
+}
+
+struct MockSmtpSender {
+    call_count: usize,
+    fail_on_call: Option<(usize, MockFailure)>,
+}
+
+impl SmtpSender for MockSmtpSender {
+    fn send_chunk<'a>(
+        &'a mut self,
+        _context: &'a Context,
+        _recipients: &'a [async_smtp::EmailAddress],
+        _body: &'a str,
+    ) -> BoxFuture<'a, SendResult> {
+        self.call_count += 1;
+        let count = self.call_count;
+        let fail_on = self.fail_on_call.as_ref().map(|(n, _)| *n);
+        let is_permanent = matches!(
+            self.fail_on_call,
+            Some((_, MockFailure::Permanent))
+        );
+        async move {
+            if fail_on == Some(count) {
+                if is_permanent {
+                    return SendResult::Failure(
+                        anyhow::format_err!("permanent error"),
+                    );
+                }
+                return SendResult::Retry;
+            }
+            SendResult::Success
+        }
+        .boxed()
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_send_smtp_chunks() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice = tcm.alice().await;
+
+    let recipients: Vec<_> = ["r1@ex.org", "r2@ex.org", "r3@ex.org", "r4@ex.org", "r5@ex.org"]
+        .iter()
+        .map(|a| async_smtp::EmailAddress::new(a.to_string()).unwrap())
+        .collect();
+
+    // All chunks succeed.
+    let mut sender = MockSmtpSender { call_count: 0, fail_on_call: None };
+    let (status, processed) =
+        send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
+    assert!(matches!(status, SendResult::Success));
+    assert_eq!(processed, 5);
+    assert_eq!(sender.call_count, 3); // chunks: [2, 2, 1]
+
+    // Second chunk gets a transient error, only first chunk's recipients are processed.
+    let mut sender =
+        MockSmtpSender { call_count: 0, fail_on_call: Some((2, MockFailure::Transient)) };
+    let (status, processed) =
+        send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
+    assert!(matches!(status, SendResult::Retry));
+    assert_eq!(processed, 2);
+    assert_eq!(sender.call_count, 2);
+
+    // Last chunk gets a transient error, first two chunks' recipients are processed.
+    let mut sender =
+        MockSmtpSender { call_count: 0, fail_on_call: Some((3, MockFailure::Transient)) };
+    let (status, processed) =
+        send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
+    assert!(matches!(status, SendResult::Retry));
+    assert_eq!(processed, 4);
+    assert_eq!(sender.call_count, 3);
+
+    // Second chunk gets a permanent error; processed includes the failed chunk.
+    let mut sender =
+        MockSmtpSender { call_count: 0, fail_on_call: Some((2, MockFailure::Permanent)) };
+    let (status, processed) =
+        send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
+    assert!(matches!(status, SendResult::Failure(_)));
+    assert_eq!(processed, 4);
+    assert_eq!(sender.call_count, 2);
+
+    // Last chunk gets a permanent error; processed includes the failed chunk.
+    let mut sender =
+        MockSmtpSender { call_count: 0, fail_on_call: Some((3, MockFailure::Permanent)) };
+    let (status, processed) =
+        send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
+    assert!(matches!(status, SendResult::Failure(_)));
+    assert_eq!(processed, 6); // capped at (i+1)*chunk_size, may exceed len
+    assert_eq!(sender.call_count, 3);
+
+    Ok(())
+}

src/sql/migrations.rs

@@ -2385,6 +2385,15 @@ UPDATE msgs SET state=19 WHERE state=24; -- Change OutPreparing to OutFailed.
         .await?;
     }
 
+    inc_and_check(&mut migration_version, 153)?;
+    if dbversion < migration_version {
+        sql.execute_migration(
+            "ALTER TABLE transports ADD COLUMN max_smtp_rcpt_to INTEGER NOT NULL DEFAULT 0",
+            migration_version,
+        )
+        .await?;
+    }
+
     let new_version = sql
         .get_raw_config_int(VERSION_CFG)
         .await?

src/timesmearing.rs

@@ -0,0 +1,194 @@
+//! # Time smearing.
+//!
+//! As e-mails typically only use a second-based-resolution for timestamps,
+//! the order of two mails sent within one second is unclear.
+//! This is bad e.g. when forwarding some messages from a chat -
+//! these messages will appear at the recipient easily out of order.
+//!
+//! We work around this issue by not sending out two mails with the same timestamp.
+//! For this purpose, in short, we track the last timestamp used in `last_smeared_timestamp`
+//! when another timestamp is needed in the same second, we use `last_smeared_timestamp+1`
+//! after some moments without messages sent out,
+//! `last_smeared_timestamp` is again in sync with the normal time.
+//!
+//! However, we do not do all this for the far future,
+//! but at max `MAX_SECONDS_TO_LEND_FROM_FUTURE`
+
+use std::cmp::{max, min};
+use std::sync::atomic::{AtomicI64, Ordering};
+
+pub(crate) const MAX_SECONDS_TO_LEND_FROM_FUTURE: i64 = 30;
+
+/// Smeared timestamp generator.
+#[derive(Debug)]
+pub struct SmearedTimestamp {
+    /// Next timestamp available for allocation.
+    smeared_timestamp: AtomicI64,
+}
+
+impl SmearedTimestamp {
+    /// Creates a new smeared timestamp generator.
+    pub fn new() -> Self {
+        Self {
+            smeared_timestamp: AtomicI64::new(0),
+        }
+    }
+
+    /// Allocates `count` unique timestamps.
+    ///
+    /// Returns the first allocated timestamp.
+    #[expect(clippy::arithmetic_side_effects)]
+    pub fn create_n(&self, now: i64, count: i64) -> i64 {
+        let mut prev = self.smeared_timestamp.load(Ordering::Relaxed);
+        loop {
+            // Advance the timestamp if it is in the past,
+            // but keep `count - 1` timestamps from the past if possible.
+            let t = max(prev, now - count + 1);
+
+            // Rewind the time back if there is no room
+            // to allocate `count` timestamps without going too far into the future.
+            // Not going too far into the future
+            // is more important than generating unique timestamps.
+            let first = min(t, now + MAX_SECONDS_TO_LEND_FROM_FUTURE - count + 1);
+
+            // Allocate `count` timestamps by advancing the current timestamp.
+            let next = first + count;
+
+            if let Err(x) = self.smeared_timestamp.compare_exchange_weak(
+                prev,
+                next,
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+            ) {
+                prev = x;
+            } else {
+                return first;
+            }
+        }
+    }
+
+    /// Creates a single timestamp.
+    pub fn create(&self, now: i64) -> i64 {
+        self.create_n(now, 1)
+    }
+
+    /// Returns the current smeared timestamp.
+    pub fn current(&self) -> i64 {
+        self.smeared_timestamp.load(Ordering::Relaxed)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_utils::TestContext;
+    use crate::tools::{
+        SystemTime, create_smeared_timestamp, create_smeared_timestamps, smeared_time, time,
+    };
+
+    #[test]
+    fn test_smeared_timestamp() {
+        let smeared_timestamp = SmearedTimestamp::new();
+        let now = time();
+
+        assert_eq!(smeared_timestamp.current(), 0);
+
+        for i in 0..MAX_SECONDS_TO_LEND_FROM_FUTURE {
+            assert_eq!(smeared_timestamp.create(now), now + i);
+        }
+        assert_eq!(
+            smeared_timestamp.create(now),
+            now + MAX_SECONDS_TO_LEND_FROM_FUTURE
+        );
+        assert_eq!(
+            smeared_timestamp.create(now),
+            now + MAX_SECONDS_TO_LEND_FROM_FUTURE
+        );
+
+        // System time rewinds back by 1000 seconds.
+        let now = now - 1000;
+        assert_eq!(
+            smeared_timestamp.create(now),
+            now + MAX_SECONDS_TO_LEND_FROM_FUTURE
+        );
+        assert_eq!(
+            smeared_timestamp.create(now),
+            now + MAX_SECONDS_TO_LEND_FROM_FUTURE
+        );
+        assert_eq!(
+            smeared_timestamp.create(now + 1),
+            now + MAX_SECONDS_TO_LEND_FROM_FUTURE + 1
+        );
+        assert_eq!(smeared_timestamp.create(now + 100), now + 100);
+        assert_eq!(smeared_timestamp.create(now + 100), now + 101);
+        assert_eq!(smeared_timestamp.create(now + 100), now + 102);
+    }
+
+    #[test]
+    fn test_create_n_smeared_timestamps() {
+        let smeared_timestamp = SmearedTimestamp::new();
+        let now = time();
+
+        // Create a single timestamp to initialize the generator.
+        assert_eq!(smeared_timestamp.create(now), now);
+
+        // Wait a minute.
+        let now = now + 60;
+
+        // Simulate forwarding 7 messages.
+        let forwarded_messages = 7;
+
+        // We have not sent anything for a minute,
+        // so we can take the current timestamp and take 6 timestamps from the past.
+        assert_eq!(smeared_timestamp.create_n(now, forwarded_messages), now - 6);
+
+        assert_eq!(smeared_timestamp.current(), now + 1);
+
+        // Wait 4 seconds.
+        // Now we have 3 free timestamps in the past.
+        let now = now + 4;
+
+        assert_eq!(smeared_timestamp.current(), now - 3);
+
+        // Forward another 7 messages.
+        // We can only lend 3 timestamps from the past.
+        assert_eq!(smeared_timestamp.create_n(now, forwarded_messages), now - 3);
+
+        // We had to borrow 3 timestamps from the future
+        // because there were not enough timestamps in the past.
+        assert_eq!(smeared_timestamp.current(), now + 4);
+
+        // Forward another 32 messages.
+        // We cannot use more than 30 timestamps from the future,
+        // so we use 30 timestamps from the future,
+        // the current timestamp and one timestamp from the past.
+        assert_eq!(smeared_timestamp.create_n(now, 32), now - 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_create_smeared_timestamp() {
+        let t = TestContext::new().await;
+        assert_ne!(create_smeared_timestamp(&t), create_smeared_timestamp(&t));
+        assert!(
+            create_smeared_timestamp(&t)
+                >= SystemTime::now()
+                    .duration_since(SystemTime::UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs() as i64
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_create_smeared_timestamps() {
+        let t = TestContext::new().await;
+        let count = MAX_SECONDS_TO_LEND_FROM_FUTURE - 1;
+        let start = create_smeared_timestamps(&t, count as usize);
+        let next = smeared_time(&t);
+        assert!((start + count - 1) < next);
+
+        let count = MAX_SECONDS_TO_LEND_FROM_FUTURE + 30;
+        let start = create_smeared_timestamps(&t, count as usize);
+        let next = smeared_time(&t);
+        assert!((start + count - 1) < next);
+    }
+}

src/tools.rs

@@ -180,6 +180,29 @@ pub(crate) fn gm2local_offset() -> i64 {
     i64::from(lt.offset().local_minus_utc())
 }
 
+/// Returns the current smeared timestamp,
+///
+/// The returned timestamp MAY NOT be unique and MUST NOT go to "Date" header.
+pub(crate) fn smeared_time(context: &Context) -> i64 {
+    let now = time();
+    let ts = context.smeared_timestamp.current();
+    std::cmp::max(ts, now)
+}
+
+/// Returns a timestamp that is guaranteed to be unique.
+pub(crate) fn create_smeared_timestamp(context: &Context) -> i64 {
+    let now = time();
+    context.smeared_timestamp.create(now)
+}
+
+// creates `count` timestamps that are guaranteed to be unique.
+// the first created timestamps is returned directly,
+// get the other timestamps just by adding 1..count-1
+pub(crate) fn create_smeared_timestamps(context: &Context, count: usize) -> i64 {
+    let now = time();
+    context.smeared_timestamp.create_n(now, count as i64)
+}
+
 /// Returns the last release timestamp as a unix timestamp compatible for comparison with time() and
 /// database times.
 pub fn get_release_timestamp() -> i64 {

src/webxdc.rs

@@ -46,7 +46,7 @@ use crate::mimefactory::RECOMMENDED_FILE_SIZE;
 use crate::mimeparser::SystemMessage;
 use crate::param::Param;
 use crate::param::Params;
-use crate::tools::{create_id, get_abs_path, time};
+use crate::tools::{create_id, create_smeared_timestamp, get_abs_path};
 
 /// The current API version.
 /// If `min_api` in manifest.toml is set to a larger value,
@@ -558,7 +558,7 @@ impl Context {
             .create_status_update_record(
                 &instance,
                 status_update,
-                time(),
+                create_smeared_timestamp(self),
                 send_now,
                 ContactId::SELF,
             )

test-data/golden/test_sync_broadcast_alice1

@@ -1,7 +1,7 @@
 OutBroadcast#Chat#1001: Channel [0 member(s)]
 --------------------------------------------------------------------------------
 Msg#1001: info (Contact#Contact#Info): Messages are end-to-end encrypted. [NOTICED][INFO]
-Msg#1006🔒: Me (Contact#Contact#Self): Member bob@example.net added. [INFO] √
 Msg#1008🔒: Me (Contact#Contact#Self): hi  √
+Msg#1006🔒: Me (Contact#Contact#Self): Member bob@example.net added. [INFO] √
 Msg#1009🔒: Me (Contact#Contact#Self): You removed member bob@example.net. [INFO] √
 --------------------------------------------------------------------------------