Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-08-14 12:59:36 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix!: keep primary transport device-local

Browse Source 
Devices no longer implicitely use the From address of sync messages
to determine their primary transport. Receivers have no concept of it and own devices
may disagree on which relay is reachable because of VPN or different networks:

1. Make setting a primary transport (`configured_addr`) a per-device non-synced operation.

2. Transport rows (add/remove/unpublish) keep syncing like before.

3. A device reelects a primary if a sync message unpublished/removed the current primary
   if there is a better candidate.

4. `TransportsModified` event is emitted at most once on an incoming transport sync message.

Users will notice the change in that changing primary transport in settings/advanced/relays
will not synchronize to other devices anymore.
This commit is contained in:
holger krekel
2026-07-31 17:25:23 +02:00
parent 2cacdbfd4b
commit cd42efb36d
5 changed files with 158 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
deltachat-rpc-client/tests/test_multitransport.py
View File

@@ -150,12 +150,9 @@ def test_transport_synchronization(acfactory, log) -> None:
log.section("ac1 changes the primary transport")
ac1.set_config("configured_addr", transport3["addr"])
# One event for updated `add_timestamp` of the new primary transport,
# one event for the `configured_addr` update.
ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
[transport1, transport3] = ac1_clone.list_transports()
assert ac1_clone.get_config("configured_addr") == addr3
assert ac1_clone.get_config("configured_addr") == transport1["addr"]
log.section("ac1 removes the first transport")
ac1.delete_transport(transport1["addr"])
@@ -174,7 +171,7 @@ def test_transport_synchronization(acfactory, log) -> None:
def test_transport_sync_new_as_primary(acfactory, log) -> None:
"""Test synchronization of new transport as primary between devices."""
"""Test that a transport promoted on one device is usable on other devices."""
ac1, bob = acfactory.get_online_accounts(2)
ac1_clone = ac1.clone()
ac1_clone.bring_online()
@@ -193,10 +190,9 @@ def test_transport_sync_new_as_primary(acfactory, log) -> None:
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"]
assert ac1_clone.get_config("configured_addr") == transport1["addr"]
log.section("ac1_clone receives a message via the new primary transport")
log.section("ac1_clone receives a message via the new transport")
ac1_chat = ac1.create_chat(bob)
ac1_chat.send_text("Hello!")
bob_chat_id = bob.wait_for_incoming_msg_event().chat_id

4
src/config.rs
View File

