Allow pausing IO scheduler from inside core

To handle backups the UIs have to make sure they do stop the IO
scheduler and also don't accidentally restart it while working on it.
Since they have to call start_io from a bunch of locations this can be
a bit difficult to manage.

This introduces a mechanism for the core to pause IO for some time,
which is used by the imex function.  It interacts well with other
calls to dc_start_io() and dc_stop_io() making sure that when resumed
the scheduler will be running or not as the latest calls to them.

This was a little more invasive then hoped due to the scheduler.  The
additional abstraction of the scheduler on the context seems a nice
improvement though.

src/accounts.rs

@@ -271,14 +271,14 @@ impl Accounts {
     /// Notifies all accounts that the network may have become available.
     pub async fn maybe_network(&self) {
         for account in self.accounts.values() {
-            account.maybe_network().await;
+            account.scheduler.maybe_network().await;
         }
     }
 
     /// Notifies all accounts that the network connection may have been lost.
     pub async fn maybe_network_lost(&self) {
         for account in self.accounts.values() {
-            account.maybe_network_lost().await;
+            account.scheduler.maybe_network_lost(account).await;
         }
     }
 

src/chat.rs

@@ -639,7 +639,10 @@ impl ChatId {
         context.emit_msgs_changed_without_ids();
 
         context.set_config(Config::LastHousekeeping, None).await?;
-        context.interrupt_inbox(InterruptInfo::new(false)).await;
+        context
+            .scheduler
+            .interrupt_inbox(InterruptInfo::new(false))
+            .await;
 
         if chat.is_self_talk() {
             let mut msg = Message::new(Viewtype::Text);
@@ -1667,7 +1670,7 @@ impl Chat {
 
             maybe_set_logging_xdc(context, msg, self.id).await?;
         }
-        context.interrupt_ephemeral_task().await;
+        context.scheduler.interrupt_ephemeral_task().await;
         Ok(msg.id)
     }
 }
@@ -2201,7 +2204,10 @@ async fn send_msg_inner(context: &Context, chat_id: ChatId, msg: &mut Message) -
             context.emit_event(EventType::LocationChanged(Some(ContactId::SELF)));
         }
 
-        context.interrupt_smtp(InterruptInfo::new(false)).await;
+        context
+            .scheduler
+            .interrupt_smtp(InterruptInfo::new(false))
+            .await;
     }
 
     Ok(msg.id)
@@ -3433,7 +3439,10 @@ pub async fn forward_msgs(context: &Context, msg_ids: &[MsgId], chat_id: ChatId)
                     .await?;
                 curr_timestamp += 1;
                 if create_send_msg_job(context, new_msg_id).await?.is_some() {
-                    context.interrupt_smtp(InterruptInfo::new(false)).await;
+                    context
+                        .scheduler
+                        .interrupt_smtp(InterruptInfo::new(false))
+                        .await;
                 }
             }
             created_chats.push(chat_id);
@@ -3488,7 +3497,10 @@ pub async fn resend_msgs(context: &Context, msg_ids: &[MsgId]) -> Result<()> {
                 msg_id: msg.id,
             });
             if create_send_msg_job(context, msg.id).await?.is_some() {
-                context.interrupt_smtp(InterruptInfo::new(false)).await;
+                context
+                    .scheduler
+                    .interrupt_smtp(InterruptInfo::new(false))
+                    .await;
             }
         }
     }

src/config.rs

