fix: prevent transport de-synchronization because of early fetch cancellation

Came across this while investigating more test_transport_synchronization flakiness,
sometimes missing TransportsModified events or getting a missing configured_addr.
The underlying problem was that stopping IO was triggered immediately during
receiving sync messages, potentially *canceling* the processing of the sync message,
effectively de-syncing the device's view on transports.

src/context.rs

@@ -4,6 +4,7 @@ use std::collections::{BTreeMap, HashMap};
 use std::ffi::OsString;
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
+use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, OnceLock, Weak};
 use std::time::Duration;
 
@@ -258,6 +259,14 @@ pub struct InnerContext {
     /// This causes [`Context::wait_next_msgs`] to wake up.
     pub(crate) new_msgs_notify: Notify,
 
+    /// Whether IO should be restarted after the current fetch cycle completed.
+    ///
+    /// Set when a fetched transport sync message modified the transports.
+    /// Restarting from within the inbox loop would cancel it,
+    /// losing the remaining processing of the sync message
+    /// which is already stored and is never fetched again.
+    pub(crate) restart_io_after_fetch: AtomicBool,
+
     /// Server ID response if ID capability is supported
     /// and the server returned non-NIL on the inbox connection.
     /// <https://datatracker.ietf.org/doc/html/rfc2971>
@@ -486,6 +495,7 @@ impl Context {
             ratelimit: RwLock::new(Ratelimit::new(Duration::new(3, 0), 3.0)), // Allow at least 1 message every second + a burst of 3.
             quota: RwLock::new(BTreeMap::new()),
             new_msgs_notify,
+            restart_io_after_fetch: AtomicBool::new(false),
             server_id: RwLock::new(None),
             metadata: RwLock::new(BTreeMap::new()),
             creation_time: tools::Time::now(),

src/scheduler.rs

@@ -1,5 +1,8 @@
 use std::cmp;
+use std::future::Future;
 use std::num::NonZeroUsize;
+use std::pin::Pin;
+use std::sync::atomic::Ordering;
 
 use anyhow::{Context as _, Error, Result, bail};
 use async_channel::{self as channel, Receiver, Sender};
@@ -411,6 +414,12 @@ async fn inbox_loop(
         .await;
 }
 
+/// Same as `context.restart_io_if_running()`, but `Box::pin`ed and with a `+ Send` bound
+/// to break the async type cycle with the IMAP loop it restarts.
+fn restart_io_if_running_boxed(context: Context) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+    Box::pin(async move { context.restart_io_if_running().await })
+}
+
 async fn inbox_fetch_idle(ctx: &Context, imap: &mut Imap, mut session: Session) -> Result<Session> {
     let transport_id = session.transport_id();
 
@@ -496,6 +505,12 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, mut session: Session)
         .await
         .context("download_msgs")?;
 
+    if ctx.restart_io_after_fetch.swap(false, Ordering::Relaxed) {
+        // Restarting IO cancels the IMAP loop.
+        // Therefore, we only restart when we're anyways about to go IDLE.
+        task::spawn(restart_io_if_running_boxed(ctx.clone()));
+    }
+
     connection.connectivity.set_idle(ctx);
 
     ctx.emit_event(EventType::ImapInboxIdle);

src/transport.rs

@@ -9,7 +9,7 @@
 //! and configured list of connection candidates.
 
 use std::fmt;
-use std::pin::Pin;
+use std::sync::atomic::Ordering;
 
 use anyhow::{Context as _, Result, bail, format_err};
 use deltachat_contact_tools::{EmailAddress, addr_normalize};
@@ -671,18 +671,14 @@ pub(crate) async fn sync_transports(
 
     if modified {
         context.self_public_key.lock().await.take();
-        tokio::task::spawn(restart_io_if_running_boxed(context.clone()));
+        context
+            .restart_io_after_fetch
+            .store(true, Ordering::Relaxed);
         context.emit_event(EventType::TransportsModified);
     }
     Ok(())
 }
 
-/// Same as `context.restart_io_if_running()`, but `Box::pin`ed and with a `+ Send` bound,
-/// so that it can be called recursively.
-fn restart_io_if_running_boxed(context: Context) -> Pin<Box<dyn Future<Output = ()> + Send>> {
-    Box::pin(async move { context.restart_io_if_running().await })
-}
-
 /// Adds transport entry to the `transports` table with empty configuration.
 pub(crate) async fn add_pseudo_transport(context: &Context, addr: &str) -> Result<()> {
     context.sql

src/transport/transport_tests.rs

@@ -235,6 +235,32 @@ async fn test_promote_transport_same_second() -> Result<()> {
     promote_transport_and_check_success(alice, alice2, "alice@otherprovider.com").await
 }
 
+/// Tests that `sync_transports()` requests an IO restart
+/// if and only if it modified anything.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_sync_transports_requests_io_restart() -> Result<()> {
+    let alice = &TestContext::new_alice().await;
+
+    let data = TransportData {
+        configured: dummy_configured_login_param("alice@otherprovider.com").into(),
+        entered: EnteredLoginParam {
+            addr: "alice@otherprovider.com".to_string(),
+            ..Default::default()
+        },
+        timestamp: time(),
+        is_published: true,
+    };
+    let data = std::slice::from_ref(&data);
+    sync_transports(alice, data, &[]).await?;
+    assert!(alice.restart_io_after_fetch.swap(false, Ordering::Relaxed));
+
+    // Applying the same data again modifies nothing.
+    sync_transports(alice, data, &[]).await?;
+    assert!(!alice.restart_io_after_fetch.load(Ordering::Relaxed));
+
+    Ok(())
+}
+
 /// Promotes `addr` to primary on `alice` and checks the change syncs to `alice2`.
 async fn promote_transport_and_check_success(
     alice: &TestContext,