showcase wild cancel is fixed

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.

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"]
 

deltachat-rpc-client/tests/test_wild_cancel.py

@@ -0,0 +1,59 @@
+"""Shows the IO restart spawned by `sync_transports()` cancelling the
+`receive_imf()` that is still processing the sync message,
+losing the `configured_addr` update and its `TransportsModified` event.
+"""
+
+from queue import Empty
+
+from deltachat_rpc_client import AttrDict, EventType
+
+EVENT_TIMEOUT = 10
+
+
+def next_event(account, timeout=EVENT_TIMEOUT):
+    try:
+        return AttrDict(account._rpc.get_queue(account.id).get(timeout=timeout))
+    except Empty:
+        return None
+
+
+def drain_events(account, quiet=1):
+    while next_event(account, quiet) is not None:
+        pass
+
+
+def wait_for_transports_modified(account):
+    """Return True on TRANSPORTS_MODIFIED, False once IO restarted without it.
+
+    Stopping IO awaits the inbox loop, so a completed restart means
+    the message is not being processed anymore and no event is coming.
+    """
+    while True:
+        event = next_event(account)
+        if event is None:
+            return False
+        if event.kind == EventType.TRANSPORTS_MODIFIED:
+            return True
+        # ": starting IO" also excludes the "restarting IO" that precedes it
+        if event.kind == EventType.INFO and event.msg.endswith(": starting IO"):
+            return False
+
+
+def test_wild_cancel_loses_primary_transport(acfactory):
+    ac1 = acfactory.get_online_account()
+    ac1_clone = ac1.clone()
+    ac1_clone.bring_online()
+
+    ac1.add_transport_from_qr(acfactory.get_account_qr())
+    [transport1, transport2] = ac1.list_transports()
+    assert wait_for_transports_modified(ac1_clone)
+    assert ac1_clone.get_config("configured_addr") == transport1["addr"]
+    drain_events(ac1_clone)
+
+    new_addr = transport2["addr"]
+    ac1.set_config("configured_addr", new_addr)
+    assert wait_for_transports_modified(ac1_clone)
+    second = wait_for_transports_modified(ac1_clone)
+    configured_addr = ac1_clone.get_config("configured_addr")
+    assert second, f"no second TRANSPORTS_MODIFIED, configured_addr={configured_addr!r}, expected {new_addr!r}"
+    assert configured_addr == new_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/receive_imf.rs

@@ -796,6 +796,23 @@ pub(crate) async fn receive_imf_inner(
                     .execute_sync_items(sync_items, mime_parser.timestamp_sent)
                     .await;
 
+                // DEMO ONLY
+                // widens the window in which the IO restart spawned by `sync_transports()`
+                // cancels this very task, dropping the `configured_addr` update
+                // and the second `TransportsModified` event below.
+                // The sync message is not processed again,
+                // so the device stays on the old primary transport.
+                // 0.1ms is enough to lose the race every time,
+                // set `DC_WILD_CANCEL_SLEEP_MS=0` to restore the original timing.
+                let millis: f64 = std::env::var("DC_WILD_CANCEL_SLEEP_MS")
+                    .ok()
+                    .and_then(|value| value.parse().ok())
+                    .unwrap_or(0.1);
+                if millis > 0.0 {
+                    info!(context, "Wild-cancel demo: sleeping {millis}ms.");
+                    tokio::time::sleep(std::time::Duration::from_secs_f64(millis / 1000.0)).await;
+                }
+
                 // Receiving encrypted message from self updates primary transport.
                 let from_addr = &mime_parser.from.addr;
 

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],