@@ -443,7 +443,7 @@ impl Context {
             Config::DeleteDeviceAfter => {
                 let ret = self.sql.set_raw_config(key.as_ref(), value).await;
                 // Interrupt ephemeral loop to delete old messages immediately.
-                self.interrupt_ephemeral_task().await;
+                self.scheduler.interrupt_ephemeral_task().await;
                 ret?
             }
             Config::Displayname => {

src/configure.rs

@@ -59,7 +59,7 @@ impl Context {
     /// Configures this account with the currently set parameters.
     pub async fn configure(&self) -> Result<()> {
         ensure!(
-            self.scheduler.read().await.is_none(),
+            !self.scheduler.is_running().await,
             "cannot configure, already running"
         );
         ensure!(
@@ -469,7 +469,9 @@ async fn configure(ctx: &Context, param: &mut LoginParam) -> Result<()> {
 
     ctx.set_config_bool(Config::FetchedExistingMsgs, false)
         .await?;
-    ctx.interrupt_inbox(InterruptInfo::new(false)).await;
+    ctx.scheduler
+        .interrupt_inbox(InterruptInfo::new(false))
+        .await;
 
     progress!(ctx, 940);
     update_device_chats_handle.await??;

src/contact.rs

@@ -1466,7 +1466,10 @@ pub(crate) async fn update_last_seen(
         > 0
         && timestamp > time() - SEEN_RECENTLY_SECONDS
     {
-        context.interrupt_recently_seen(contact_id, timestamp).await;
+        context
+            .scheduler
+            .interrupt_recently_seen(contact_id, timestamp)
+            .await;
     }
     Ok(())
 }

src/context.rs

@@ -24,7 +24,7 @@ use crate::key::{DcKey, SignedPublicKey};
 use crate::login_param::LoginParam;
 use crate::message::{self, MessageState, MsgId};
 use crate::quota::QuotaInfo;
-use crate::scheduler::Scheduler;
+use crate::scheduler::{IoPausedGuard, SchedulerState};
 use crate::sql::Sql;
 use crate::stock_str::StockStrings;
 use crate::timesmearing::SmearedTimestamp;
@@ -201,7 +201,7 @@ pub struct InnerContext {
     pub(crate) translated_stockstrings: StockStrings,
     pub(crate) events: Events,
 
-    pub(crate) scheduler: RwLock<Option<Scheduler>>,
+    pub(crate) scheduler: SchedulerState,
     pub(crate) ratelimit: RwLock<Ratelimit>,
 
     /// Recently loaded quota information, if any.
@@ -370,7 +370,7 @@ impl Context {
             wrong_pw_warning_mutex: Mutex::new(()),
             translated_stockstrings: stockstrings,
             events,
-            scheduler: RwLock::new(None),
+            scheduler: SchedulerState::new(),
             ratelimit: RwLock::new(Ratelimit::new(Duration::new(60, 0), 6.0)), // Allow to send 6 messages immediately, no more than once every 10 seconds.
             quota: RwLock::new(None),
             quota_update_request: AtomicBool::new(false),
@@ -395,42 +395,33 @@ impl Context {
             warn!(self, "can not start io on a context that is not configured");
             return;
         }
-
-        info!(self, "starting IO");
-        let mut lock = self.inner.scheduler.write().await;
-        if lock.is_none() {
-            match Scheduler::start(self.clone()).await {
-                Err(err) => error!(self, "Failed to start IO: {:#}", err),
-                Ok(scheduler) => *lock = Some(scheduler),
-            }
-        }
+        self.scheduler.start(self.clone()).await;
     }
 
     /// Stops the IO scheduler.
     pub async fn stop_io(&self) {
-        // Sending an event wakes up event pollers (get_next_event)
-        // so the caller of stop_io() can arrange for proper termination.
-        // For this, the caller needs to instruct the event poller
-        // to terminate on receiving the next event and then call stop_io()
-        // which will emit the below event(s)
-        info!(self, "stopping IO");
-        if let Some(debug_logging) = self.debug_logging.read().await.as_ref() {
-            debug_logging.loop_handle.abort();
-        }
-        if let Some(scheduler) = self.inner.scheduler.write().await.take() {
-            scheduler.stop(self).await;
-        }
+        self.scheduler.stop(self).await;
     }
 
     /// Restarts the IO scheduler if it was running before
     /// when it is not running this is an no-op
     pub async fn restart_io_if_running(&self) {
-        info!(self, "restarting IO");
-        let is_running = { self.inner.scheduler.read().await.is_some() };
-        if is_running {
-            self.stop_io().await;
-            self.start_io().await;
-        }
+        self.scheduler.restart(self).await;
+    }
+
+    /// Pauses the IO scheduler.
+    ///
+    /// This temporarily pauses the IO scheduler and will make sure calls to
+    /// [`Context::start_io`] are no-ops while being paused.
+    ///
+    /// It is recommended to call [`IoPausedGuard::resume`] rather than simply dropping it.
+    pub(crate) async fn pause_io(&self) -> IoPausedGuard<'_> {
+        self.scheduler.pause(self).await
+    }
+
+    /// Indicate that the network likely has come back.
+    pub async fn maybe_network(&self) {
+        self.scheduler.maybe_network().await;
     }
 
     /// Returns a reference to the underlying SQL instance.

src/ephemeral.rs

@@ -317,7 +317,7 @@ impl MsgId {
                     paramsv![ephemeral_timestamp, ephemeral_timestamp, self],
                 )
                 .await?;
-            context.interrupt_ephemeral_task().await;
+            context.scheduler.interrupt_ephemeral_task().await;
         }
         Ok(())
     }
@@ -345,7 +345,7 @@ pub(crate) async fn start_ephemeral_timers_msgids(
         )
         .await?;
     if count > 0 {
-        context.interrupt_ephemeral_task().await;
+        context.scheduler.interrupt_ephemeral_task().await;
     }
     Ok(())
 }

src/imap.rs

@@ -475,7 +475,7 @@ impl Imap {
             // Note that the `Config::DeleteDeviceAfter` timer starts as soon as the messages are
             // fetched while the per-chat ephemeral timers start as soon as the messages are marked
             // as noticed.
-            context.interrupt_ephemeral_task().await;
+            context.scheduler.interrupt_ephemeral_task().await;
         }
 
         let session = self
@@ -2224,7 +2224,10 @@ pub(crate) async fn markseen_on_imap_table(context: &Context, message_id: &str)
             paramsv![message_id],
         )
         .await?;
-    context.interrupt_inbox(InterruptInfo::new(false)).await;
+    context
+        .scheduler
+        .interrupt_inbox(InterruptInfo::new(false))
+        .await;
 
     Ok(())
 }

src/imex.rs

@@ -86,13 +86,17 @@ pub async fn imex(
 ) -> Result<()> {
     let cancel = context.alloc_ongoing().await?;
 
-    let res = imex_inner(context, what, path, passphrase)
-        .race(async {
-            cancel.recv().await.ok();
-            Err(format_err!("canceled"))
-        })
-        .await;
-
+    let res = {
+        let mut guard = context.pause_io().await;
+        let res = imex_inner(context, what, path, passphrase)
+            .race(async {
+                cancel.recv().await.ok();
+                Err(format_err!("canceled"))
+            })
+            .await;
+        guard.resume().await;
+        res
+    };
     context.free_ongoing().await;
 
     if let Err(err) = res.as_ref() {
@@ -413,7 +417,7 @@ async fn import_backup(
         "Cannot import backups to accounts in use."
     );
     ensure!(
-        context.scheduler.read().await.is_none(),
+        !context.scheduler.is_running().await,
         "cannot import backup, IO is running"
     );
 
@@ -523,7 +527,7 @@ async fn export_backup(context: &Context, dir: &Path, passphrase: String) -> Res
     sql::housekeeping(context).await.ok_or_log(context);
 
     ensure!(
-        context.scheduler.read().await.is_none(),
+        !context.scheduler.is_running().await,
         "cannot export backup, IO is running"
     );
 

src/job.rs

@@ -238,6 +238,7 @@ fn get_backoff_time_offset(tries: u32) -> i64 {
 pub(crate) async fn schedule_resync(context: &Context) -> Result<()> {
     context.resync_request.store(true, Ordering::Relaxed);
     context
+        .scheduler
         .interrupt_inbox(InterruptInfo {
             probe_network: false,
         })
@@ -250,7 +251,10 @@ pub async fn add(context: &Context, job: Job) -> Result<()> {
     job.save(context).await.context("failed to save job")?;
 
     info!(context, "interrupt: imap");
-    context.interrupt_inbox(InterruptInfo::new(false)).await;
+    context
+        .scheduler
+        .interrupt_inbox(InterruptInfo::new(false))
+        .await;
     Ok(())
 }
 

src/location.rs

@@ -267,7 +267,7 @@ pub async fn send_locations_to_chat(
     }
     context.emit_event(EventType::ChatModified(chat_id));
     if 0 != seconds {
-        context.interrupt_location().await;
+        context.scheduler.interrupt_location().await;
     }
     Ok(())
 }

src/message.rs

@@ -1419,7 +1419,10 @@ pub async fn delete_msgs(context: &Context, msg_ids: &[MsgId]) -> Result<()> {
     }
 
     // Interrupt Inbox loop to start message deletion and run housekeeping.
-    context.interrupt_inbox(InterruptInfo::new(false)).await;
+    context
+        .scheduler
+        .interrupt_inbox(InterruptInfo::new(false))
+        .await;
     Ok(())
 }
 
@@ -1531,7 +1534,10 @@ pub async fn markseen_msgs(context: &Context, msg_ids: Vec<MsgId>) -> Result<()>
                         )
                         .await
                         .context("failed to insert into smtp_mdns")?;
-                    context.interrupt_smtp(InterruptInfo::new(false)).await;
+                    context
+                        .scheduler
+                        .interrupt_smtp(InterruptInfo::new(false))
+                        .await;
                 }
             }
             updated_chat_ids.insert(curr_chat_id);

