diff --git a/deltachat-rpc-client/tests/test_multitransport.py b/deltachat-rpc-client/tests/test_multitransport.py index 65eb766365..a8bfc09bde 100644 --- a/deltachat-rpc-client/tests/test_multitransport.py +++ b/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"] diff --git a/src/config.rs b/src/config.rs index df0a7b74c1..2890b16874 100644 --- a/src/config.rs +++ b/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?; } } _ => { diff --git a/src/context.rs b/src/context.rs index e5782bd384..6b2839d27a 100644 --- a/src/context.rs +++ b/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. /// @@ -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(), diff --git a/src/scheduler.rs b/src/scheduler.rs index f44a4021f4..74a586740e 100644 --- a/src/scheduler.rs +++ b/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 + 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 { 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); diff --git a/src/transport.rs b/src/transport.rs index 97a23dbd8f..3457b015fa 100644 --- a/src/transport.rs +++ b/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 + 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 diff --git a/src/transport/transport_tests.rs b/src/transport/transport_tests.rs index ec44a7bfd0..c48dca6ffe 100644 --- a/src/transport/transport_tests.rs +++ b/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],