feat: perform background fetch from all transports

With I/O stopped, `background_fetch()` connected only to the transport of `configured_addr`
and we now instead fan out to all transports in a controlled loop.
If a first transport finished fetching new messages
cancel all other attempts and return.

This is meant to address the problem that amzd described
where a profile with one functioning and one hanging transport,
shows the first notification, then hangs 15 seconds waiting for the hanging transport.
meanwhile a second NSE arrives and dies, and the second message is not notified
or only generically.

Also drop the quota check from this background fetch path:
its result is in-memory only, discarded when the iOS notification service exits,
and the regular scheduler fetching refreshes it every 60s anyway.
Moreover, quota errors/running full is pretty rare
since relays generally automatically stay under quota these days.
It's another round trip for each transport of each profile and simply not necessary.

Also adds previously missing online tests.

deltachat-ffi/deltachat.h

@@ -3188,7 +3188,8 @@ 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.
  *

deltachat-jsonrpc/src/api.rs

@@ -2038,6 +2038,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

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/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)
+    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)
-    qr = acf.get_account_qr()
-
-    alice.add_transport_from_qr(qr)
-    alice.bring_online()
-
-    bob_chat = bob.create_chat(alice)
-    alice.create_chat(bob)
 
     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

src/constants.rs

@@ -172,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
+    /// If I/O is stopped, fetches over a dedicated connection per transport
-    /// and fetches from Inbox and DeltaChat folders.
+    /// 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
+            self.scheduler.background_fetch_any(self).await?;
-            // 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:#}.");
-            }
         }
 
         info!(
             self,
-            "background_fetch done for {address} took {:?}.",
+            "background_fetch done, took {:?}.",
             time_elapsed(&time_start),
         );
 

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 let Ok(true) = fetched {
+                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)]