mirror of
https://github.com/chatmail/core.git
synced 2026-09-22 04:58:47 +03:00
fix: make background_fetch not wait on or trigger smtp connections
Also adds tests and docs to respective functions, clarifying background fetching behaviour and the `ACCOUNTS_BACKGROUND_FETCH_DONE` event, that came up in questions/discussions with UI devs lately.
This commit is contained in:
@@ -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.
|
* 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,
|
* The `DC_EVENT_ACCOUNTS_BACKGROUND_FETCH_DONE` event is emitted at the end,
|
||||||
* also on timeout, when another background fetch is already running
|
* also on timeout, when another background fetch is already running
|
||||||
* and when the call is ignored because the timeout is too small,
|
* 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.
|
* 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
|
* 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
|
* @memberof dc_accounts_t
|
||||||
* @param accounts The account manager as created by dc_accounts_new().
|
* @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,
|
* A call made while another background fetch is running gets the event immediately,
|
||||||
* and the running fetch keeps emitting events until its own marker.
|
* 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
|
* This event is only emitted by the account manager
|
||||||
*/
|
*/
|
||||||
|
|
||||||
|
|||||||
@@ -278,10 +278,26 @@ impl CommandApi {
|
|||||||
|
|
||||||
/// Performs a background fetch for all accounts in parallel with a timeout.
|
/// 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,
|
/// The `AccountsBackgroundFetchDone` event is emitted at the end even in case of timeout,
|
||||||
/// and immediately if another background fetch is already running.
|
/// 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
|
/// 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<()> {
|
async fn background_fetch(&self, timeout_in_seconds: f64) -> Result<()> {
|
||||||
let future = {
|
let future = {
|
||||||
let lock = self.accounts.read().await;
|
let lock = self.accounts.read().await;
|
||||||
@@ -292,6 +308,11 @@ impl CommandApi {
|
|||||||
Ok(())
|
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<()> {
|
async fn stop_background_fetch(&self) -> Result<()> {
|
||||||
self.accounts.read().await.stop_background_fetch();
|
self.accounts.read().await.stop_background_fetch();
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
@@ -70,6 +70,7 @@ class EventType(str, Enum):
|
|||||||
SELFAVATAR_CHANGED = "SelfavatarChanged"
|
SELFAVATAR_CHANGED = "SelfavatarChanged"
|
||||||
WEBXDC_STATUS_UPDATE = "WebxdcStatusUpdate"
|
WEBXDC_STATUS_UPDATE = "WebxdcStatusUpdate"
|
||||||
WEBXDC_INSTANCE_DELETED = "WebxdcInstanceDeleted"
|
WEBXDC_INSTANCE_DELETED = "WebxdcInstanceDeleted"
|
||||||
|
ACCOUNTS_BACKGROUND_FETCH_DONE = "AccountsBackgroundFetchDone"
|
||||||
CHATLIST_CHANGED = "ChatlistChanged"
|
CHATLIST_CHANGED = "ChatlistChanged"
|
||||||
CHATLIST_ITEM_CHANGED = "ChatlistItemChanged"
|
CHATLIST_ITEM_CHANGED = "ChatlistItemChanged"
|
||||||
ACCOUNTS_CHANGED = "AccountsChanged"
|
ACCOUNTS_CHANGED = "AccountsChanged"
|
||||||
|
|||||||
@@ -48,6 +48,13 @@ class DeltaChat:
|
|||||||
"""Stop ongoing background fetch."""
|
"""Stop ongoing background fetch."""
|
||||||
self.rpc.stop_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:
|
def maybe_network(self) -> None:
|
||||||
"""Indicate that the network conditions might have changed."""
|
"""Indicate that the network conditions might have changed."""
|
||||||
self.rpc.maybe_network()
|
self.rpc.maybe_network()
|
||||||
|
|||||||
@@ -1354,6 +1354,22 @@ def test_background_fetch(acf, dc):
|
|||||||
break
|
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):
|
def test_message_exists(acf):
|
||||||
ac1, ac2 = acf.get_online_accounts(2)
|
ac1, ac2 = acf.get_online_accounts(2)
|
||||||
chat = ac1.create_chat(ac2)
|
chat = ac1.create_chat(ac2)
|
||||||
|
|||||||
@@ -482,11 +482,15 @@ impl Accounts {
|
|||||||
/// return immediately even before the timeout expiration
|
/// return immediately even before the timeout expiration
|
||||||
/// or finishing fetching.
|
/// or finishing fetching.
|
||||||
///
|
///
|
||||||
|
/// Pending outgoing messages are not waited for and not triggered.
|
||||||
|
///
|
||||||
/// The `AccountsBackgroundFetchDone` event is emitted at the end,
|
/// The `AccountsBackgroundFetchDone` event is emitted at the end,
|
||||||
/// process all events until you get this one and you can safely return to the background
|
/// 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 notifications caused by timing race conditions.
|
||||||
/// If another background fetch is already running,
|
/// If another background fetch is already running,
|
||||||
/// nothing is fetched and the event is emitted immediately.
|
/// 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,
|
/// Returns a future that resolves when background fetch is done,
|
||||||
/// but does not capture `&self`.
|
/// but does not capture `&self`.
|
||||||
|
|||||||
@@ -601,9 +601,13 @@ impl Context {
|
|||||||
|
|
||||||
/// Does a single round of fetching messages from all transports 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 IO is stopped, pauses the scheduler and fetches over a dedicated connection
|
||||||
/// If I/O is stopped, fetches over a dedicated connection per transport
|
/// per transport, returning as soon as one of them fetched messages.
|
||||||
/// and returns 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<()> {
|
pub async fn background_fetch(&self) -> Result<()> {
|
||||||
if !(self.is_configured().await?) {
|
if !(self.is_configured().await?) {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
@@ -613,8 +617,9 @@ impl Context {
|
|||||||
info!(self, "background_fetch started.");
|
info!(self, "background_fetch started.");
|
||||||
|
|
||||||
if self.scheduler.is_running().await {
|
if self.scheduler.is_running().await {
|
||||||
self.scheduler.maybe_network().await;
|
self.scheduler.interrupt_inbox_idle().await;
|
||||||
self.wait_for_all_work_done().await;
|
let include_smtp = false;
|
||||||
|
self.wait_for_work_done(include_smtp).await;
|
||||||
} else {
|
} else {
|
||||||
self.scheduler.background_fetch_any(self).await?;
|
self.scheduler.background_fetch_any(self).await?;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -363,6 +363,9 @@ pub enum EventType {
|
|||||||
/// A call made while another background fetch is running gets the event immediately,
|
/// A call made while another background fetch is running gets the event immediately,
|
||||||
/// and the running fetch keeps emitting events until its own marker.
|
/// 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.
|
/// This event is only emitted by the account manager.
|
||||||
AccountsBackgroundFetchDone,
|
AccountsBackgroundFetchDone,
|
||||||
/// Inform that set of chats or the order of the chats in the chatlist has changed.
|
/// Inform that set of chats or the order of the chats in the chatlist has changed.
|
||||||
|
|||||||
@@ -215,10 +215,17 @@ impl SchedulerState {
|
|||||||
|
|
||||||
/// Indicate that the network likely has come back.
|
/// Indicate that the network likely has come back.
|
||||||
pub(crate) async fn maybe_network(&self) {
|
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 inner = self.inner.read().await;
|
||||||
let inboxes = match *inner {
|
let inboxes = match *inner {
|
||||||
InnerSchedulerState::Started(ref scheduler) => {
|
InnerSchedulerState::Started(ref scheduler) => {
|
||||||
scheduler.maybe_network();
|
scheduler.interrupt_inbox();
|
||||||
scheduler
|
scheduler
|
||||||
.inboxes
|
.inboxes
|
||||||
.iter()
|
.iter()
|
||||||
@@ -799,13 +806,6 @@ impl Scheduler {
|
|||||||
self.inboxes.iter()
|
self.inboxes.iter()
|
||||||
}
|
}
|
||||||
|
|
||||||
fn maybe_network(&self) {
|
|
||||||
for b in self.boxes() {
|
|
||||||
b.conn_state.interrupt();
|
|
||||||
}
|
|
||||||
self.interrupt_smtp();
|
|
||||||
}
|
|
||||||
|
|
||||||
fn maybe_network_lost(&self) {
|
fn maybe_network_lost(&self) {
|
||||||
for b in self.boxes() {
|
for b in self.boxes() {
|
||||||
b.conn_state.interrupt();
|
b.conn_state.interrupt();
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
use core::fmt;
|
use core::fmt;
|
||||||
use std::cmp::min;
|
use std::cmp::min;
|
||||||
use std::{iter::once, ops::Deref, sync::Arc};
|
use std::{ops::Deref, sync::Arc};
|
||||||
|
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
use humansize::{BINARY, format_size};
|
use humansize::{BINARY, format_size};
|
||||||
@@ -531,14 +531,15 @@ impl Context {
|
|||||||
Ok(ret)
|
Ok(ret)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns true if all background work is done.
|
/// Returns true if all background work is done,
|
||||||
async fn all_work_done(&self) -> bool {
|
/// 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 lock = self.scheduler.inner.read().await;
|
||||||
let stores: Vec<_> = match *lock {
|
let stores: Vec<_> = match *lock {
|
||||||
InnerSchedulerState::Started(ref sched) => sched
|
InnerSchedulerState::Started(ref sched) => sched
|
||||||
.boxes()
|
.boxes()
|
||||||
.map(|b| &b.conn_state.state)
|
.map(|b| &b.conn_state.state)
|
||||||
.chain(once(&sched.smtp.state))
|
.chain(include_smtp.then_some(&sched.smtp.state))
|
||||||
.map(|state| state.connectivity.clone())
|
.map(|state| state.connectivity.clone())
|
||||||
.collect(),
|
.collect(),
|
||||||
_ => return false,
|
_ => return false,
|
||||||
@@ -555,19 +556,26 @@ impl Context {
|
|||||||
|
|
||||||
/// Waits until background work is finished.
|
/// Waits until background work is finished.
|
||||||
pub async fn wait_for_all_work_done(&self) {
|
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,
|
// Ideally we could wait for connectivity change events,
|
||||||
// but sleep loop is good enough.
|
// but sleep loop is good enough.
|
||||||
|
|
||||||
// First 100 ms sleep in chunks of 10 ms.
|
// First 100 ms sleep in chunks of 10 ms.
|
||||||
for _ in 0..10 {
|
for _ in 0..10 {
|
||||||
if self.all_work_done().await {
|
if self.work_done(include_smtp).await {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
// If we are not finished in 100 ms, keep waking up every 100 ms.
|
// 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;
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -576,6 +584,34 @@ impl Context {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
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)]
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||||
async fn test_combine_connectivities() {
|
async fn test_combine_connectivities() {
|
||||||
|
|||||||
Reference in New Issue
Block a user