fix various transport multi-device de-synchronizinzation issues

deltachat-rpc-client/tests/test_multitransport.py

@@ -192,6 +192,7 @@ def test_transport_sync_new_as_primary(acfactory, log) -> None:
     log.section("ac1 changes the primary transport")
     ac1.set_config("configured_addr", transport2["addr"])
 
+    ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
     ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
     assert ac1_clone.get_config("configured_addr") == transport2["addr"]
 

src/config.rs

@@ -785,16 +785,19 @@ impl Context {
                                 (addr,),
                             )?;
 
-                            // Update the timestamp for the primary transport
-                            // so it becomes the first in `get_all_self_addrs()` list
-                            // and the list of relays distributed in the public key.
-                            // This ensures that messages will be sent
-                            // to the primary relay by the contacts
-                            // and will be fetched in background_fetch()
-                            // which only fetches from the primary transport.
+                            // `is_published=1`: an unpublished primary would be missing
+                            // from the relay list in the public key, so contacts would
+                            // never send to it.
+                            //
+                            // The timestamp must strictly increase because
+                            // other devices ignore the row update otherwise,
+                            // and contacts only adopt the re-signed key
+                            // if its signature timestamp increases.
                             transaction
                                 .execute(
-                                    "UPDATE transports SET add_timestamp=?, is_published=1 WHERE addr=?",
+                                    "UPDATE transports
+                                     SET add_timestamp=MAX(?, add_timestamp+1), is_published=1
+                                     WHERE addr=?",
                                     (time(), addr),
                                 )
                                 .context(
@@ -811,8 +814,9 @@ impl Context {
                             Ok(())
                         })
                         .await?;
-                    send_sync_transports(self).await?;
+                    // Invalidate the cache so the sync message cannot read a stale primary address.
                     self.sql.uncache_raw_config("configured_addr").await;
+                    send_sync_transports(self).await?;
                 }
             }
             _ => {

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};
@@ -407,6 +410,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();
 
@@ -491,6 +500,11 @@ 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) {
+        // Stopping IO from within the inbox loop would cancel it.
+        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};
@@ -666,18 +666,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

@@ -115,6 +115,19 @@ fn dummy_configured_login_param(addr: &str) -> ConfiguredLoginParam {
     }
 }
 
+async fn add_dummy_transport(t: &TestContext, addr: &str) -> Result<()> {
+    dummy_configured_login_param(addr)
+        .save_to_transports_table(
+            t,
+            &EnteredLoginParam {
+                addr: addr.to_string(),
+                ..Default::default()
+            },
+            time(),
+        )
+        .await
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_is_published_flag() -> Result<()> {
     let mut tcm = TestContextManager::new();
@@ -138,16 +151,7 @@ async fn test_is_published_flag() -> Result<()> {
     )
     .await;
 
-    dummy_configured_login_param("alice@otherprovider.com")
-        .save_to_transports_table(
-            alice,
-            &EnteredLoginParam {
-                addr: "alice@otherprovider.com".to_string(),
-                ..Default::default()
-            },
-            time(),
-        )
-        .await?;
+    add_dummy_transport(alice, "alice@otherprovider.com").await?;
     send_sync_transports(alice).await?;
     sync_and_check_recipients(alice, alice2, "alice@otherprovider.com alice@example.org").await;
 
@@ -195,10 +199,7 @@ async fn test_is_published_flag() -> Result<()> {
 
     SystemTime::shift(Duration::from_secs(2));
 
-    alice
-        .set_config(Config::ConfiguredAddr, Some("alice@otherprovider.com"))
-        .await?;
-    sync_and_check_recipients(alice, alice2, "alice@example.org alice@otherprovider.com").await;
+    promote_transport_and_check_success(alice, alice2, "alice@otherprovider.com").await?;
 
     check_addrs(
         alice,
@@ -215,6 +216,87 @@ async fn test_is_published_flag() -> Result<()> {
     Ok(())
 }
 
+/// Tests that changing the primary transport propagates to other devices
+/// even if the promoted transport was added within the same second.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_promote_transport_same_second() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice = &tcm.alice().await;
+    let alice2 = &tcm.alice().await;
+    for a in [alice, alice2] {
+        a.set_config_bool(Config::SyncMsgs, true).await?;
+        a.set_config_bool(Config::BccSelf, true).await?;
+    }
+
+    add_dummy_transport(alice, "alice@otherprovider.com").await?;
+    send_sync_transports(alice).await?;
+    sync_and_check_recipients(alice, alice2, "alice@otherprovider.com alice@example.org").await;
+
+    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,
+    alice2: &TestContext,
+    addr: &str,
+) -> Result<()> {
+    let old_timestamp = add_timestamp(alice2, addr).await;
+    alice.set_config(Config::ConfiguredAddr, Some(addr)).await?;
+    assert!(add_timestamp(alice, addr).await > old_timestamp);
+
+    alice.send_sync_msg().await?.unwrap();
+    let sync_msg = alice.pop_sent_msg().await;
+    assert_eq!(sync_msg.recipients, format!("alice@example.org {addr}"));
+    // Other devices switch their primary transport
+    // based on the From address of the sync message.
+    assert!(sync_msg.payload.contains(&format!("From: <{addr}>")));
+    alice2.recv_msg_trash(&sync_msg).await;
+
+    // add_timestamp must monotonically increase because
+    // other devices ignore the change otherwise.
+    assert!(add_timestamp(alice2, addr).await > old_timestamp);
+    assert_eq!(
+        alice2.get_config(Config::ConfiguredAddr).await?.as_deref(),
+        Some(addr)
+    );
+    Ok(())
+}
+
+async fn add_timestamp(t: &TestContext, addr: &str) -> i64 {
+    t.sql
+        .query_get_value("SELECT add_timestamp FROM transports WHERE addr=?", (addr,))
+        .await
+        .unwrap()
+        .unwrap()
+}
+
 struct Addresses {
     primary: &'static str,
     secondary_published: &'static [&'static str],