src/quota.rs

@@ -115,7 +115,9 @@ impl Context {
         let requested = self.quota_update_request.swap(true, Ordering::Relaxed);
         if !requested {
             // Quota update was not requested before.
-            self.interrupt_inbox(InterruptInfo::new(false)).await;
+            self.scheduler
+                .interrupt_inbox(InterruptInfo::new(false))
+                .await;
         }
         Ok(())
     }

src/scheduler.rs

@@ -5,6 +5,7 @@ use anyhow::{bail, Context as _, Result};
 use async_channel::{self as channel, Receiver, Sender};
 use futures::future::try_join_all;
 use futures_lite::FutureExt;
+use tokio::sync::{RwLock, RwLockWriteGuard};
 use tokio::task;
 
 use self::connectivity::ConnectivityStore;
@@ -23,6 +24,208 @@ use crate::tools::{duration_to_str, maybe_add_time_based_warnings};
 
 pub(crate) mod connectivity;
 
+/// State of the IO scheduler, as stored on the [`Context`].
+///
+/// The IO scheduler can be stopped or started, but core can also pause it.  After pausing
+/// the IO scheduler will be restarted only if it was running before paused or
+/// [`Context::start_io`] was called in the meantime while it was paused.
+#[derive(Debug, Default)]
+pub(crate) struct SchedulerState {
+    inner: RwLock<InnerSchedulerState>,
+}
+
+impl SchedulerState {
+    pub(crate) fn new() -> Self {
+        Default::default()
+    }
+
+    /// Whether the scheduler is currently running.
+    pub(crate) async fn is_running(&self) -> bool {
+        let inner = self.inner.read().await;
+        inner.scheduler.is_some()
+    }
+
+    /// Starts the scheduler if it is not yet started.
+    pub(crate) async fn start(&self, context: Context) {
+        let mut inner = self.inner.write().await;
+        inner.started = true;
+        if inner.scheduler.is_none() {
+            Self::do_start(inner, context).await;
+        }
+    }
+
+    /// Starts the scheduler if it is not yet started.
+    async fn do_start(mut inner: RwLockWriteGuard<'_, InnerSchedulerState>, context: Context) {
+        info!(context, "starting IO");
+        let ctx = context.clone();
+        match Scheduler::start(context).await {
+            Ok(scheduler) => inner.scheduler = Some(scheduler),
+            Err(err) => error!(&ctx, "Failed to start IO: {:#}", err),
+        }
+    }
+
+    /// Stops the scheduler if it is currently running.
+    pub(crate) async fn stop(&self, context: &Context) {
+        let mut inner = self.inner.write().await;
+        inner.started = false;
+        Self::do_stop(inner, context).await;
+    }
+
+    /// Stops the scheduler if it is currently running.
+    async fn do_stop(mut inner: RwLockWriteGuard<'_, InnerSchedulerState>, context: &Context) {
+        // Sending an event wakes up event pollers (get_next_event)
+        // so the caller of stop_io() can arrange for proper termination.
+        // For this, the caller needs to instruct the event poller
+        // to terminate on receiving the next event and then call stop_io()
+        // which will emit the below event(s)
+        info!(context, "stopping IO");
+        if let Some(debug_logging) = context.debug_logging.read().await.as_ref() {
+            debug_logging.loop_handle.abort();
+        }
+        if let Some(scheduler) = inner.scheduler.take() {
+            scheduler.stop(context).await;
+        }
+    }
+
+    /// Pauses the scheduler.
+    ///
+    /// If it is currently running the scheduler will be stopped.  When
+    /// [`IoPausedGuard::resume`] is called the scheduler is started again.  If in the
+    /// meantime [`SchedulerState::start`] or [`SchedulerState::stop`] is called resume will
+    /// do the right thing.
+    pub(crate) async fn pause<'a>(&'_ self, context: &'a Context) -> IoPausedGuard<'a> {
+        let mut inner = self.inner.write().await;
+        inner.paused = true;
+        Self::do_stop(inner, context).await;
+        IoPausedGuard {
+            context,
+            done: false,
+        }
+    }
+
+    /// Restarts the scheduler, only if it is running.
+    pub(crate) async fn restart(&self, context: &Context) {
+        info!(context, "restarting IO");
+        if self.is_running().await {
+            self.stop(context).await;
+            self.start(context.clone()).await;
+        }
+    }
+
+    /// Indicate that the network likely has come back.
+    pub(crate) async fn maybe_network(&self) {
+        let inner = self.inner.read().await;
+        let (inbox, oboxes) = match inner.scheduler {
+            Some(ref scheduler) => {
+                scheduler.maybe_network();
+                let inbox = scheduler.inbox.conn_state.state.connectivity.clone();
+                let oboxes = scheduler
+                    .oboxes
+                    .iter()
+                    .map(|b| b.conn_state.state.connectivity.clone())
+                    .collect::<Vec<_>>();
+                (inbox, oboxes)
+            }
+            None => return,
+        };
+        drop(inner);
+        // TODO: maybe this called code should move into scheduler.maybe_network() instead?
+        connectivity::idle_interrupted(inbox, oboxes).await;
+    }
+
+    /// Indicate that the network likely is lost.
+    pub(crate) async fn maybe_network_lost(&self, context: &Context) {
+        let inner = self.inner.read().await;
+        let stores = match inner.scheduler {
+            Some(ref scheduler) => {
+                scheduler.maybe_network_lost();
+                scheduler
+                    .boxes()
+                    .map(|b| b.conn_state.state.connectivity.clone())
+                    .collect()
+            }
+            None => return,
+        };
+        drop(inner);
+        // TODO; maybe this called code should move into scheduler.maybe_network_lost()
+        // instead?
+        connectivity::maybe_network_lost(context, stores).await;
+    }
+
+    pub(crate) async fn interrupt_inbox(&self, info: InterruptInfo) {
+        let inner = self.inner.read().await;
+        if let Some(ref scheduler) = inner.scheduler {
+            scheduler.interrupt_inbox(info);
+        }
+    }
+
+    pub(crate) async fn interrupt_smtp(&self, info: InterruptInfo) {
+        let inner = self.inner.read().await;
+        if let Some(ref scheduler) = inner.scheduler {
+            scheduler.interrupt_smtp(info);
+        }
+    }
+
+    pub(crate) async fn interrupt_ephemeral_task(&self) {
+        let inner = self.inner.read().await;
+        if let Some(ref scheduler) = inner.scheduler {
+            scheduler.interrupt_ephemeral_task();
+        }
+    }
+
+    pub(crate) async fn interrupt_location(&self) {
+        let inner = self.inner.read().await;
+        if let Some(ref scheduler) = inner.scheduler {
+            scheduler.interrupt_location();
+        }
+    }
+
+    pub(crate) async fn interrupt_recently_seen(&self, contact_id: ContactId, timestamp: i64) {
+        let inner = self.inner.read().await;
+        if let Some(ref scheduler) = inner.scheduler {
+            scheduler.interrupt_recently_seen(contact_id, timestamp);
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct InnerSchedulerState {
+    scheduler: Option<Scheduler>,
+    started: bool,
+    paused: bool,
+}
+
+#[derive(Debug)]
+pub(crate) struct IoPausedGuard<'a> {
+    context: &'a Context,
+    done: bool,
+}
+
+impl<'a> IoPausedGuard<'a> {
+    pub(crate) async fn resume(&mut self) {
+        self.done = true;
+        let inner = self.context.scheduler.inner.write().await;
+        if inner.started && inner.scheduler.is_none() {
+            SchedulerState::do_start(inner, self.context.clone()).await;
+        }
+    }
+}
+
+impl<'a> Drop for IoPausedGuard<'a> {
+    fn drop(&mut self) {
+        if self.done {
+            return;
+        }
+        let context = self.context.clone();
+        tokio::spawn(async move {
+            let inner = context.scheduler.inner.write().await;
+            if inner.started && inner.scheduler.is_none() {
+                SchedulerState::do_start(inner, context.clone()).await;
+            }
+        });
+    }
+}
+
 #[derive(Debug)]
 struct SchedBox {
     meaning: FolderMeaning,
@@ -46,56 +249,6 @@ pub(crate) struct Scheduler {
     recently_seen_loop: RecentlySeenLoop,
 }
 
-impl Context {
-    /// Indicate that the network likely has come back.
-    pub async fn maybe_network(&self) {
-        let lock = self.scheduler.read().await;
-        if let Some(scheduler) = &*lock {
-            scheduler.maybe_network();
-        }
-        connectivity::idle_interrupted(lock).await;
-    }
-
-    /// Indicate that the network likely is lost.
-    pub async fn maybe_network_lost(&self) {
-        let lock = self.scheduler.read().await;
-        if let Some(scheduler) = &*lock {
-            scheduler.maybe_network_lost();
-        }
-        connectivity::maybe_network_lost(self, lock).await;
-    }
-
-    pub(crate) async fn interrupt_inbox(&self, info: InterruptInfo) {
-        if let Some(scheduler) = &*self.scheduler.read().await {
-            scheduler.interrupt_inbox(info);
-        }
-    }
-
-    pub(crate) async fn interrupt_smtp(&self, info: InterruptInfo) {
-        if let Some(scheduler) = &*self.scheduler.read().await {
-            scheduler.interrupt_smtp(info);
-        }
-    }
-
-    pub(crate) async fn interrupt_ephemeral_task(&self) {
-        if let Some(scheduler) = &*self.scheduler.read().await {
-            scheduler.interrupt_ephemeral_task();
-        }
-    }
-
-    pub(crate) async fn interrupt_location(&self) {
-        if let Some(scheduler) = &*self.scheduler.read().await {
-            scheduler.interrupt_location();
-        }
-    }
-
-    pub(crate) async fn interrupt_recently_seen(&self, contact_id: ContactId, timestamp: i64) {
-        if let Some(scheduler) = &*self.scheduler.read().await {
-            scheduler.interrupt_recently_seen(contact_id, timestamp);
-        }
-    }
-}
-
 async fn inbox_loop(ctx: Context, started: Sender<()>, inbox_handlers: ImapConnectionHandlers) {
     use futures::future::FutureExt;
 

src/scheduler/connectivity.rs

@@ -3,7 +3,7 @@ use std::{iter::once, ops::Deref, sync::Arc};
 
 use anyhow::{anyhow, Result};
 use humansize::{format_size, BINARY};
-use tokio::sync::{Mutex, RwLockReadGuard};
+use tokio::sync::Mutex;
 
 use crate::events::EventType;
 use crate::imap::{scan_folders::get_watched_folder_configs, FolderMeaning};
@@ -12,7 +12,7 @@ use crate::quota::{
 };
 use crate::tools::time;
 use crate::{context::Context, log::LogExt};
-use crate::{scheduler::Scheduler, stock_str, tools};
+use crate::{stock_str, tools};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, EnumProperty, PartialOrd, Ord)]
 pub enum Connectivity {
@@ -156,19 +156,7 @@ impl ConnectivityStore {
 /// Set all folder states to InterruptingIdle in case they were `Connected` before.
 /// Called during `dc_maybe_network()` to make sure that `dc_accounts_all_work_done()`
 /// returns false immediately after `dc_maybe_network()`.
-pub(crate) async fn idle_interrupted(scheduler: RwLockReadGuard<'_, Option<Scheduler>>) {
-    let (inbox, oboxes) = match &*scheduler {
-        Some(Scheduler { inbox, oboxes, .. }) => (
-            inbox.conn_state.state.connectivity.clone(),
-            oboxes
-                .iter()
-                .map(|b| b.conn_state.state.connectivity.clone())
-                .collect::<Vec<_>>(),
-        ),
-        None => return,
-    };
-    drop(scheduler);
-
+pub(crate) async fn idle_interrupted(inbox: ConnectivityStore, oboxes: Vec<ConnectivityStore>) {
     let mut connectivity_lock = inbox.0.lock().await;
     // For the inbox, we also have to set the connectivity to InterruptingIdle if it was
     // NotConfigured before: If all folders are NotConfigured, dc_get_connectivity()
@@ -195,19 +183,7 @@ pub(crate) async fn idle_interrupted(scheduler: RwLockReadGuard<'_, Option<Sched
 /// Set the connectivity to "Not connected" after a call to dc_maybe_network_lost().
 /// If we did not do this, the connectivity would stay "Connected" for quite a long time
 /// after `maybe_network_lost()` was called.
-pub(crate) async fn maybe_network_lost(
-    context: &Context,
-    scheduler: RwLockReadGuard<'_, Option<Scheduler>>,
-) {
-    let stores: Vec<_> = match &*scheduler {
-        Some(sched) => sched
-            .boxes()
-            .map(|b| b.conn_state.state.connectivity.clone())
-            .collect(),
-        None => return,
-    };
-    drop(scheduler);
-
+pub(crate) async fn maybe_network_lost(context: &Context, stores: Vec<ConnectivityStore>) {
     for store in &stores {
         let mut connectivity_lock = store.0.lock().await;
         if !matches!(
@@ -249,9 +225,9 @@ impl Context {
     ///
     /// If the connectivity changes, a DC_EVENT_CONNECTIVITY_CHANGED will be emitted.
     pub async fn get_connectivity(&self) -> Connectivity {
-        let lock = self.scheduler.read().await;
-        let stores: Vec<_> = match &*lock {
-            Some(sched) => sched
+        let lock = self.scheduler.inner.read().await;
+        let stores: Vec<_> = match lock.scheduler {
+            Some(ref sched) => sched
                 .boxes()
                 .map(|b| b.conn_state.state.connectivity.clone())
                 .collect(),
@@ -332,9 +308,9 @@ impl Context {
         //                              Get the states from the RwLock
         // =============================================================================================
 
-        let lock = self.scheduler.read().await;
-        let (folders_states, smtp) = match &*lock {
-            Some(sched) => (
+        let lock = self.scheduler.inner.read().await;
+        let (folders_states, smtp) = match lock.scheduler {
+            Some(ref sched) => (
                 sched
                     .boxes()
                     .map(|b| (b.meaning, b.conn_state.state.connectivity.clone()))
@@ -503,9 +479,9 @@ impl Context {
 
     /// Returns true if all background work is done.
     pub async fn all_work_done(&self) -> bool {
-        let lock = self.scheduler.read().await;
-        let stores: Vec<_> = match &*lock {
-            Some(sched) => sched
+        let lock = self.scheduler.inner.read().await;
+        let stores: Vec<_> = match lock.scheduler {
+            Some(ref sched) => sched
                 .boxes()
                 .map(|b| &b.conn_state.state)
                 .chain(once(&sched.smtp.state))

src/webxdc.rs

@@ -421,7 +421,9 @@ impl Context {
                  DO UPDATE SET last_serial=excluded.last_serial, descr=excluded.descr",
                 paramsv![instance.id, status_update_serial, status_update_serial, descr],
             ).await?;
-            self.interrupt_smtp(InterruptInfo::new(false)).await;
+            self.scheduler
+                .interrupt_smtp(InterruptInfo::new(false))
+                .await;
         }
         Ok(())
     }