showcase wild cancel failure

Uncaching only after send_sync_transports() could send the sync message
from the old primary address, so other devices never switched.

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/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/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,61 @@ 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
+}
+
+/// 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],