fix: don't ignore join_next() errors in background_fetch_any()

join_next() failing means something went really wrong, e.g. panic inside a task.
We also don't cancel the tasks other than by dropping JoinSet,
so join_next() should never return an error.

Cargo.lock

@@ -1327,7 +1327,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat"
-version = "2.60.0"
+version = "2.61.0-dev"
 dependencies = [
  "anyhow",
  "astral-tokio-tar",
@@ -1435,7 +1435,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-jsonrpc"
-version = "2.60.0"
+version = "2.61.0-dev"
 dependencies = [
  "anyhow",
  "async-channel 2.5.0",
@@ -1456,14 +1456,14 @@ dependencies = [
 
 [[package]]
 name = "deltachat-jsonrpc-bindings"
-version = "2.60.0"
+version = "2.61.0-dev"
 dependencies = [
  "deltachat-jsonrpc",
 ]
 
 [[package]]
 name = "deltachat-repl"
-version = "2.60.0"
+version = "2.61.0-dev"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -1479,7 +1479,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-rpc-server"
-version = "2.60.0"
+version = "2.61.0-dev"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -1508,7 +1508,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat_ffi"
-version = "2.60.0"
+version = "2.61.0-dev"
 dependencies = [
  "anyhow",
  "deltachat",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat"
-version = "2.60.0"
+version = "2.61.0-dev"
 edition = "2024"
 license = "MPL-2.0"
 rust-version = "1.89"

deltachat-ffi/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat_ffi"
-version = "2.60.0"
+version = "2.61.0-dev"
 description = "Deltachat FFI"
 edition = "2024"
 readme = "README.md"

deltachat-ffi/deltachat.h

@@ -526,9 +526,10 @@ int             dc_set_config                (dc_context_t* context, const char*
  *
  * - `sys.version` = get the version string e.g. as `1.2.3` or as `1.2.3special4`.
  * - `sys.msgsize_max_recommended` = maximal recommended attachment size in bytes.
- *                    All possible overheads are already subtracted and this value can be used e.g. for direct comparison
- *                    with the size of a file the user wants to attach. If an attachment is larger than this value,
- *                    an error (no warning as it should be shown to the user) is logged but the attachment is sent anyway.
+ *                    All possible overheads are already subtracted and this value can be used
+ *                    e.g. for direct comparison with the size of a file the user wants to attach.
+ *                    If an attachment is larger than this value, the message is sent anyway,
+ *                    but email servers are likely to reject the message when receiving it or before trying to send.
  * - `sys.config_keys` = get a space-separated list of all config-keys available.
  *                    The config-keys are the keys that can be passed to the parameter `key` of this function.
  *
@@ -3187,19 +3188,22 @@ void           dc_accounts_maybe_network_lost    (dc_accounts_t* accounts);
 
 /**
  * Perform a background fetch for all accounts in parallel with a timeout.
- * Pauses the scheduler, fetches messages from imap and then resumes the scheduler.
+ * Pauses the scheduler, fetches from all transports at once and then resumes the scheduler.
+ * The fetch for an account ends as soon as one of its transports received messages.
  *
  * dc_accounts_background_fetch() was created for the iOS Background fetch.
  *
- * The `DC_EVENT_ACCOUNTS_BACKGROUND_FETCH_DONE` event is emitted at the end
- * even in case of timeout, unless the function fails and returns 0.
+ * The `DC_EVENT_ACCOUNTS_BACKGROUND_FETCH_DONE` event is emitted at the end,
+ * also on timeout, when another background fetch is already running
+ * and when the call is ignored because the timeout is too small,
+ * so it is safe to wait for the event whenever `accounts` is not NULL.
  * Process all events until you get this one and you can safely return to the background
  * without forgetting to create notifications caused by timing race conditions.
  *
  * @memberof dc_accounts_t
  * @param accounts The account manager as created by dc_accounts_new().
  * @param timeout The timeout in seconds
- * @return Return 1 if DC_EVENT_ACCOUNTS_BACKGROUND_FETCH_DONE was emitted and 0 otherwise.
+ * @return Return 0 if the call was ignored because `accounts` is NULL or the timeout is too small, 1 otherwise.
  */
 int            dc_accounts_background_fetch    (dc_accounts_t* accounts, uint64_t timeout);
 
@@ -6362,11 +6366,14 @@ void dc_event_unref(dc_event_t* event);
 #define DC_EVENT_WEBXDC_REALTIME_ADVERTISEMENT    2151
 
 /**
- * Tells that the Background fetch was completed (or timed out).
+ * Tells that a call to dc_accounts_background_fetch() is done:
+ * the fetch completed, timed out, was stopped or was not started.
+ *
+ * For the call that started the fetch, this event acts as a marker:
+ * when you reach it, all events emitted during the fetch were processed.
+ * A call made while another background fetch is running gets the event immediately,
+ * and the running fetch keeps emitting events until its own marker.
  *
- * This event acts as a marker, when you reach this event you can be sure
- * that all events emitted during the background fetch were processed.
- * 
  * This event is only emitted by the account manager
  */
 

deltachat-ffi/src/lib.rs

@@ -4750,12 +4750,17 @@ pub unsafe extern "C" fn dc_accounts_background_fetch(
     accounts: *const dc_accounts_t,
     timeout_in_seconds: u64,
 ) -> libc::c_int {
-    if accounts.is_null() || timeout_in_seconds <= 2 {
+    if accounts.is_null() {
         eprintln!("ignoring careless call to dc_accounts_background_fetch()");
         return 0;
     }
 
     let accounts = unsafe { &*accounts };
+    if timeout_in_seconds <= 2 {
+        eprintln!("ignoring careless call to dc_accounts_background_fetch(): timeout too small");
+        block_on(accounts.read()).emit_event(EventType::AccountsBackgroundFetchDone);
+        return 0;
+    }
     let background_fetch_future = {
         let lock = block_on(accounts.read());
         lock.background_fetch(Duration::from_secs(timeout_in_seconds))

deltachat-jsonrpc-bindings/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-jsonrpc-bindings"
-version = "2.60.0"
+version = "2.61.0-dev"
 description = "Autogenerate DeltaChat JSON-RPC API bindings at build time"
 edition = "2024"
 license = "MPL-2.0"

deltachat-jsonrpc-bindings/typescript/package.json

@@ -54,5 +54,5 @@
   },
   "type": "module",
   "types": "dist/deltachat.d.ts",
-  "version": "2.60.0"
+  "version": "2.61.0-dev"
 }

deltachat-jsonrpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-jsonrpc"
-version = "2.60.0"
+version = "2.61.0-dev"
 description = "DeltaChat JSON-RPC API"
 edition = "2024"
 license = "MPL-2.0"

deltachat-jsonrpc/src/api.rs

@@ -278,7 +278,8 @@ impl CommandApi {
 
     /// Performs a background fetch for all accounts in parallel with a timeout.
     ///
-    /// The `AccountsBackgroundFetchDone` event is emitted at the end even in case of timeout.
+    /// The `AccountsBackgroundFetchDone` event is emitted at the end even in case of timeout,
+    /// and immediately if another background fetch is already running.
     /// Process all events until you get this one and you can safely return to the background
     /// without forgetting to create notifications caused by timing race conditions.
     async fn background_fetch(&self, timeout_in_seconds: f64) -> Result<()> {
@@ -2038,6 +2039,14 @@ impl CommandApi {
         Ok(())
     }
 
+    /// Waits until all transports are idle or failed and no background work is left.
+    /// Never returns unless I/O is started. Must ONLY be used by tests.
+    async fn wait_for_all_work_done(&self, account_id: u32) -> Result<()> {
+        let ctx = self.get_context(account_id).await?;
+        ctx.wait_for_all_work_done().await;
+        Ok(())
+    }
+
     /// Get the current connectivity, i.e. whether the device is connected to the IMAP server.
     /// One of:
     /// - DC_CONNECTIVITY_NOT_CONNECTED (1000): Show e.g. the string "Not connected" or a red dot
@@ -2775,6 +2784,15 @@ impl CommandApi {
                 .map(JsonrpcAppSource::from_core_type),
         )
     }
+
+    /// Returns true if all accounts have empty outgoing message queue.
+    ///
+    /// This API is intended to be used by UIs
+    /// to request that operating system does not put the application in background
+    /// while there are still outgoing messages that are not sent out.
+    async fn is_sending_finished(&self) -> Result<bool> {
+        self.accounts.read().await.is_sending_finished().await
+    }
 }
 
 // Helper functions (to prevent code duplication)

deltachat-jsonrpc/src/api/types/events.rs

@@ -394,11 +394,15 @@ pub enum EventType {
         msg_id: u32,
     },
 
-    /// Tells that the Background fetch was completed (or timed out).
-    /// This event acts as a marker, when you reach this event you can be sure
-    /// that all events emitted during the background fetch were processed.
+    /// Tells that a background fetch call is done:
+    /// the fetch completed, timed out, was stopped or was not started.
     ///
-    /// This event is only emitted by the account manager
+    /// For the call that started the fetch, this event acts as a marker:
+    /// all events emitted during the fetch were processed once it is reached.
+    /// A call made while another background fetch is running gets the event immediately,
+    /// and the running fetch keeps emitting events until its own marker.
+    ///
+    /// This event is only emitted by the account manager.
     AccountsBackgroundFetchDone,
     /// Inform that set of chats or the order of the chats in the chatlist has changed.
     ///

deltachat-repl/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-repl"
-version = "2.60.0"
+version = "2.61.0-dev"
 license = "MPL-2.0"
 edition = "2024"
 repository = "https://github.com/chatmail/core"

deltachat-rpc-client/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "deltachat-rpc-client"
-version = "2.60.0"
+version = "2.61.0-dev"
 license = "MPL-2.0"
 description = "Python client for Delta Chat core JSON-RPC interface"
 classifiers = [

deltachat-rpc-client/src/deltachat_rpc_client/account.py

@@ -150,9 +150,10 @@ class Account:
         return transports
 
     def bring_online(self):
-        """Start I/O and wait until IMAP becomes IDLE."""
+        """Start I/O, wait until all transports became IDLE and drop the events seen so far."""
         self.start_io()
-        self.wait_for_event(EventType.IMAP_INBOX_IDLE)
+        self._rpc.wait_for_all_work_done(self.id)
+        self.clear_all_events()
 
     def create_contact(self, obj: Union[int, str, Contact, "Account"], name: Optional[str] = None) -> Contact:
         """Create a new Contact or return an existing one.

deltachat-rpc-client/src/deltachat_rpc_client/deltachat.py

@@ -67,3 +67,7 @@ class DeltaChat:
     def stop_sending_locations(self) -> None:
         """Stop sending locations to all chats."""
         return self.rpc.stop_sending_locations()
+
+    def is_sending_finished(self) -> bool:
+        """Return true if sending queues of all accounts are empty."""
+        return self.rpc.is_sending_finished()

deltachat-rpc-client/tests/conftest.py

@@ -22,8 +22,10 @@ ALL = "1:*"
 class DirectImap:
     """Internal Python-level IMAP handling."""
 
-    def __init__(self, account: Account) -> None:
+    def __init__(self, account: Account, addr=None, password=None) -> None:
         self.account = account
+        self.addr = addr or account.get_config("addr")
+        self.password = password or account.get_config("mail_pw")
         self.logid = account.get_config("displayname") or id(account)
         self._idling = False
         self.connect()
@@ -33,9 +35,9 @@ class DirectImap:
         host = self.account.get_config("configured_mail_server")
         port = 993
 
-        user = self.account.get_config("addr")
+        user = self.addr
         host = user.rsplit("@")[-1]
-        pw = self.account.get_config("mail_pw")
+        pw = self.password
 
         ssl_context = ssl.create_default_context()
         if host.startswith("_"):

deltachat-rpc-client/tests/test_multitransport.py

@@ -1,3 +1,4 @@
+import time
 import urllib.parse
 
 import pytest
@@ -7,6 +8,22 @@ from deltachat_rpc_client.const import ChatType, DownloadState
 from deltachat_rpc_client.rpc import JsonRpcError
 
 
+def alice_with_two_transports_and_bob(acf):
+    alice, bob = acf.get_online_accounts(2)
+    alice.add_transport_from_qr(acf.get_account_qr())
+    alice.bring_online()
+    return alice, alice.create_chat(bob), bob.create_chat(alice)
+
+
+def messages_with_text(chat, text):
+    return [msg for msg in chat.get_messages() if msg.get_snapshot().text == text]
+
+
+def wait_for_imap_message(imap):
+    while not imap.get_all_messages():
+        time.sleep(1)
+
+
 def test_add_second_address(acf) -> None:
     account = acf.new_configured_account()
     assert len(account.list_transports()) == 1
@@ -251,11 +268,10 @@ def test_message_info_imap_urls(acf) -> None:
     alice, bob = acf.get_online_accounts(2)
 
     qr = acf.get_account_qr()
-    for i in range(3):
+    for _ in range(3):
         alice.add_transport_from_qr(qr)
         # Wait for all transports to go IDLE after adding each one.
-        for _ in range(i + 1):
-            alice.bring_online()
+        alice.bring_online()
 
     # Enable multi-device mode so messages are not deleted immediately.
     alice.set_config("bcc_self", "1")
@@ -287,14 +303,7 @@ def test_message_info_imap_urls(acf) -> None:
 
 def test_remove_primary_transport(acf, log) -> None:
     """Test that after removing the primary relay, Alice can still receive messages."""
-    alice, bob = acf.get_online_accounts(2)
-    qr = acf.get_account_qr()
-
-    alice.add_transport_from_qr(qr)
-    alice.bring_online()
-
-    bob_chat = bob.create_chat(alice)
-    alice.create_chat(bob)
+    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)
 
     log.section("Alice sets up second transport")
     [transport1, transport2] = alice.list_transports()
@@ -313,7 +322,7 @@ def test_remove_primary_transport(acf, log) -> None:
     msg2 = alice.wait_for_incoming_msg().get_snapshot()
     assert msg2.text == "Hello again!"
     assert msg2.chat.get_basic_snapshot().chat_type == ChatType.SINGLE
-    assert msg2.chat == alice.create_chat(bob)
+    assert msg2.chat == alice_chat
 
 
 def test_qr_works_after_removing_primary_transport(acf, log) -> None:
@@ -344,3 +353,33 @@ def test_qr_works_after_removing_primary_transport(acf, log) -> None:
     bob.secure_join(chat_qr)
     alice.wait_for_securejoin_inviter_success()
     bob.wait_for_securejoin_joiner_success()
+
+
+def test_background_fetch_from_second_transport(acf, direct_imap, dc):
+    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)
+    [transport1, transport2] = alice.list_transports()
+    assert alice.get_config("configured_addr") == transport1["addr"]
+
+    alice.stop_io()
+    bob_chat.send_text("hello")
+    imap1 = direct_imap(alice, transport1["addr"], transport1["password"])
+    wait_for_imap_message(direct_imap(alice, transport2["addr"], transport2["password"]))
+    wait_for_imap_message(imap1)
+
+    # Leave the message on the second transport only.
+    imap1.delete("1:*")
+
+    dc.background_fetch(300)
+    assert len(messages_with_text(alice_chat, "hello")) == 1
+
+
+def test_background_fetch_no_duplicates(acf, direct_imap, dc):
+    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)
+
+    alice.stop_io()
+    bob_chat.send_text("hello")
+    for transport in alice.list_transports():
+        wait_for_imap_message(direct_imap(alice, transport["addr"], transport["password"]))
+
+    dc.background_fetch(300)
+    assert len(messages_with_text(alice_chat, "hello")) == 1

deltachat-rpc-client/tests/test_something.py

@@ -1435,3 +1435,35 @@ def test_large_message(acf, rpcdata) -> None:
     assert msg.id == msgs_changed_event.msg_id
     snapshot = msg.get_snapshot()
     assert snapshot.text == "Hello World, this message is bigger than 5 bytes"
+
+
+def test_is_sending_finished(dc, acf) -> None:
+    alice, bob = acf.get_online_accounts(2)
+
+    alice_chat_bob = alice.create_chat(bob)
+    bob_chat_alice = bob.create_chat(alice)
+
+    assert dc.is_sending_finished()
+
+    alice_chat_bob.send_text("Hello!")
+    alice.wait_for_event(EventType.SMTP_MESSAGE_SENT)
+
+    assert dc.is_sending_finished()
+
+    alice.stop_io()
+    bob.stop_io()
+
+    bob_chat_alice.send_text("Hello back!")
+    alice_chat_bob.send_text("Hello again!")
+
+    assert not dc.is_sending_finished()
+
+    alice.start_io()
+    alice.wait_for_event(EventType.SMTP_MESSAGE_SENT)
+
+    assert not dc.is_sending_finished()
+
+    bob.start_io()
+    bob.wait_for_event(EventType.SMTP_MESSAGE_SENT)
+
+    assert dc.is_sending_finished()

deltachat-rpc-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-rpc-server"
-version = "2.60.0"
+version = "2.61.0-dev"
 description = "DeltaChat JSON-RPC server"
 edition = "2024"
 readme = "README.md"

deltachat-rpc-server/npm-package/package.json

@@ -15,5 +15,5 @@
   },
   "type": "module",
   "types": "index.d.ts",
-  "version": "2.60.0"
+  "version": "2.61.0-dev"
 }

python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "deltachat"
-version = "2.60.0"
+version = "2.61.0-dev"
 license = "MPL-2.0"
 description = "Python bindings for the Delta Chat Core library using CFFI against the Rust-implemented libdeltachat"
 readme = "README.rst"

src/accounts.rs

@@ -26,6 +26,7 @@ use crate::events::{Event, EventEmitter, EventType, Events};
 use crate::location;
 use crate::log::warn;
 use crate::push::PushSubscriber;
+use crate::smtp;
 use crate::stock_str::StockStrings;
 
 /// Account manager, that can handle multiple accounts in a single place.
@@ -442,7 +443,12 @@ impl Accounts {
         interrupt_receiver: Option<Receiver<()>>,
     ) {
         let Some(interrupt_receiver) = interrupt_receiver else {
-            // Nothing to do if we got no interrupt receiver.
+            // Another background fetch is already running.
+            // Emit the event anyway so that a caller waiting for it does not hang.
+            events.emit(Event {
+                id: 0,
+                typ: EventType::AccountsBackgroundFetchDone,
+            });
             return;
         };
         if let Err(_err) = tokio::time::timeout(
@@ -479,6 +485,8 @@ impl Accounts {
     /// The `AccountsBackgroundFetchDone` event is emitted at the end,
     /// process all events until you get this one and you can safely return to the background
     /// without forgetting to create notifications caused by timing race conditions.
+    /// If another background fetch is already running,
+    /// nothing is fetched and the event is emitted immediately.
     ///
     /// Returns a future that resolves when background fetch is done,
     /// but does not capture `&self`.
@@ -509,6 +517,20 @@ impl Accounts {
         )
     }
 
+    /// Returns true if there are no pending messages for sending.
+    ///
+    /// This is intended to be used by UIs to request not moving the app to background
+    /// when there are messages left in the queue.
+    pub async fn is_sending_finished(&self) -> Result<bool> {
+        let accounts: Vec<Context> = self.accounts.values().cloned().collect();
+        for account in accounts {
+            if !smtp::is_queue_empty(&account).await? {
+                return Ok(false);
+            }
+        }
+        Ok(true)
+    }
+
     /// Interrupts ongoing background_fetch() call,
     /// making it return early.
     ///
@@ -1224,6 +1246,29 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_background_fetch_emits_done_when_already_running() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let writable = true;
+        let accounts = Accounts::new(dir.path().join("accounts"), writable).await?;
+        let event_emitter = accounts.get_event_emitter();
+
+        let timeout = std::time::Duration::from_secs(3);
+        let first = accounts.background_fetch(timeout);
+        let second = accounts.background_fetch(timeout);
+        tokio::join!(first, second);
+
+        let mut done = 0;
+        while let Ok(event) = event_emitter.try_recv() {
+            if matches!(event.typ, EventType::AccountsBackgroundFetchDone) {
+                done += 1;
+            }
+        }
+        assert_eq!(done, 2);
+
+        Ok(())
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_encrypted_account() -> Result<()> {
         let dir = tempfile::tempdir().context("failed to create tempdir")?;

src/blob.rs

@@ -386,7 +386,7 @@ impl<'a> BlobObject<'a> {
             let exceeds_wh = img.width() > max_wh || img.height() > max_wh;
             let exceeds_max_bytes = nr_bytes > max_bytes as u64;
 
-            let jpeg_quality = 75;
+            let jpeg_quality = 75; // 70-80 is the sweet spot of quality vs. bytes/pixel. if one wants to spend more bytes in quality, better increase resolution
             let ofmt = match fmt {
                 ImageFormat::Png if !exceeds_max_bytes => ImageOutputFormat::Png,
                 ImageFormat::Jpeg => {

src/blob/blob_tests.rs

@@ -508,17 +508,17 @@ async fn test_recode_image_rgba_png_to_jpeg() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_recode_image_huge_jpg() {
-    let bytes = include_bytes!("../../test-data/image/screenshot.jpg");
+    let bytes = include_bytes!("../../test-data/image/screenshot-2560x1440.jpg");
     SendImageCheckMediaquality {
         viewtype: Viewtype::Image,
         media_quality_config: "0",
         bytes,
         extension: "jpg",
         has_exif: true,
-        original_width: 1920,
-        original_height: 1080,
-        compressed_width: 1704,
-        compressed_height: 959,
+        original_width: 2560,
+        original_height: 1440,
+        compressed_width: 2344,
+        compressed_height: 1319,
         ..Default::default()
     }
     .test()

src/constants.rs

@@ -148,8 +148,9 @@ pub const DC_DESIRED_TEXT_LINE_LEN: usize = 100;
 /// `char`s), not Unicode Grapheme Clusters.
 pub const DC_DESIRED_TEXT_LEN: usize = DC_DESIRED_TEXT_LINE_LEN * DC_DESIRED_TEXT_LINES;
 
-// max. weight of images to send w/o recoding
-pub const BALANCED_IMAGE_BYTES: usize = 500_000;
+/// max. weight of images to send w/o recoding.
+// this is an estimation to the size we get when recoding high detail images.
+pub const BALANCED_IMAGE_BYTES: usize = 940_000;
 pub const WORSE_IMAGE_BYTES: usize = 130_000;
 
 // max. width/height and bytes of an avatar
@@ -159,7 +160,7 @@ pub(crate) const WORSE_AVATAR_SIZE: u32 = 256;
 pub(crate) const WORSE_AVATAR_BYTES: usize = 20_000; // this also fits to Outlook servers don't allowing headers larger than 32k.
 
 // max. width/height of images scaled down because of being too huge
-pub const BALANCED_IMAGE_SIZE: u32 = 1280;
+pub const BALANCED_IMAGE_SIZE: u32 = 1760;
 pub const WORSE_IMAGE_SIZE: u32 = 640;
 
 /// Limit for received images size. Bigger images become `Viewtype::File` to avoid excessive memory
@@ -171,9 +172,6 @@ pub const MAX_RCVD_IMAGE_PIXELS: u32 = 50_000_000;
 // Relays typically advertise their limit via IMAP METADATA.
 pub(crate) const DEFAULT_MAX_SMTP_RCPT_TO: u32 = 50;
 
-/// 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
-
 /// How far in the future the sender timestamp of a message is allowed to be, in seconds. Also used
 /// in the group membership consistency algo to reject outdated membership changes.
 pub(crate) const TIMESTAMP_SENT_TOLERANCE: i64 = 60;

src/context.rs

@@ -16,11 +16,11 @@ 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::{self, DC_VERSION_STR};
 use crate::contact::{Contact, ContactId};
 use crate::debug_logging::DebugLogging;
 use crate::events::{Event, EventEmitter, EventType, Events};
-use crate::imap::{Imap, ServerMetadata};
+use crate::imap::ServerMetadata;
 use crate::log::warn;
 use crate::logged_debug_assert;
 use crate::message::{self, MessageState, MsgId};
@@ -599,56 +599,29 @@ impl Context {
         Ok(constants::DEFAULT_MAX_SMTP_RCPT_TO)
     }
 
-    /// Does a single round of fetching from IMAP and returns.
+    /// Does a single round of fetching messages from all transports and returns.
     ///
     /// Can be used even if I/O is currently stopped.
-    /// If I/O is currently stopped, starts a new IMAP connection
-    /// and fetches from Inbox and DeltaChat folders.
+    /// If I/O is stopped, fetches over a dedicated connection per transport
+    /// and returns as soon as one of them fetched messages.
     pub async fn background_fetch(&self) -> Result<()> {
         if !(self.is_configured().await?) {
             return Ok(());
         }
 
-        let address = self.get_primary_self_addr().await?;
         let time_start = tools::Time::now();
-        info!(self, "background_fetch started fetching {address}.");
+        info!(self, "background_fetch started.");
 
         if self.scheduler.is_running().await {
             self.scheduler.maybe_network().await;
             self.wait_for_all_work_done().await;
         } else {
-            // Pause the scheduler to ensure another connection does not start
-            // while we are fetching on a dedicated connection.
-            let _pause_guard = self.scheduler.pause(self).await?;
-
-            // Start a new dedicated connection.
-            let mut connection = Imap::new_configured(self, channel::bounded(1).1).await?;
-            let mut session = connection.prepare(self).await?;
-
-            // Fetch IMAP folders.
-            let folder = connection.folder.clone();
-            connection
-                .fetch_move_delete(self, &mut session, &folder)
-                .await?;
-
-            // Update quota (to send warning if full) - but only check it once in a while.
-            // note: For now this only checks quota of primary transport,
-            // because background check only checks primary transport at the moment
-            if self
-                .quota_needs_update(
-                    session.transport_id(),
-                    DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT,
-                )
-                .await
-                && let Err(err) = self.update_recent_quota(&mut session, &folder).await
-            {
-                warn!(self, "Failed to update quota: {err:#}.");
-            }
+            self.scheduler.background_fetch_any(self).await?;
         }
 
         info!(
             self,
-            "background_fetch done for {address} took {:?}.",
+            "background_fetch done, took {:?}.",
             time_elapsed(&time_start),
         );
 

src/events/payload.rs

@@ -357,11 +357,15 @@ pub enum EventType {
         msg_id: MsgId,
     },
 
-    /// Tells that the Background fetch was completed (or timed out).
-    /// This event acts as a marker, when you reach this event you can be sure
-    /// that all events emitted during the background fetch were processed.
+    /// Tells that a background fetch call is done:
+    /// the fetch completed, timed out, was stopped or was not started.
     ///
-    /// This event is only emitted by the account manager
+    /// For the call that started the fetch, this event acts as a marker:
+    /// all events emitted during the fetch were processed once it is reached.
+    /// A call made while another background fetch is running gets the event immediately,
+    /// and the running fetch keeps emitting events until its own marker.
+    ///
+    /// This event is only emitted by the account manager.
     AccountsBackgroundFetchDone,
     /// Inform that set of chats or the order of the chats in the chatlist has changed.
     ///

src/imap.rs

@@ -19,6 +19,7 @@ use async_imap::types::{Fetch, Flag, UnsolicitedResponse};
 use futures::{FutureExt as _, TryStreamExt};
 use futures_lite::FutureExt;
 use ratelimit::Ratelimit;
+use tokio_util::sync::CancellationToken;
 use url::Url;
 
 use crate::chat::{self, add_device_msg};
@@ -104,6 +105,10 @@ pub(crate) struct Imap {
 
     /// IMAP UID resync request receiver.
     pub(crate) resync_request_receiver: async_channel::Receiver<()>,
+
+    /// The background fetch is cancelled once messages are fetched from one of the transports,
+    /// so that the other transports fetch nothing.
+    pub(crate) background_fetch_stop_token: Option<CancellationToken>,
 }
 
 #[derive(Debug, Default)]
@@ -237,21 +242,10 @@ impl Imap {
             ratelimit: Ratelimit::new(Duration::new(120, 0), 2.0),
             resync_request_sender,
             resync_request_receiver,
+            background_fetch_stop_token: None,
         })
     }
 
-    /// Creates new disconnected IMAP client using configured parameters.
-    pub async fn new_configured(
-        context: &Context,
-        idle_interrupt_receiver: Receiver<()>,
-    ) -> Result<Self> {
-        let (transport_id, param) = ConfiguredLoginParam::load(context)
-            .await?
-            .context("Not configured")?;
-        let imap = Self::new(context, transport_id, param, idle_interrupt_receiver).await?;
-        Ok(imap)
-    }
-
     /// Returns transport ID of the IMAP client.
     pub fn transport_id(&self) -> u32 {
         self.transport_id
@@ -428,12 +422,14 @@ impl Imap {
     ///
     /// Prefetches headers and downloads new message from the folder, moves messages away from the
     /// folder and deletes messages in the folder.
+    ///
+    /// Returns true if at least one message was fetched.
     pub async fn fetch_move_delete(
         &mut self,
         context: &Context,
         session: &mut Session,
         watch_folder: &str,
-    ) -> Result<()> {
+    ) -> Result<bool> {
         ensure_and_debug_assert!(!watch_folder.is_empty(), "Watched folder cannot be empty");
         if !context.sql.is_open().await {
             // probably shutdown
@@ -463,7 +459,7 @@ impl Imap {
             .await
             .context("move_delete_messages")?;
 
-        Ok(())
+        Ok(msgs_fetched)
     }
 
     /// Fetches new messages.
@@ -532,6 +528,19 @@ impl Imap {
             .context("prefetch")?;
         let read_cnt = msgs.len();
         let _fetch_msgs_lock_guard = context.fetch_msgs_mutex.lock().await;
+        if let Some(stop_token) = &self.background_fetch_stop_token {
+            if stop_token.is_cancelled() {
+                // This also stops the transport that cancelled the token,
+                // so one background fetch receives at most `uids_to_prefetch` messages.
+                return Ok((0, false));
+            }
+            if read_cnt > 0 {
+                // Cancel the background fetch on the other transports,
+                // so that `background_fetch_any()` can return as soon as messages are received
+                // and the UI can show a notification.
+                stop_token.cancel();
+            }
+        }
 
         let mut uids_fetch: Vec<u32> = Vec::new();
         let mut available_post_msgs: Vec<String> = Vec::new();

src/scheduler.rs

@@ -9,7 +9,7 @@ use async_channel::{self as channel, Receiver, Sender};
 use futures::future::try_join_all;
 use futures_lite::FutureExt;
 use tokio::sync::{RwLock, oneshot};
-use tokio::task;
+use tokio::task::{self, JoinSet};
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 
@@ -282,6 +282,60 @@ impl SchedulerState {
             scheduler.interrupt_recently_seen(contact_id, timestamp);
         }
     }
+
+    /// Fetches from all transports at once, each on a dedicated connection,
+    /// with I/O paused so that the scheduler does not connect as well.
+    ///
+    /// Returns as soon as one transport fetched messages:
+    /// the others then fetch nothing more and are dropped,
+    /// so that a caller woken up by a push notification
+    /// does not wait for a transport that may never answer.
+    pub(crate) async fn background_fetch_any(&self, context: &Context) -> Result<()> {
+        let _pause_guard = self.pause(context).await?;
+
+        let stop_token = CancellationToken::new();
+        let mut set = JoinSet::new();
+        for (transport_id, param) in ConfiguredLoginParam::load_all(context).await? {
+            let context = context.clone();
+            let stop_token = stop_token.clone();
+            set.spawn(async move {
+                match background_fetch_from_transport(&context, transport_id, param, stop_token)
+                    .await
+                {
+                    Ok(fetched) => fetched,
+                    Err(err) => {
+                        warn!(context, "Transport {transport_id}: fetch failed: {err:#}.");
+                        false
+                    }
+                }
+            });
+        }
+
+        while let Some(fetched) = set.join_next().await {
+            if fetched.context("background_fetch_any: Failed to join a task")? {
+                break;
+            }
+        }
+        Ok(())
+    }
+}
+
+async fn background_fetch_from_transport(
+    context: &Context,
+    transport_id: u32,
+    param: ConfiguredLoginParam,
+    stop_token: CancellationToken,
+) -> Result<bool> {
+    // A single fetch has nothing to interrupt.
+    let (_, idle_interrupt_receiver) = channel::bounded(1);
+    let mut connection = Imap::new(context, transport_id, param, idle_interrupt_receiver).await?;
+    connection.background_fetch_stop_token = Some(stop_token);
+    let mut session = connection.prepare(context).await?;
+
+    let folder = connection.folder.clone();
+    connection
+        .fetch_move_delete(context, &mut session, &folder)
+        .await
 }
 
 #[derive(Debug, Default)]

src/smtp.rs

@@ -751,3 +751,9 @@ pub(crate) async fn add_self_recipients(
 
     Ok(())
 }
+
+/// Returns true if SMTP queue is empty.
+pub(crate) async fn is_queue_empty(context: &Context) -> Result<bool> {
+    let sending_finished = !context.sql.exists("SELECT COUNT(*) FROM smtp", ()).await?;
+    Ok(sending_finished)
+}

src/tests/pre_messages/additional_text.rs

@@ -34,7 +34,7 @@ async fn test_additional_text_on_different_viewtypes() -> Result<()> {
     let (pre_message, _, _) = send_large_image_message(alice, a_group_id).await?;
     let msg = bob.recv_msg(&pre_message).await;
     assert_eq!(msg.text, "test".to_owned());
-    assert_eq!(msg.get_text(), "test [Image – 228.45 KiB]".to_owned());
+    assert_eq!(msg.get_text(), "test [Image – 275.46 KiB]".to_owned());
 
     Ok(())
 }

src/tests/pre_messages/receiving.rs

@@ -501,9 +501,9 @@ async fn test_receive_pre_message_image() -> Result<()> {
     // test that metadata is correctly returned by methods
     assert_eq!(msg.get_post_message_viewtype(), Some(Viewtype::Image));
     // recoded image dimensions
-    assert_eq!(msg.get_filebytes(bob).await?, Some(233935));
-    assert_eq!(msg.get_height(), 1704);
-    assert_eq!(msg.get_width(), 959);
+    assert_eq!(msg.get_filebytes(bob).await?, Some(282068));
+    assert_eq!(msg.get_height(), 1920);
+    assert_eq!(msg.get_width(), 1080);
 
     Ok(())
 }