diff --git a/deltachat-ffi/deltachat.h b/deltachat-ffi/deltachat.h index 52fcb355d6..69f9646bb1 100644 --- a/deltachat-ffi/deltachat.h +++ b/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 */ 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 f2432c1be6..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<()> { @@ -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 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/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..6a014ba18e 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.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/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/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..5e5209a539 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/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. /// diff --git a/src/imap.rs b/src/imap.rs index fd4be4f31f..b880b2e85c 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,9 @@ pub(crate) struct Imap { /// IMAP UID resync request receiver. pub(crate) resync_request_receiver: async_channel::Receiver<()>, + + /// Cancelled once a batch found messages, later batches fetch nothing. + pub(crate) background_fetch_stop_token: Option, } #[derive(Debug, Default)] @@ -237,21 +241,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 +421,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 +458,7 @@ impl Imap { .await .context("move_delete_messages")?; - Ok(()) + Ok(msgs_fetched) } /// Fetches new messages. @@ -532,6 +527,14 @@ 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() { + return Ok((0, false)); + } + if read_cnt > 0 { + 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 fbdff516fb..21ba3ac8e3 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)]