diff --git a/deltachat-ffi/deltachat.h b/deltachat-ffi/deltachat.h index 0bb03b9e4..29dfbccdf 100644 --- a/deltachat-ffi/deltachat.h +++ b/deltachat-ffi/deltachat.h @@ -3187,17 +3187,28 @@ 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 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. + * For an account with IO stopped, the scheduler is paused + * and every transport is fetched concurrently on a dedicated connection. + * The account is done as soon as one transport received messages, the others stop. + * Only one batch of messages is fetched per transport this way, + * so a larger backlog is left to the next call or to started IO. + * + * For an account with IO running, IMAP IDLE is interrupted on every transport + * and the account is done once every transport is. + * + * The call never waits for outgoing messages and never triggers sending them itself. + * Received messages may still queue replies, securejoin handshakes for example, + * which go out only while IO is running. * * 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. + * without forgetting to create a generic notification if no message was fetched. + * The event carries no data identifying the call it belongs to, + * so it marks your own call only if no concurrent background fetch is happening. * * @memberof dc_accounts_t * @param accounts The account manager as created by dc_accounts_new(). @@ -6328,6 +6339,10 @@ void dc_event_unref(dc_event_t* event); * A call made while another background fetch is running gets the event immediately, * and the running fetch keeps emitting events until its own marker. * + * The event carries no data identifying the call it belongs to, + * so it marks your own call only if no concurrent background fetch is happening. + * Your own call has finished when dc_accounts_background_fetch() returns. + * * This event is only emitted by the account manager */ diff --git a/deltachat-jsonrpc/src/api.rs b/deltachat-jsonrpc/src/api.rs index 857224d92..1ffce2bb4 100644 --- a/deltachat-jsonrpc/src/api.rs +++ b/deltachat-jsonrpc/src/api.rs @@ -278,10 +278,26 @@ impl CommandApi { /// Performs a background fetch for all accounts in parallel with a timeout. /// + /// For an account with IO stopped, the scheduler is paused + /// and every transport is fetched concurrently on a dedicated connection. + /// The account is done as soon as one transport received messages, the others stop. + /// Only one batch of messages is fetched per transport this way, + /// so a larger backlog is left to the next call or to started IO. + /// + /// For an account with IO running, IMAP IDLE is interrupted on every transport + /// and the account is done once every transport is. + /// + /// The call never waits for outgoing messages and never triggers sending them itself. + /// Received messages may still queue replies, securejoin handshakes for example, + /// which go out only while IO is running. + /// Use `is_sending_finished()` to tell whether the outgoing queue is empty. + /// /// 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. + /// without forgetting to create a generic notification if no message was fetched. + /// The event carries no data identifying the call it belongs to, + /// so it marks your own call only if no concurrent background fetch is happening. async fn background_fetch(&self, timeout_in_seconds: f64) -> Result<()> { let future = { let lock = self.accounts.read().await; @@ -292,6 +308,11 @@ impl CommandApi { Ok(()) } + /// Stops an ongoing `background_fetch()` call, making it return early + /// without waiting for the remaining transports or for the timeout. + /// + /// The `AccountsBackgroundFetchDone` event is emitted as usual. + /// Does nothing if no background fetch is running. async fn stop_background_fetch(&self) -> Result<()> { self.accounts.read().await.stop_background_fetch(); Ok(()) diff --git a/deltachat-rpc-client/src/deltachat_rpc_client/const.py b/deltachat-rpc-client/src/deltachat_rpc_client/const.py index 846475b0e..93796a037 100644 --- a/deltachat-rpc-client/src/deltachat_rpc_client/const.py +++ b/deltachat-rpc-client/src/deltachat_rpc_client/const.py @@ -70,6 +70,7 @@ class EventType(str, Enum): SELFAVATAR_CHANGED = "SelfavatarChanged" WEBXDC_STATUS_UPDATE = "WebxdcStatusUpdate" WEBXDC_INSTANCE_DELETED = "WebxdcInstanceDeleted" + ACCOUNTS_BACKGROUND_FETCH_DONE = "AccountsBackgroundFetchDone" CHATLIST_CHANGED = "ChatlistChanged" CHATLIST_ITEM_CHANGED = "ChatlistItemChanged" ACCOUNTS_CHANGED = "AccountsChanged" diff --git a/deltachat-rpc-client/src/deltachat_rpc_client/deltachat.py b/deltachat-rpc-client/src/deltachat_rpc_client/deltachat.py index 9b976210d..21c000bad 100644 --- a/deltachat-rpc-client/src/deltachat_rpc_client/deltachat.py +++ b/deltachat-rpc-client/src/deltachat_rpc_client/deltachat.py @@ -48,6 +48,13 @@ class DeltaChat: """Stop ongoing background fetch.""" self.rpc.stop_background_fetch() + def wait_for_event(self, event_type=None) -> AttrDict: + """Wait until the next account manager event and return it.""" + while True: + next_event = AttrDict(self.rpc.wait_for_event(0)) + if event_type is None or next_event.kind == event_type: + return next_event + def maybe_network(self) -> None: """Indicate that the network conditions might have changed.""" self.rpc.maybe_network() diff --git a/deltachat-rpc-client/tests/test_something.py b/deltachat-rpc-client/tests/test_something.py index d5c0dabdf..cd04380b9 100644 --- a/deltachat-rpc-client/tests/test_something.py +++ b/deltachat-rpc-client/tests/test_something.py @@ -1354,6 +1354,22 @@ def test_background_fetch(acf, dc): break +def test_background_fetch_does_not_wait_for_sending(dc, acf): + alice, bob = acf.get_online_accounts(2) + alice_chat_bob = alice.create_chat(bob) + + alice.stop_io() + text = "x" * 200_000 + for _ in range(50): + alice_chat_bob.send_text(text) + assert not dc.is_sending_finished() + + alice.start_io() + dc.background_fetch(50) + dc.wait_for_event(EventType.ACCOUNTS_BACKGROUND_FETCH_DONE) + assert not dc.is_sending_finished() + + def test_message_exists(acf): ac1, ac2 = acf.get_online_accounts(2) chat = ac1.create_chat(ac2) diff --git a/src/accounts.rs b/src/accounts.rs index 20e901cc8..66f3b32f9 100644 --- a/src/accounts.rs +++ b/src/accounts.rs @@ -482,11 +482,15 @@ impl Accounts { /// return immediately even before the timeout expiration /// or finishing fetching. /// + /// Pending outgoing messages are not waited for and not triggered. + /// /// 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. + /// The event carries no data identifying the call it belongs to, + /// so it only safely refers to your call if no concurrent background fetch is happening. /// /// Returns a future that resolves when background fetch is done, /// but does not capture `&self`. diff --git a/src/context.rs b/src/context.rs index 5e5209a53..7d5384116 100644 --- a/src/context.rs +++ b/src/context.rs @@ -601,9 +601,13 @@ impl Context { /// 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 stopped, fetches over a dedicated connection per transport - /// and returns as soon as one of them fetched messages. + /// If IO is stopped, pauses the scheduler and fetches over a dedicated connection + /// per transport, returning as soon as one of them fetched messages. + /// If IO is running, interrupts IMAP IDLE on all transports + /// and waits until they are done fetching. + /// + /// Does not wait for outgoing messages to be sent out, + /// use [`crate::accounts::Accounts::is_sending_finished`] for that. pub async fn background_fetch(&self) -> Result<()> { if !(self.is_configured().await?) { return Ok(()); @@ -613,8 +617,9 @@ impl Context { info!(self, "background_fetch started."); if self.scheduler.is_running().await { - self.scheduler.maybe_network().await; - self.wait_for_all_work_done().await; + self.scheduler.interrupt_inbox_idle().await; + let include_smtp = false; + self.wait_for_work_done(include_smtp).await; } else { self.scheduler.background_fetch_any(self).await?; } diff --git a/src/events/payload.rs b/src/events/payload.rs index 196a8999c..4f419d487 100644 --- a/src/events/payload.rs +++ b/src/events/payload.rs @@ -363,6 +363,9 @@ pub enum EventType { /// A call made while another background fetch is running gets the event immediately, /// and the running fetch keeps emitting events until its own marker. /// + /// The event carries no data identifying the call it belongs to, + /// so it is unambiguous only if there are no concurrent background fetch calls. + /// /// 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/scheduler.rs b/src/scheduler.rs index 21ba3ac8e..6264181a7 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -215,10 +215,17 @@ impl SchedulerState { /// Indicate that the network likely has come back. pub(crate) async fn maybe_network(&self) { + self.interrupt_inbox_idle().await; + self.interrupt_smtp().await; + } + + /// Interrupts IDLE on all transports so that they fetch, + /// and marks them as having work to do. + pub(crate) async fn interrupt_inbox_idle(&self) { let inner = self.inner.read().await; let inboxes = match *inner { InnerSchedulerState::Started(ref scheduler) => { - scheduler.maybe_network(); + scheduler.interrupt_inbox(); scheduler .inboxes .iter() @@ -799,13 +806,6 @@ impl Scheduler { self.inboxes.iter() } - fn maybe_network(&self) { - for b in self.boxes() { - b.conn_state.interrupt(); - } - self.interrupt_smtp(); - } - fn maybe_network_lost(&self) { for b in self.boxes() { b.conn_state.interrupt(); diff --git a/src/scheduler/connectivity.rs b/src/scheduler/connectivity.rs index bed1d54df..f21ea14d4 100644 --- a/src/scheduler/connectivity.rs +++ b/src/scheduler/connectivity.rs @@ -1,6 +1,6 @@ use core::fmt; use std::cmp::min; -use std::{iter::once, ops::Deref, sync::Arc}; +use std::{ops::Deref, sync::Arc}; use anyhow::Result; use humansize::{BINARY, format_size}; @@ -531,14 +531,15 @@ impl Context { Ok(ret) } - /// Returns true if all background work is done. - async fn all_work_done(&self) -> bool { + /// Returns true if all background work is done, + /// checking the outgoing message queue only if `include_smtp` is set. + async fn work_done(&self, include_smtp: bool) -> bool { let lock = self.scheduler.inner.read().await; let stores: Vec<_> = match *lock { InnerSchedulerState::Started(ref sched) => sched .boxes() .map(|b| &b.conn_state.state) - .chain(once(&sched.smtp.state)) + .chain(include_smtp.then_some(&sched.smtp.state)) .map(|state| state.connectivity.clone()) .collect(), _ => return false, @@ -555,19 +556,26 @@ impl Context { /// Waits until background work is finished. pub async fn wait_for_all_work_done(&self) { + let include_smtp = true; + self.wait_for_work_done(include_smtp).await + } + + /// Waits until background work is finished, + /// checking the outgoing message queue only if `include_smtp` is set. + pub(crate) async fn wait_for_work_done(&self, include_smtp: bool) { // Ideally we could wait for connectivity change events, // but sleep loop is good enough. // First 100 ms sleep in chunks of 10 ms. for _ in 0..10 { - if self.all_work_done().await { + if self.work_done(include_smtp).await { break; } tokio::time::sleep(std::time::Duration::from_millis(10)).await; } // If we are not finished in 100 ms, keep waking up every 100 ms. - while !self.all_work_done().await { + while !self.work_done(include_smtp).await { tokio::time::sleep(std::time::Duration::from_millis(100)).await; } } @@ -576,6 +584,34 @@ impl Context { #[cfg(test)] mod tests { use super::*; + use crate::test_utils::TestContext; + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_background_fetch_leaves_smtp_alone() -> Result<()> { + let alice = TestContext::new_alice().await; + alice.start_io().await; + alice.wait_for_all_work_done().await; + + let smtp = match *alice.scheduler.inner.read().await { + InnerSchedulerState::Started(ref scheduler) => { + scheduler.smtp.state.connectivity.clone() + } + _ => panic!("scheduler is not running"), + }; + smtp.set_working(&alice); + + alice.background_fetch().await?; + assert!(!smtp.get_all_work_done()); + assert!(alice.scheduler.is_running().await); + + alice + .assert_warns_or_errors(&[ + "No IMAP connection candidates provided", + "IMAP got rate limited", + ]) + .await; + Ok(()) + } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_combine_connectivities() {