@@ -192,7 +192,9 @@ pub enum Config {
#[strum(props(default = "0"))]
DeleteDeviceAfter,
/// The primary email address.
/// The primary email address, used for sending and background fetch.
///
/// Device-local, other devices keep their own primary transport.
ConfiguredAddr,
/// Deprecated(2026-04).

40
src/receive_imf.rs
View File

@@ -797,46 +797,6 @@ pub(crate) async fn receive_imf_inner(
context
.execute_sync_items(sync_items, mime_parser.timestamp_sent)
.await;
// Receiving encrypted message from self updates primary transport.
let from_addr = &mime_parser.from.addr;
let transport_changed = context
.sql
.transaction(|transaction| {
let transport_exists = transaction.query_row(
"SELECT COUNT(*) FROM transports WHERE addr=?",
(from_addr,),
|row| {
let count: i64 = row.get(0)?;
Ok(count > 0)
},
)?;
let transport_changed = if transport_exists {
transaction.execute(
"
UPDATE config SET value=? WHERE keyname='configured_addr' AND value!=?1
",
(from_addr,),
)? > 0
} else {
warn!(
context,
"Received sync message from unknown address {from_addr:?}."
);
false
};
Ok(transport_changed)
})
.await?;
if transport_changed {
info!(context, "Primary transport changed to {from_addr:?}.");
context.sql.uncache_raw_config("configured_addr").await;
context.self_public_key.lock().await.take();
context.emit_event(EventType::TransportsModified);
}
} else {
warn!(context, "Sync items are not encrypted.");
}

69
src/transport.rs
View File

@@ -636,20 +636,16 @@ pub(crate) async fn sync_transports(
modified |= save_transport(context, entered, configured, *timestamp, *is_published).await?;
}
context
let reelected = context
.sql
.transaction(|transaction| {
let configured_addr = transaction.query_row(
"SELECT value FROM config WHERE keyname='configured_addr'",
(),
|row| {
let addr: String = row.get(0)?;
Ok(addr)
},
)?;
for RemovedTransportData { addr, timestamp } in removed_transports {
if *addr == configured_addr {
continue;
let count: i64 =
transaction
.query_row("SELECT COUNT(*) FROM transports", (), |row| row.get(0))?;
if count <= 1 {
// Removing the last transport would unconfigure the account.
break;
}
modified |= transaction.execute(
"DELETE FROM transports
@@ -665,10 +661,17 @@ pub(crate) async fn sync_transports(
(addr, timestamp),
)?;
}
Ok(())
maybe_reelect_local_primary(transaction)
})
.await?;
if let Some(new_addr) = reelected {
info!(context, "Re-elected primary transport {new_addr:?}.");
context.sql.uncache_raw_config("configured_addr").await;
modified = true;
}
if modified {
context.self_public_key.lock().await.take();
context
@@ -679,6 +682,48 @@ pub(crate) async fn sync_transports(
Ok(())
}
/// Elects a new primary transport for the device if the current one
/// is not published or vanished, and there is a better candidate.
///
/// Returns the newly elected address if the primary transport changed.
fn maybe_reelect_local_primary(transaction: &mut rusqlite::Transaction) -> Result<Option<String>> {
let configured_addr: String = transaction.query_row(
"SELECT value FROM config WHERE keyname='configured_addr'",
(),
|row| row.get(0),
)?;
// Newest transports first, they are the most likely to work.
let transports: Vec<(String, bool)> = transaction
.prepare(
"SELECT addr, is_published FROM transports
ORDER BY add_timestamp DESC, id DESC",
)?
.query_map((), |row| Ok((row.get(0)?, row.get(1)?)))?
.collect::<rusqlite::Result<_>>()?;
// Nothing to do if the current primary is still there and published.
if transports
.iter()
.any(|(addr, is_published)| *is_published && *addr == configured_addr)
{
return Ok(None);
}
// Take an unpublished transport only if nothing is published.
let published = transports.iter().find(|(_, is_published)| *is_published);
let Some((new_addr, _)) = published.or_else(|| transports.first()) else {
return Ok(None);
};
if *new_addr == configured_addr {
// The primary transport may be the only remaining one.
return Ok(None);
}
transaction.execute(
"UPDATE config SET value=? WHERE keyname='configured_addr'",
(new_addr,),
)?;
Ok(Some(new_addr.clone()))
}
/// Adds transport entry to the `transports` table with empty configuration.
pub(crate) async fn add_pseudo_transport(context: &Context, addr: &str) -> Result<()> {
context.sql

113
src/transport/transport_tests.rs
View File

@@ -115,6 +115,18 @@ fn dummy_configured_login_param(addr: &str) -> ConfiguredLoginParam {
}
}
fn dummy_transport_data(addr: &str, is_published: bool) -> TransportData {
TransportData {
configured: dummy_configured_login_param(addr).into(),
entered: EnteredLoginParam {
addr: addr.to_string(),
..Default::default()
},
timestamp: time(),
is_published,
}
}
async fn add_dummy_transport(t: &TestContext, addr: &str) -> Result<()> {
dummy_configured_login_param(addr)
.save_to_transports_table(
@@ -199,7 +211,11 @@ async fn test_is_published_flag() -> Result<()> {
SystemTime::shift(Duration::from_secs(2));
promote_transport_and_check_success(alice, alice2, "alice@otherprovider.com").await?;
promote_transport_and_sync(alice, alice2, "alice@otherprovider.com").await?;
alice2
.set_config(Config::ConfiguredAddr, Some("alice@otherprovider.com"))
.await?;
check_addrs(
alice,
@@ -216,8 +232,8 @@ 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.
/// Tests that promoting a transport bumps its `add_timestamp` on other devices
/// even if it 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();
@@ -232,7 +248,7 @@ async fn test_promote_transport_same_second() -> Result<()> {
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
promote_transport_and_sync(alice, alice2, "alice@otherprovider.com").await
}
/// Tests that `sync_transports()` requests an IO restart
@@ -241,15 +257,7 @@ async fn test_promote_transport_same_second() -> Result<()> {
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 = dummy_transport_data("alice@otherprovider.com", true);
let data = std::slice::from_ref(&data);
sync_transports(alice, data, &[]).await?;
assert!(alice.restart_io_after_fetch.swap(false, Ordering::Relaxed));
@@ -261,21 +269,23 @@ async fn test_sync_transports_requests_io_restart() -> Result<()> {
Ok(())
}
/// Promotes `addr` to primary on `alice` and checks the change syncs to `alice2`.
async fn promote_transport_and_check_success(
/// Promotes `addr` on `alice` and syncs the transport update to `alice2`,
/// whose own primary transport must stay unchanged.
async fn promote_transport_and_sync(
alice: &TestContext,
alice2: &TestContext,
addr: &str,
) -> Result<()> {
let old_timestamp = add_timestamp(alice2, addr).await;
let alice2_primary = alice2.get_config(Config::ConfiguredAddr).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.
// The sync message comes from the new primary,
// which must not make `alice2` adopt it as its own primary.
assert!(sync_msg.payload.contains(&format!("From: <{addr}>")));
alice2.recv_msg_trash(&sync_msg).await;
@@ -283,8 +293,8 @@ async fn promote_transport_and_check_success(
// 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)
alice2.get_config(Config::ConfiguredAddr).await?,
alice2_primary
);
Ok(())
}
@@ -297,6 +307,71 @@ async fn add_timestamp(t: &TestContext, addr: &str) -> i64 {
.unwrap()
}
/// Tests that the local primary transport is re-elected
/// if a synced change unpublished or removed it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_reelect_local_primary() -> Result<()> {
let alice = &TestContext::new_alice().await;
add_dummy_transport(alice, "alice@otherprovider.com").await?;
// Another device unpublished the primary transport.
let unpublished = dummy_transport_data("alice@example.org", false);
sync_transports(alice, std::slice::from_ref(&unpublished), &[]).await?;
assert_eq!(
alice.get_config(Config::ConfiguredAddr).await?.as_deref(),
Some("alice@otherprovider.com")
);
// Another device removed the new primary transport.
let removed = RemovedTransportData {
addr: "alice@otherprovider.com".to_string(),
timestamp: time(),
};
sync_transports(alice, &[], std::slice::from_ref(&removed)).await?;
assert_eq!(
alice.get_config(Config::ConfiguredAddr).await?.as_deref(),
Some("alice@example.org")
);
Ok(())
}
/// Tests which transport is elected as the local primary one.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_maybe_reelect_local_primary() -> Result<()> {
let t = &TestContext::new_alice().await;
// The only transport is kept even if it is unpublished.
t.sql
.execute("UPDATE transports SET is_published=0", ())
.await?;
assert_eq!(t.sql.transaction(maybe_reelect_local_primary).await?, None);
// A published primary transport is kept even if newer transports exist.
t.sql
.execute("UPDATE transports SET is_published=1", ())
.await?;
add_dummy_transport(t, "alice@one.com").await?;
assert_eq!(t.sql.transaction(maybe_reelect_local_primary).await?, None);
// The most recently added published transport is elected.
SystemTime::shift(Duration::from_secs(2));
add_dummy_transport(t, "alice@two.com").await?;
t.sql
.execute(
"UPDATE transports SET is_published=0 WHERE addr=?",
("alice@example.org",),
)
.await?;
assert_eq!(
t.sql
.transaction(maybe_reelect_local_primary)
.await?
.as_deref(),
Some("alice@two.com")
);
Ok(())
}
struct Addresses {
primary: &'static str,
secondary_published: &'static [&'static str],