From 2cacdbfd4be5f354371a6d7a4867b207e05cef28 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Fri, 31 Jul 2026 03:03:39 +0200 Subject: [PATCH] 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 | 10 ++++++++++ src/scheduler.rs | 15 +++++++++++++++ src/transport.rs | 12 ++++-------- src/transport/transport_tests.rs | 26 ++++++++++++++++++++++++++ 4 files changed, 55 insertions(+), 8 deletions(-) diff --git a/src/context.rs b/src/context.rs index 6cf192d60..fc50fa2c0 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 70c64b8c6..30a27eb8d 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}; @@ -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 + 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(); @@ -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); diff --git a/src/transport.rs b/src/transport.rs index 2fee17bf6..d6fbe2105 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}; @@ -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 + 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 136c14491..c48dca6ff 100644 --- a/src/transport/transport_tests.rs +++ b/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,