diff --git a/deltachat-ffi/deltachat.h b/deltachat-ffi/deltachat.h index 81939c63b..80beedfb8 100644 --- a/deltachat-ffi/deltachat.h +++ b/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. * diff --git a/deltachat-jsonrpc/src/api.rs b/deltachat-jsonrpc/src/api.rs index 9df9e1f9b..e571ff979 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 9a91bafe2..06cd3d3b5 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 @@ 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. diff --git a/deltachat-rpc-client/tests/conftest.py b/deltachat-rpc-client/tests/conftest.py index b8f160045..9bf4a1f75 100644 --- a/deltachat-rpc-client/tests/conftest.py +++ b/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("_"): diff --git a/deltachat-rpc-client/tests/test_multitransport.py b/deltachat-rpc-client/tests/test_multitransport.py index 2f6fd35c2..6a014ba18 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,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 diff --git a/src/constants.rs b/src/constants.rs index 9ba41f90f..6a9462378 100644 --- a/src/constants.rs +++ b/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; diff --git a/src/context.rs b/src/context.rs index 418d81f91..5e5209a53 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.background_fetch_any(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 fd4be4f31..4afaa5354 100644 --- a/src/imap.rs +++ b/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, } #[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 { - 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 { 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 = Vec::new(); let mut available_post_msgs: Vec = Vec::new(); diff --git a/src/scheduler.rs b/src/scheduler.rs index fbdff516f..21ba3ac8e 100644 --- a/src/scheduler.rs +++ b/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 { + // 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)]