From 0ea4b08f630a7638871745d610302e17dad4cfb4 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Thu, 3 Sep 2026 00:23:00 +0000 Subject: [PATCH 1/2] 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 | 3 +- deltachat-jsonrpc/src/api.rs | 8 +++ .../src/deltachat_rpc_client/account.py | 5 +- deltachat-rpc-client/tests/conftest.py | 8 ++- .../tests/test_multitransport.py | 65 +++++++++++++++---- src/constants.rs | 3 - src/context.rs | 43 +++--------- src/imap.rs | 18 ++--- src/scheduler.rs | 48 ++++++++++++++ 9 files changed, 131 insertions(+), 70 deletions(-) diff --git a/deltachat-ffi/deltachat.h b/deltachat-ffi/deltachat.h index 52fcb355d6..4d39ca2fad 100644 --- a/deltachat-ffi/deltachat.h +++ b/deltachat-ffi/deltachat.h @@ -3187,7 +3187,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. * diff --git a/deltachat-jsonrpc/src/api.rs b/deltachat-jsonrpc/src/api.rs index f2432c1be6..aea1559301 100644 --- a/deltachat-jsonrpc/src/api.rs +++ b/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 diff --git a/deltachat-rpc-client/src/deltachat_rpc_client/account.py b/deltachat-rpc-client/src/deltachat_rpc_client/account.py index 9a91bafe27..06cd3d3b5a 100644 --- a/deltachat-rpc-client/src/deltachat_rpc_client/account.py +++ b/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. diff --git a/deltachat-rpc-client/tests/conftest.py b/deltachat-rpc-client/tests/conftest.py index b8f1600457..9bf4a1f75e 100644 --- a/deltachat-rpc-client/tests/conftest.py +++ b/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("_"): diff --git a/deltachat-rpc-client/tests/test_multitransport.py b/deltachat-rpc-client/tests/test_multitransport.py index 2f6fd35c27..13cbe108fd 100644 --- a/deltachat-rpc-client/tests/test_multitransport.py +++ b/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 diff --git a/src/constants.rs b/src/constants.rs index 9ded94cfae..6ee2ed1819 100644 --- a/src/constants.rs +++ b/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; diff --git a/src/context.rs b/src/context.rs index 418d81f916..d2bcecf503 100644 --- a/src/context.rs +++ b/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), ); diff --git a/src/imap.rs b/src/imap.rs index fd4be4f31f..533531e986 100644 --- a/src/imap.rs +++ b/src/imap.rs @@ -240,18 +240,6 @@ impl Imap { }) } - /// Creates new disconnected IMAP client using configured parameters. - pub async fn new_configured( - context: &Context, - idle_interrupt_receiver: Receiver<()>, - ) -> Result { - 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 +416,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 { ensure_and_debug_assert!(!watch_folder.is_empty(), "Watched folder cannot be empty"); if !context.sql.is_open().await { // probably shutdown @@ -463,7 +453,7 @@ impl Imap { .await .context("move_delete_messages")?; - Ok(()) + Ok(msgs_fetched) } /// Fetches new messages. diff --git a/src/scheduler.rs b/src/scheduler.rs index fbdff516fb..3483b11cfb 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -7,6 +7,7 @@ use std::sync::atomic::Ordering; use anyhow::{Context as _, Error, Result, bail}; use async_channel::{self as channel, Receiver, Sender}; use futures::future::try_join_all; +use futures::stream::{FuturesUnordered, StreamExt}; use futures_lite::FutureExt; use tokio::sync::{RwLock, oneshot}; use tokio::task; @@ -282,6 +283,53 @@ 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 and drops the fetches + /// still in flight, so that a caller woken up by a push notification + /// does not wait for a transport that may never answer. + pub(crate) async fn fetch_from_all_transports(&self, context: &Context) -> Result<()> { + let _pause_guard = self.pause(context).await?; + + let mut futures: FuturesUnordered<_> = ConfiguredLoginParam::load_all(context) + .await? + .into_iter() + .map(|(transport_id, param)| async move { + match fetch_from_transport(context, transport_id, param).await { + Ok(fetched) => fetched, + Err(err) => { + warn!(context, "Transport {transport_id}: fetch failed: {err:#}."); + false + } + } + }) + .collect(); + + while let Some(fetched) = futures.next().await { + if fetched { + break; + } + } + Ok(()) + } +} + +async fn fetch_from_transport( + context: &Context, + transport_id: u32, + param: ConfiguredLoginParam, +) -> Result { + // 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?; + let mut session = connection.prepare(context).await?; + + let folder = connection.folder.clone(); + connection + .fetch_move_delete(context, &mut session, &folder) + .await } #[derive(Debug, Default)] From da42f2734adfb782d4744920593ddde316dcc292 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Thu, 3 Sep 2026 00:24:00 +0000 Subject: [PATCH 2/2] fix: always emit AccountsBackgroundFetchDone A second background_fetch() while one is already running returned without emitting the event, and the FFI returned 1 for it, so a UI waiting for an event hangs dc_get_next_event(). Emit the event in any case, so waiting for it is safe. --- deltachat-ffi/deltachat.h | 19 +++++++++----- deltachat-ffi/src/lib.rs | 7 ++++- deltachat-jsonrpc/src/api.rs | 3 ++- deltachat-jsonrpc/src/api/types/events.rs | 12 ++++++--- src/accounts.rs | 32 ++++++++++++++++++++++- src/events/payload.rs | 12 ++++++--- 6 files changed, 67 insertions(+), 18 deletions(-) diff --git a/deltachat-ffi/deltachat.h b/deltachat-ffi/deltachat.h index 4d39ca2fad..69f9646bb1 100644 --- a/deltachat-ffi/deltachat.h +++ b/deltachat-ffi/deltachat.h @@ -3192,15 +3192,17 @@ void dc_accounts_maybe_network_lost (dc_accounts_t* accounts); * * 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); @@ -6363,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 */ diff --git a/deltachat-ffi/src/lib.rs b/deltachat-ffi/src/lib.rs index 66e05403ad..bb28fe05ae 100644 --- a/deltachat-ffi/src/lib.rs +++ b/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)) diff --git a/deltachat-jsonrpc/src/api.rs b/deltachat-jsonrpc/src/api.rs index aea1559301..a68e314262 100644 --- a/deltachat-jsonrpc/src/api.rs +++ b/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<()> { diff --git a/deltachat-jsonrpc/src/api/types/events.rs b/deltachat-jsonrpc/src/api/types/events.rs index de193db2b2..0bfa3138a5 100644 --- a/deltachat-jsonrpc/src/api/types/events.rs +++ b/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. /// diff --git a/src/accounts.rs b/src/accounts.rs index 1efaa63d10..9420dc1e9e 100644 --- a/src/accounts.rs +++ b/src/accounts.rs @@ -442,7 +442,12 @@ impl Accounts { interrupt_receiver: Option>, ) { 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")?; diff --git a/src/events/payload.rs b/src/events/payload.rs index 4e07d6709c..383714a77c 100644 --- a/src/events/payload.rs +++ b/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. ///