feat: perform background fetch from all transports

deltachat-ffi/deltachat.h

@@ -3187,19 +3187,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 +6365,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

@@ -4771,12 +4771,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/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

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-rpc-client/src/deltachat_rpc_client/account.py

@@ -150,9 +150,10 @@ def list_transports(self):
         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 @@
 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 @@ def connect(self):
         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,24 @@
 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, timeout=60):
+    deadline = time.time() + timeout
+    while not imap.get_all_messages():
+        assert time.time() < deadline, f"no message arrived in {imap.addr}"
+        time.sleep(1)
+
+
 def test_add_second_address(acf) -> None:
     account = acf.new_configured_account()
     assert len(account.list_transports()) == 1
@@ -251,11 +270,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 +305,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 +324,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 +355,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(30)
+    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(30)
+    assert len(messages_with_text(alice_chat, "hello")) == 1

src/accounts.rs

@@ -442,7 +442,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 +484,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`.
@@ -1224,6 +1231,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/constants.rs

@@ -171,9 +171,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.fetch_from_all_transports(self).await?;
         }
 
         info!(
             self,
-            "background_fetch done for {address} took {:?}.",
+            "background_fetch done, took {:?}.",
             time_elapsed(&time_start),
         );