Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-09-22 04:58:47 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat!: remove a relay immediately instead of unpublishing it

Browse Source 
Removing a relay now takes effect immediately:

- the profile stops fetching and advertising it,

- secondary devices immediately apply the removal through the transport sync,

Upgrading removes unpublished relays and triggers keyupdates.

BREAKING CHANGE: set_transport_unpublished() is removed: UIs call delete_transport() when the user removes a relay.

BREAKING CHANGE: list_transports_ex() and the TransportListEntry type are removed: use list_transports().

BREAKING CHANGE: delete_transport() no longer refuses to remove the primary transport: it refuses only to remove the last one and re-elects the sending transport as needed.

BREAKING CHANGE: TransportsModified is now also emitted on the device modifying the transports, not only on devices applying the synced change.

Deprecated: DC_STR_PHASING_OUT
This commit is contained in:
holger krekel
2026-08-29 11:30:46 +02:00
parent 3d61e0f349
commit 70a01a6813
33 changed files with 473 additions and 827 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
deltachat-ffi/deltachat.h
View File

@@ -6474,9 +6474,10 @@ void dc_event_unref(dc_event_t* event);
* Transport relay added/deleted or default has changed.
* UI should update the list.
*
* The event is emitted when the transports are modified on another device
* using the JSON-RPC calls `add_or_update_transport`, `add_transport_from_qr`, `delete_transport`,
* `set_transport_unpublished` or `set_config(configured_addr)`.
* The event is emitted on the device modifying the transports
* as well as on other devices applying the synced change,
* for the JSON-RPC calls `add_or_update_transport`, `add_transport_from_qr`,
* `delete_transport` or `set_config(configured_addr)`.
*/
#define DC_EVENT_TRANSPORTS_MODIFIED 2600
@@ -7282,11 +7283,7 @@ void dc_event_unref(dc_event_t* event);
/// "Message pinned by %1$s."
#define DC_STR_MESSAGE_PINNED_BY_OTHER 244
/// "Phasing out"
///
/// Used in connectivity view to flag unpublished relays.
/// This should match the wording used for relay deletion confirmation,
/// saying "Before deletion, it will be gradually phased out so your contacts can switch over smoothly"
/// @deprecated 2026-08-31
#define DC_STR_PHASING_OUT 245
/**

66
deltachat-jsonrpc/src/api.rs
View File

@@ -65,7 +65,6 @@ use self::types::{
};
use crate::api::types::appversions::JsonrpcAppSource;
use crate::api::types::chat_list::{ChatListItemFetchResult, get_chat_list_item_by_id};
use crate::api::types::login_param::TransportListEntry;
use crate::api::types::qr::{QrObject, SecurejoinSource, SecurejoinUiPath};
#[derive(Debug)]
@@ -502,8 +501,7 @@ impl CommandApi {
/// - [Self::add_transport_from_qr()] to add a transport
/// from a server encoded in a QR code.
/// - [Self::list_transports()] to get a list of all configured transports.
/// - [Self::set_transport_unpublished()] to remove a transport.
/// - [Self::set_transport_unpublished()] to set whether contacts see this transport.
/// - [Self::delete_transport()] to remove a transport.
async fn add_or_update_transport(
&self,
account_id: u32,
@@ -528,32 +526,8 @@ impl CommandApi {
/// Returns the list of all email accounts that are used as a transport in the current profile.
/// Use [Self::add_or_update_transport()] to add or change a transport
/// and [Self::set_transport_unpublished()] to remove a transport.
/// and [Self::delete_transport()] to remove a transport.
async fn list_transports(&self, account_id: u32) -> Result<Vec<EnteredLoginParam>> {
let ctx = self.get_context(account_id).await?;
let res = ctx
.list_transports()
.await?
.into_iter()
.filter(|t| !t.is_unpublished)
.map(|t| t.param.into())
.collect();
Ok(res)
}
/// Deprecated 2026-06: This is not needed by UI implementations anymore,
/// because unpublished relays now count as removed from the user point of view,
/// and must not be shown in the list of relays.
/// This means that UIs should use `list_transports()` instead of this function.
///
/// Returns the list of all email accounts that are used as a transport in the current profile.
///
/// As opposed to `list_transports()`, this function also returns unpublished transports,
/// and for each returned transport it returns the information whether or not is `unpublished`.
///
/// Use [Self::add_or_update_transport()] to add or change a transport
/// and [Self::set_transport_unpublished()] to change whether a transport is 'published'.
async fn list_transports_ex(&self, account_id: u32) -> Result<Vec<TransportListEntry>> {
let ctx = self.get_context(account_id).await?;
let res = ctx
.list_transports()
@@ -564,41 +538,17 @@ impl CommandApi {
Ok(res)
}
/// Immediately deletes a transport, potentially causing messages not to arrive.
/// This must ONLY be used by the automated tests.
/// UI implementations must use [`Self::set_transport_unpublished`] instead.
/// Removes a transport.
/// UIs should call this function when the user removes a relay.
///
/// The last transport cannot be removed.
/// If the removed transport was the one used for sending,
/// another one is chosen automatically.
async fn delete_transport(&self, account_id: u32, addr: String) -> Result<()> {
let ctx = self.get_context(account_id).await?;
ctx.delete_transport(&addr).await
}
/// Change whether the transport is unpublished.
/// UIs should call this function when the user clicks on "Remove".
/// Core will keep listening on this transport for some time,
/// and automatically remove it once it is no longer needed.
///
/// Unpublished transports are not advertised to contacts,
/// and self-sent messages are not sent there,
/// so that we don't cause extra messages to the corresponding inbox,
/// but can still receive messages from contacts who don't know our new transport addresses yet.
///
/// When more transports are added by [`Self::add_or_update_transport()`] or [`Self::add_transport_from_qr`],
/// the least recently needed unpublished transport is automatically removed
/// if this is necessary in order to stay below the maximum number of allowed relays.
/// Also, unpublished transports that are not used to receive any new messages for a time defined by
/// [`UNPUBLISHED_TRANSPORT_KEEP_TIME`] are automatically removed.
///
/// [`UNPUBLISHED_TRANSPORT_KEEP_TIME`]: deltachat::sql::UNPUBLISHED_TRANSPORT_KEEP_TIME
async fn set_transport_unpublished(
&self,
account_id: u32,
addr: String,
unpublished: bool,
) -> Result<()> {
let ctx = self.get_context(account_id).await?;
ctx.set_transport_unpublished(&addr, unpublished).await
}
/// Signal an ongoing process to stop.
async fn stop_ongoing_process(&self, account_id: u32) -> Result<()> {
let ctx = self.get_context(account_id).await?;

6
deltachat-jsonrpc/src/api/types/events.rs
View File

@@ -478,9 +478,9 @@ pub enum EventType {
///
/// UI should update the list.
///
/// This event is emitted when transport
/// synchronization messages arrives,
/// but not when the UI modifies the transport list by itself.
/// The event is emitted on the device modifying
/// the transports as well as on other devices
/// applying the synced change.
TransportsModified,
}

19
deltachat-jsonrpc/src/api/types/login_param.rs
View File

@@ -4,16 +4,6 @@ use serde::Deserialize;
use serde::Serialize;
use yerpc::TypeDef;
#[derive(Serialize, TypeDef, schemars::JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct TransportListEntry {
/// The login data entered by the user.
pub param: EnteredLoginParam,
/// Whether this transport is set to 'unpublished'.
/// See `set_transport_unpublished` / `setTransportUnpublished` for details.
pub is_unpublished: bool,
}
/// Login parameters entered by the user.
///
/// Usually it will be enough to only set `addr` and `password`,
@@ -68,15 +58,6 @@ pub struct EnteredLoginParam {
pub certificate_checks: Option<EnteredCertificateChecks>,
}
impl From<dc::TransportListEntry> for TransportListEntry {
fn from(transport: dc::TransportListEntry) -> Self {
TransportListEntry {
param: transport.param.into(),
is_unpublished: transport.is_unpublished,
}
}
}
impl From<dc::EnteredLoginParam> for EnteredLoginParam {
fn from(param: dc::EnteredLoginParam) -> Self {
let imap_security: Socket = param.imap.security.into();

4
deltachat-rpc-client/src/deltachat_rpc_client/account.py
View File

@@ -143,10 +143,6 @@ class Account:
"""Delete a transport."""
self._rpc.delete_transport(self.id, addr)
def set_transport_unpublished(self, addr: str, unpublished: bool = True):
"""Unpublish the transport."""
self._rpc.set_transport_unpublished(self.id, addr, unpublished)
@futuremethod
def list_transports(self):
"""Return the list of all email accounts that are used as a transport in the current profile."""

13
deltachat-rpc-client/tests/test_cross_core.py
View File

@@ -89,8 +89,9 @@ def test_second_device(acf, alice_and_remote_bob) -> None:
assert new_account.get_config("addr") == remote_eval("bob.get_config('addr')")
def test_keyupdate_against_core_2_48_march_2026(acf, alice_and_remote_bob):
"""Test 2.48 Bob learns a new relay of Alice from a keyupdate, and is shown nothing."""
@pytest.mark.parametrize("replace_relay", [False, True], ids=["add", "replace"])
def test_keyupdate_against_core_2_48_march_2026(acf, alice_and_remote_bob, replace_relay):
"""Test 2.48 Bob learns a relay change of Alice from a keyupdate, and is shown nothing."""
alice, alice_contact_bob, remote_eval = alice_and_remote_bob("2.48.0")
def bob_sees():
@@ -115,8 +116,10 @@ def test_keyupdate_against_core_2_48_march_2026(acf, alice_and_remote_bob):
# without waiting, the re-signed key can tie with the copy Bob holds, keeping his.
time.sleep(2)
alice.add_transport_from_qr(acf.get_account_qr())
alice.bring_online()
(new_addr,) = [t["addr"] for t in alice.list_transports() if t["addr"] != old_addr]
if replace_relay:
alice.delete_transport(old_addr)
alice.bring_online()
# The 2.48 core has no encryption enforcement, but the keyupdate MDN without
# referenced message keeps it invisible; merging happens before the trashing.
@@ -130,3 +133,7 @@ def test_keyupdate_against_core_2_48_march_2026(acf, alice_and_remote_bob):
# It also leaves no trace: no chat with Alice, no message anywhere,
# and no address-contact for the address it was sent from.
assert bob_sees() == before
if replace_relay:
remote_eval("bob_contact_alice.create_chat().send_text('hello after replacement')")
assert alice.wait_for_incoming_msg().get_snapshot().text == "hello after replacement"

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

@@ -20,13 +20,18 @@ def test_add_second_address(acf) -> None:
first_addr = account.list_transports()[0]["addr"]
second_addr = account.list_transports()[1]["addr"]
third_addr = account.list_transports()[2]["addr"]
# Cannot delete the first address.
with pytest.raises(JsonRpcError):
account.delete_transport(first_addr)
assert account.get_config("configured_addr") == first_addr
account.delete_transport(first_addr)
assert len(account.list_transports()) == 2
assert account.get_config("configured_addr") != first_addr
account.delete_transport(second_addr)
assert len(account.list_transports()) == 2
assert len(account.list_transports()) == 1
with pytest.raises(JsonRpcError):
account.delete_transport(third_addr)
def test_change_address(acf) -> None:
@@ -63,8 +68,6 @@ def test_change_address(acf) -> None:
alice_vcard = alice.self_contact.make_vcard()
assert old_alice_addr not in alice_vcard
assert new_alice_addr in alice_vcard
with pytest.raises(JsonRpcError):
alice.delete_transport(new_alice_addr)
alice.start_io()
alice_chat_bob.send_text("Hello again!")
@@ -122,6 +125,10 @@ def test_transport_synchronization(acf, log) -> None:
if "scheduler is running" in ev.msg:
return
def wait_transports(ac, n):
while len(ac.list_transports()) != n:
ac.wait_for_event(EventType.TRANSPORTS_MODIFIED)
ac1, ac2 = acf.get_online_accounts(2)
ac1_clone = ac1.clone()
ac1_clone.bring_online()
@@ -129,15 +136,13 @@ def test_transport_synchronization(acf, log) -> None:
qr = acf.get_account_qr()
ac1.add_transport_from_qr(qr)
ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
wait_transports(ac1_clone, 2)
wait_for_io_started(ac1_clone)
assert len(ac1.list_transports()) == 2
assert len(ac1_clone.list_transports()) == 2
ac1_clone.add_transport_from_qr(qr)
ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
wait_transports(ac1, 3)
wait_for_io_started(ac1)
assert len(ac1.list_transports()) == 3
assert len(ac1_clone.list_transports()) == 3
log.section("ac1 clone removes second transport")
@@ -145,21 +150,17 @@ def test_transport_synchronization(acf, log) -> None:
addr3 = transport3["addr"]
ac1_clone.delete_transport(transport2["addr"])
ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
wait_transports(ac1, 2)
wait_for_io_started(ac1)
[transport1, transport3] = ac1.list_transports()
log.section("ac1 changes the primary transport")
log.section("ac1 changes the sending transport")
ac1.set_config("configured_addr", transport3["addr"])
ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
[transport1, transport3] = ac1_clone.list_transports()
assert ac1_clone.get_config("configured_addr") == transport1["addr"]
log.section("ac1 removes the first transport")
ac1.delete_transport(transport1["addr"])
ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
wait_transports(ac1_clone, 1)
wait_for_io_started(ac1_clone)
[transport3] = ac1_clone.list_transports()
assert transport3["addr"] == addr3
@@ -181,6 +182,7 @@ def test_transport_sync_new_as_primary(acf, log) -> None:
qr = acf.get_account_qr()
ac1.add_transport_from_qr(qr)
ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
ac1_transports = ac1.list_transports()
assert len(ac1_transports) == 2
[transport1, transport2] = ac1_transports
@@ -190,6 +192,7 @@ def test_transport_sync_new_as_primary(acf, log) -> None:
log.section("ac1 changes the primary transport")
ac1.set_config("configured_addr", transport2["addr"])
ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
assert ac1_clone.get_config("configured_addr") == transport1["addr"]
@@ -236,18 +239,8 @@ def test_transport_limit(acf) -> None:
account.add_transport_from_qr(qr)
second_addr = account.list_transports()[1]["addr"]
third_addr = account.list_transports()[2]["addr"]
# test that adding a transport after unpublishing one works again
account.set_transport_unpublished(second_addr)
account.add_transport_from_qr(qr)
with pytest.raises(JsonRpcError):
account.add_transport_from_qr(qr)
# UIs are not expected to delete transports directly,
# but we still test that adding a transport
# after deleting one instead of unpublishing works.
account.delete_transport(third_addr)
account.delete_transport(second_addr)
account.add_transport_from_qr(qr)
with pytest.raises(JsonRpcError):
account.add_transport_from_qr(qr)
@@ -305,7 +298,6 @@ def test_remove_primary_transport(acf, log) -> None:
log.section("Alice sets up second transport")
[transport1, transport2] = alice.list_transports()
alice.set_config("configured_addr", transport2["addr"])
bob_chat.send_text("Hello!")
msg1 = alice.wait_for_incoming_msg().get_snapshot()
@@ -313,6 +305,7 @@ def test_remove_primary_transport(acf, log) -> None:
log.section("Alice removes the primary relay")
alice.delete_transport(transport1["addr"])
assert alice.get_config("configured_addr") == transport2["addr"]
alice.stop_io()
alice.start_io()

5
docs/schema.sql
View File

@@ -626,13 +626,10 @@ CREATE TABLE transports (
-- over this table and `removed_transports`, and contacts keep the newest one.
add_timestamp INTEGER NOT NULL DEFAULT 0,
-- True if the transport address is published
-- by sending it in the public key signature.
-- Unused since migration 165, which removed unpublished transports.
is_published INTEGER DEFAULT 1 NOT NULL,
-- Time when the transport was last used to receive a message.
-- Used to remove the least recently used transport
-- when a new transport is added and there are too many relays already.
last_rcvd_timestamp INTEGER NOT NULL DEFAULT 0,
UNIQUE(addr)
);

3
src/autorelay/autorelay_tests.rs
View File

@@ -139,8 +139,7 @@ async fn test_maybe_add_additional_relays_does_nothing_after_finishing_once() ->
assert!(relay_added);
let transports = t.list_transports().await?;
t.delete_transport(&transports.last().unwrap().param.addr)
.await?;
t.delete_transport(&transports.last().unwrap().addr).await?;
SystemTime::shift(Duration::from_secs(
AUTOMATIC_ADDITION_DEBOUNCE_SECONDS as u64 + 1,

55
src/config.rs
View File

@@ -19,7 +19,7 @@ use crate::log::LogExt;
use crate::mimefactory::RECOMMENDED_FILE_SIZE;
use crate::sync::{self, Sync::*, SyncData};
use crate::tools::{get_abs_path, time};
use crate::transport::{add_pseudo_transport, send_sync_transports};
use crate::transport::{add_pseudo_transport, send_sync_transports, transport_addrs};
use crate::{constants, stats};
/// The available configuration keys.
@@ -794,10 +794,6 @@ impl Context {
(addr,),
)?;
// `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
@@ -805,7 +801,7 @@ impl Context {
transaction
.execute(
"UPDATE transports
SET add_timestamp=MAX(?, add_timestamp+1), is_published=1
SET add_timestamp=MAX(?, add_timestamp+1)
WHERE addr=?",
(time(), addr),
)
@@ -915,7 +911,7 @@ impl Context {
return Ok(true);
}
Ok(self
.get_all_self_addrs()
.get_self_addrs()
.await?
.iter()
.any(|a| addr_cmp(addr, a)))
@@ -937,49 +933,10 @@ impl Context {
}
/// Returns all self addresses, newest first.
pub(crate) async fn get_all_self_addrs(&self) -> Result<Vec<String>> {
pub(crate) async fn get_self_addrs(&self) -> Result<Vec<String>> {
let query_only = true;
self.sql
.query_map_vec(
"SELECT addr FROM transports ORDER BY add_timestamp DESC, id DESC",
(),
|row| {
let addr: String = row.get(0)?;
Ok(addr)
},
)
.await
}
/// Returns all published self addresses, newest first.
/// See `[Context::set_transport_unpublished]`
pub(crate) async fn get_published_self_addrs(&self) -> Result<Vec<String>> {
self.sql
.query_map_vec(
"SELECT addr FROM transports WHERE is_published=1 ORDER BY add_timestamp DESC, id DESC",
(),
|row| {
let addr: String = row.get(0)?;
Ok(addr)
},
)
.await
}
/// Returns all published secondary self addresses.
/// See `[Context::set_transport_unpublished]`
pub(crate) async fn get_published_secondary_self_addrs(&self) -> Result<Vec<String>> {
self.sql
.query_map_vec(
"SELECT addr FROM transports
WHERE is_published
AND addr NOT IN (SELECT value FROM config WHERE keyname='configured_addr')
ORDER BY add_timestamp DESC, id DESC",
(),
|row| {
let addr: String = row.get(0)?;
Ok(addr)
},
)
.transaction_ext(query_only, |transaction| transport_addrs(transaction))
.await
}

283
src/configure.rs
View File

@@ -18,6 +18,7 @@ use deltachat_contact_tools::{EmailAddress, addr_normalize};
use futures::FutureExt;
use futures_lite::FutureExt as _;
use percent_encoding::utf8_percent_encode;
use rusqlite::OptionalExtension;
use server_params::{ServerParams, expand_param_vector};
use tokio::task;
@@ -26,8 +27,8 @@ use crate::constants::NON_ALPHANUMERIC_WITHOUT_DOT;
use crate::context::Context;
use crate::imap::Imap;
use crate::log::warn;
use crate::login_param::EnteredCertificateChecks;
pub use crate::login_param::EnteredLoginParam;
use crate::login_param::{EnteredCertificateChecks, TransportListEntry};
use crate::net::proxy::ProxyConfig;
use crate::provider::{self, Protocol, Socket};
use crate::qr::{login_param_from_account_qr, login_param_from_login_qr};
@@ -36,7 +37,8 @@ use crate::sync::Sync::Nosync;
use crate::tools::time;
use crate::transport::{
ConfiguredCertificateChecks, ConfiguredLoginParam, ConfiguredServerLoginParam,
ConnectionCandidate, send_sync_transports,
ConnectionCandidate, delete_transport_row, maybe_update_sending_transport,
purge_transport_caches, send_sync_transports, transport_addrs,
};
use crate::{EventType, stock_str};
@@ -106,7 +108,6 @@ impl Context {
/// from a server encoded in a QR code.
/// - [Self::list_transports()] to get a list of all configured transports.
/// - [Self::delete_transport()] to remove a transport.
/// - [Self::set_transport_unpublished()] to set whether contacts see this transport.
pub async fn add_or_update_transport(&self, param: &mut EnteredLoginParam) -> Result<()> {
self.stop_io().await;
let result = self.add_transport_inner(param).await;
@@ -192,25 +193,13 @@ impl Context {
/// Returns the list of all email accounts that are used as a transport in the current profile.
/// Use [Self::add_or_update_transport()] to add or change a transport
/// and [Self::delete_transport()] to delete a transport.
pub async fn list_transports(&self) -> Result<Vec<TransportListEntry>> {
let transports = self
.sql
.query_map_vec(
"SELECT entered_param, is_published FROM transports",
(),
|row| {
let param: String = row.get(0)?;
let param: EnteredLoginParam = serde_json::from_str(&param)?;
let is_published: bool = row.get(1)?;
Ok(TransportListEntry {
param,
is_unpublished: !is_published,
})
},
)
.await?;
Ok(transports)
pub async fn list_transports(&self) -> Result<Vec<EnteredLoginParam>> {
self.sql
.query_map_vec("SELECT entered_param FROM transports", (), |row| {
let param: String = row.get(0)?;
Ok(serde_json::from_str(&param)?)
})
.await
}
/// Returns the number of configured transports.
@@ -218,101 +207,51 @@ impl Context {
self.sql.count("SELECT COUNT(*) FROM transports", ()).await
}
/// Immediately deletes a transport, potentially causing messages not to arrive.
/// This must ONLY be used internally and by the automated tests.
/// UI implementations must use [`Self::set_transport_unpublished`] instead.
/// Removes a transport.
/// UIs should call this function when the user removes a relay.
///
/// The last transport cannot be removed.
/// If the removed transport was the one used for sending,
/// another one is chosen automatically.
pub async fn delete_transport(&self, addr: &str) -> Result<()> {
let now = time();
let removed_transport_id = self
let (removed_transport_id, reelected) = self
.sql
.transaction(|transaction| {
let primary_addr = transaction.query_row(
"SELECT value FROM config WHERE keyname='configured_addr'",
(),
|row| {
let addr: String = row.get(0)?;
Ok(addr)
},
)?;
if primary_addr == addr {
bail!("Cannot delete primary transport");
if transport_addrs(transaction)?.len() <= 1 {
bail!("Cannot remove the last transport");
}
let (transport_id, add_timestamp) = transaction.query_row(
"DELETE FROM transports WHERE addr=? RETURNING id, add_timestamp",
(addr,),
|row| {
let id: u32 = row.get(0)?;
let add_timestamp: i64 = row.get(1)?;
Ok((id, add_timestamp))
},
)?;
let add_timestamp: i64 = transaction
.query_row(
"SELECT add_timestamp FROM transports WHERE addr=?",
(addr,),
|row| row.get(0),
)
.optional()?
.context("Transport does not exist")?;
// Removal timestamp should not be lower than addition timestamp
// to be accepted by other devices when synced.
let remove_timestamp = std::cmp::max(now, add_timestamp);
transaction.execute(
"INSERT INTO removed_transports (addr, remove_timestamp)
VALUES (?, ?)
ON CONFLICT (addr)
DO UPDATE SET remove_timestamp = excluded.remove_timestamp",
(addr, remove_timestamp),
)?;
Ok(transport_id)
let transport_id = delete_transport_row(transaction, addr, remove_timestamp)?
.context("Transport disappeared")?;
let reelected = maybe_update_sending_transport(transaction)?;
Ok((transport_id, reelected))
})
.await?;
if let Some(new_addr) = reelected {
info!(self, "Using transport {new_addr:?} for sending now.");
self.sql.uncache_raw_config("configured_addr").await;
}
send_sync_transports(self).await?;
self.quota.write().await.remove(&removed_transport_id);
purge_transport_caches(self, removed_transport_id).await;
// Restarting all IO also stops the removed transport's IMAP loop.
// Scheduler reconciliation would stop only that one loop,
// see https://github.com/chatmail/core/issues/8513
self.restart_io_if_running().await;
Ok(())
}
/// Change whether the transport is unpublished.
/// UIs should call this function when the user clicks on "Remove".
/// Core will keep listening on this transport for some time,
/// and automatically remove it once it is no longer needed.
///
/// Unpublished transports are not advertised to contacts,
/// and self-sent messages are not sent there,
/// so that we don't cause extra messages to the corresponding inbox,
/// but can still receive messages from contacts who don't know our new transport addresses yet.
///
/// When more transports are added by [`Self::add_or_update_transport()`] or [`Self::add_transport_from_qr`],
/// the least recently needed unpublished transport is automatically removed
/// if this is necessary in order to stay below the maximum number of allowed relays.
/// Also, unpublished transports that are not used to receive any new messages for a time defined by
/// `UNPUBLISHED_TRANSPORT_KEEP_TIME` are automatically removed.
pub async fn set_transport_unpublished(&self, addr: &str, unpublished: bool) -> Result<()> {
self.sql
.transaction(|trans| {
let primary_addr: String = trans
.query_row(
"SELECT value FROM config WHERE keyname='configured_addr'",
(),
|row| row.get(0),
)
.context("Select primary address")?;
if primary_addr == addr && unpublished {
bail!("Can't set primary relay as unpublished");
}
// We need to update the timestamp so that the key's timestamp changes
// and is recognized as newer by our peers
trans
.execute(
"UPDATE transports SET is_published=?, add_timestamp=? WHERE addr=? AND is_published!=?1",
(!unpublished, time(), addr),
)
.context("Update transports")?;
Ok(())
})
.await?;
send_sync_transports(self).await?;
Ok(())
}
async fn inner_configure(&self, param: &EnteredLoginParam) -> Result<()> {
info!(self, "Configure ...");
@@ -324,7 +263,7 @@ impl Context {
)
.await?
{
self.try_make_space_for_new_relay().await?;
self.check_relay_limit().await?;
}
let skip_network = false;
@@ -350,39 +289,11 @@ impl Context {
Ok(())
}
/// This function is called before adding a new relay.
/// If the maximum number of relays ([`MAX_RELAYS`]) is already reached,
/// then it tries to make space by removing an unpublished relay.
/// If there are multiple unpublished relays,
/// the one that hasn't received a message for longest is removed.
/// If there are no unpublished relays, an error is returned.
///
/// Note that eviction happens before we know that a new relay works,
/// which is a trade-off we made in favor of implementation complexity.
async fn try_make_space_for_new_relay(&self) -> Result<()> {
if self.count_transports().await? >= MAX_RELAYS {
// Try to automatically remove the unpublished transport that wasn't used for the longest time:
if let Some(addr) = self
.sql
.query_get_value::<String>(
"SELECT addr FROM transports WHERE is_published=0
ORDER BY last_rcvd_timestamp, add_timestamp LIMIT 1",
(),
)
.await?
{
info!(
self,
"Auto-deleting relay {addr} to make space for new relay."
);
self.delete_transport(&addr).await?;
}
if self.count_transports().await? >= MAX_RELAYS {
// Apparently, all the transports are published
bail!("You have reached the maximum number of relays ({MAX_RELAYS})");
}
};
async fn check_relay_limit(&self) -> Result<()> {
ensure!(
self.count_transports().await? < MAX_RELAYS,
"You have reached the maximum number of relays ({MAX_RELAYS})"
);
Ok(())
}
}
@@ -746,13 +657,10 @@ pub enum Error {
#[cfg(test)]
mod tests {
use crate::tools::SystemTime;
use super::*;
use crate::autorelay::login_param_from_host;
use crate::config::Config;
use crate::login_param::EnteredImapLoginParam;
use crate::sql::update_transport_last_rcvd_timestamp;
use crate::test_utils::{TestContext, TestContextManager};
use crate::transport::add_pseudo_transport;
@@ -814,104 +722,27 @@ mod tests {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_try_make_place_for_new_relay() -> Result<()> {
let t = TestContext::new().await;
async fn test_relay_limit() -> Result<()> {
let mut tcm = TestContextManager::new();
let t = &tcm.unconfigured().await;
// Setting ConfiguredAddr on an unconfigured account creates a pseudo primary transport
// Setting ConfiguredAddr on an unconfigured account creates a pseudo transport
t.set_config(Config::ConfiguredAddr, Some("primary@example.org"))
.await?;
// Test that try_make_place_for_new_relay() doesn't do anything when we're below the limit
assert_eq!(t.count_transports().await?, 1);
t.try_make_space_for_new_relay().await?;
assert_eq!(t.count_transports().await?, 1);
t.check_relay_limit().await?;
for i in 0..(MAX_RELAYS - 2) {
add_pseudo_transport(&t, &format!("transport{i}@example.org")).await?;
for i in 0..(MAX_RELAYS - 1) {
add_pseudo_transport(t, &format!("transport{i}@example.org")).await?;
}
assert_eq!(t.count_transports().await?, MAX_RELAYS - 1);
t.try_make_space_for_new_relay().await?;
assert_eq!(t.count_transports().await?, MAX_RELAYS - 1);
// Test that try_make_place_for_new_relay() removes the unpublished transport
// when we're at the limit
add_pseudo_transport(&t, "unpublished@example.org").await?;
t.set_transport_unpublished("unpublished@example.org", true)
.await?;
assert_eq!(t.count_transports().await?, MAX_RELAYS);
t.try_make_space_for_new_relay().await?;
assert_eq!(t.count_transports().await?, MAX_RELAYS - 1);
assert_eq!(
t.sql
.exists(
"SELECT COUNT(*) FROM transports WHERE addr=?",
("unpublished@example.org",),
)
.await?,
false
t.check_relay_limit().await.unwrap_err().to_string(),
format!("You have reached the maximum number of relays ({MAX_RELAYS})")
);
// Test that if there are multiple unpublished relays,
// the one that was used least recently is removed
t.set_transport_unpublished("transport0@example.org", true)
.await?;
add_pseudo_transport(&t, "other_unpublished@example.org").await?;
t.set_transport_unpublished("other_unpublished@example.org", true)
.await?;
assert_eq!(t.count_transports().await?, MAX_RELAYS);
let transport0_id: u32 = t
.sql
.query_get_value(
"SELECT id FROM transports WHERE addr=?",
("transport0@example.org",),
)
.await?
.unwrap();
let other_unpublished_id: u32 = t
.sql
.query_get_value(
"SELECT id FROM transports WHERE addr=?",
("other_unpublished@example.org",),
)
.await?
.unwrap();
update_transport_last_rcvd_timestamp(&t, transport0_id).await?;
SystemTime::shift(std::time::Duration::from_secs(10));
update_transport_last_rcvd_timestamp(&t, other_unpublished_id).await?;
// Test that try_make_place_for_new_relay()
// removes the relay with the oldest last_rcvd_timestamp
t.try_make_space_for_new_relay().await?;
assert_eq!(t.count_transports().await?, MAX_RELAYS - 1);
assert_eq!(
t.sql
.exists(
"SELECT COUNT(*) FROM transports WHERE addr=?",
("transport0@example.org",),
)
.await?,
false
);
assert_eq!(
t.sql
.exists(
"SELECT COUNT(*) FROM transports WHERE addr=?",
("other_unpublished@example.org",),
)
.await?,
true
);
// Test that try_make_place_for_new_relay() fails
// if there are MAX_RELAYS published transports
add_pseudo_transport(&t, "published_extra@example.org").await?;
t.set_transport_unpublished("other_unpublished@example.org", false)
.await?;
assert_eq!(t.count_transports().await?, MAX_RELAYS);
assert!(t.try_make_space_for_new_relay().await.is_err());
assert_eq!(t.count_transports().await?, MAX_RELAYS);
t.delete_transport("transport0@example.org").await?;
t.check_relay_limit().await?;
Ok(())
}

2
src/contact.rs
View File

@@ -1174,7 +1174,7 @@ VALUES (?, ?, ?, ?, ?, ?)
query: Option<&str>,
) -> Result<Vec<ContactId>> {
let self_addrs = context
.get_all_self_addrs()
.get_self_addrs()
.await?
.into_iter()
.collect::<HashSet<_>>();

6
src/contact/contact_tests.rs
View File

@@ -153,12 +153,8 @@ async fn test_get_contacts() -> Result<()> {
let contacts = Contact::get_all(&context, 0, Some("δ")).await?;
assert_eq!(contacts.len(), 1);
// Searching for a secondary self address finds "Me",
// even if the transport is unpublished.
// Searching for another self address finds "Me".
crate::transport::add_pseudo_transport(&context, "bob@second.example").await?;
context
.set_transport_unpublished("bob@second.example", true)
.await?;
let contacts = Contact::get_all(
&context,
constants::DC_GCL_ADD_SELF,

8
src/context.rs
View File

@@ -327,9 +327,9 @@ pub struct InnerContext {
/// Mutex is also held while generating the key to avoid generating the key twice.
pub(crate) self_public_key: Mutex<Option<SignedPublicKey>>,
/// `Connectivity` values for published relays, unordered. Used to compute the aggregate connectivity,
/// `Connectivity` values for the relays, unordered. Used to compute the aggregate connectivity,
/// see [`Context::get_connectivity()`].
pub(crate) published_connectivities: parking_lot::Mutex<Vec<ConnectivityStore>>,
pub(crate) connectivities: parking_lot::Mutex<Vec<ConnectivityStore>>,
/// Timestamp after which the SMTP loop checks for a keyupdate to send, or 0 if none is due.
pub(crate) next_keyupdate_check: AtomicI64,
@@ -508,7 +508,7 @@ impl Context {
iroh: Arc::new(RwLock::new(None)),
self_fingerprint: OnceLock::new(),
self_public_key: Mutex::new(None),
published_connectivities: parking_lot::Mutex::new(Vec::new()),
connectivities: parking_lot::Mutex::new(Vec::new()),
next_keyupdate_check: AtomicI64::new(0),
};
@@ -841,7 +841,7 @@ impl Context {
let all_transports: Vec<String> = ConfiguredLoginParam::load_all(self)
.await?
.into_iter()
.map(|(transport_id, param, _)| format!("{transport_id}: {param}"))
.map(|(transport_id, param)| format!("{transport_id}: {param}"))
.collect();
let all_transports = if all_transports.is_empty() {
"Not configured".to_string()

6
src/events/payload.rs
View File

@@ -433,9 +433,9 @@ pub enum EventType {
///
/// UI should update the list.
///
/// This event is emitted when a transport
/// synchronization message modifies transports,
/// but not when the UI modifies the transport list by itself.
/// The event is emitted on the device modifying
/// the transports as well as on other devices
/// applying the synced change.
TransportsModified,
/// Event for using in tests, e.g. as a fence between normally generated events.

2
src/key.rs
View File

@@ -302,7 +302,7 @@ pub(crate) async fn load_self_public_key_opt(context: &Context) -> Result<Option
.await?
.context("No transports configured")?;
let addr = context.get_primary_self_addr().await?;
let all_addrs = context.get_published_self_addrs().await?.join(",");
let all_addrs = context.get_self_addrs().await?.join(",");
let signed_public_key =
secret_key_to_public_key(context, signed_secret_key, timestamp, &addr, &all_addrs)?;
*lock = Some(signed_public_key.clone());

14
src/keyupdate.rs
View File

@@ -155,9 +155,9 @@ fn envelope_recipients(chunk: &[KeyupdateRecipient]) -> String {
Vec::from_iter(addrs).join(" ")
}
/// Returns the published relay list in the format stored in [`Config::KeyupdateBaseline`].
async fn published_relays_joined(context: &Context) -> Result<String> {
let mut relays = context.get_published_self_addrs().await?;
/// Returns the relay list in the format stored in [`Config::KeyupdateBaseline`].
async fn relays_joined(context: &Context) -> Result<String> {
let mut relays = context.get_self_addrs().await?;
relays.sort();
Ok(relays.join(" "))
}
@@ -171,17 +171,17 @@ pub(crate) async fn schedule_keyupdate_check(context: &Context) -> Result<()> {
Ok(())
}
/// Records the currently published relay list as not needing a keyupdate, see the module docs.
/// Records the current relay list as not needing a keyupdate, see the module docs.
pub(crate) async fn set_current_relays_as_keyupdate_baseline(context: &Context) -> Result<()> {
let current = published_relays_joined(context).await?;
let current = relays_joined(context).await?;
context
.set_config_internal(Config::KeyupdateBaseline, Some(&current))
.await
}
/// Sends a keyupdate message if the published relay list differs from the recorded baseline.
/// Sends a keyupdate message if the relay list differs from the recorded baseline.
pub(crate) async fn maybe_send_keyupdate_message(context: &Context) -> Result<()> {
let current = published_relays_joined(context).await?;
let current = relays_joined(context).await?;
let last = context.get_config(Config::KeyupdateBaseline).await?;
if last.unwrap_or_default() == current {
return Ok(());

37
src/keyupdate/keyupdate_tests.rs
View File

@@ -234,14 +234,12 @@ async fn test_send_and_receive_keyupdate() -> Result<()> {
let bob_message = bob.send_text(bob_chat_id, "hi").await;
assert!(bob_message.recipients.contains("alice@relay.example.net"));
// The removal direction: unpublishing the relay sends a keyupdate
// The removal direction: deleting the relay sends a keyupdate
// whose list no longer contains it, so Bob stops sending there.
// The time shift gives the re-signed key a later signature timestamp,
// so that certificate merging prefers the removal.
SystemTime::shift(Duration::from_secs(2));
alice
.set_transport_unpublished("alice@relay.example.net", true)
.await?;
alice.delete_transport("alice@relay.example.net").await?;
maybe_send_keyupdate_message(alice).await?;
bob.recv_msg_trash(&alice.pop_sent_msg().await).await;
let bob_message = bob.send_text(bob_chat_id, "hi again").await;
@@ -250,6 +248,33 @@ async fn test_send_and_receive_keyupdate() -> Result<()> {
Ok(())
}
/// Tests the keyupdate sent when the only relay is replaced.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_keyupdate_on_replacement() -> Result<()> {
let mut tcm = TestContextManager::new();
let alice = &tcm.alice().await;
let bob = &tcm.bob().await;
send_text_message(alice, bob).await;
alice.add_transport("alice@relay.example.net").await;
SystemTime::shift(Duration::from_secs(2));
alice.delete_transport("alice@example.org").await?;
assert_eq!(
alice.get_config(Config::ConfiguredAddr).await?.as_deref(),
Some("alice@relay.example.net")
);
maybe_send_keyupdate_message(alice).await?;
let keyupdate = alice.pop_sent_msg().await;
assert_eq!(keyupdate.recipients, "bob@example.net");
let self_key = crate::key::load_self_public_key(alice).await?;
let addrs = addresses_from_public_key(&self_key);
assert_eq!(addrs, Some(vec!["alice@relay.example.net".to_string()]));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_keyupdate_trigger_dedup() -> Result<()> {
let mut tcm = TestContextManager::new();
@@ -303,8 +328,8 @@ async fn test_keyupdate_not_sent_by_synced_device() -> Result<()> {
alice.send_sync_msg().await?;
alice2.recv_msg_trash(&alice.pop_sent_msg().await).await;
// The sync was applied, so silence below is meaningful.
let published = alice2.get_published_self_addrs().await?;
assert!(published.contains(&"alice@relay.example.net".to_string()));
let addrs = alice2.get_self_addrs().await?;
assert!(addrs.contains(&"alice@relay.example.net".to_string()));
maybe_send_keyupdate_message(alice2).await?;
assert!(alice2.pop_sent_msg_opt().await.is_none());

10
src/login_param.rs
View File

@@ -115,16 +115,6 @@ pub struct EnteredSmtpLoginParam {
pub password: String,
}
/// A transport, as shown in the "relays" list in the UI.
#[derive(Debug)]
pub struct TransportListEntry {
/// The login data entered by the user.
pub param: EnteredLoginParam,
/// Whether this transport is set to 'unpublished'.
/// See [`Context::set_transport_unpublished`] for details.
pub is_unpublished: bool,
}
/// Login parameters entered by the user.
#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EnteredLoginParam {

2
src/pgp.rs
View File

@@ -439,7 +439,7 @@ pub fn merge_openpgp_certificates(
/// Returns relays addresses from the public key signature.
///
/// Not more than [`MAX_RELAYS`] relays are returned for each key.
/// This is the same constant as the maximum number of published relays
/// This is the same constant as the maximum number of relays
/// the user is allowed to have in the key.
/// If the constant is changed in the future,
/// the client with the lower constant value

8
src/scheduler.rs
View File

@@ -325,9 +325,6 @@ struct SchedBox {
/// IMAP loop task handle.
handle: task::JoinHandle<()>,
/// Relay published status.
is_published: bool,
}
/// Job and connection scheduler.
@@ -680,9 +677,7 @@ impl Scheduler {
let mut inboxes = Vec::new();
let mut start_recvs = Vec::new();
for (transport_id, configured_login_param, is_published) in
ConfiguredLoginParam::load_all(ctx).await?
{
for (transport_id, configured_login_param) in ConfiguredLoginParam::load_all(ctx).await? {
let (conn_state, inbox_handlers) =
ImapConnectionState::new(ctx, transport_id, configured_login_param.clone()).await?;
let (inbox_start_send, inbox_start_recv) = oneshot::channel();
@@ -699,7 +694,6 @@ impl Scheduler {
folder,
conn_state,
handle,
is_published,
};
inboxes.push(inbox);
start_recvs.push(inbox_start_recv);

41
src/scheduler/connectivity.rs
View File

@@ -255,7 +255,7 @@ impl Context {
///
/// If the connectivity changes, a DC_EVENT_CONNECTIVITY_CHANGED will be emitted.
pub fn get_connectivity(&self) -> Connectivity {
let stores: Vec<ConnectivityStore> = self.published_connectivities.lock().clone();
let stores: Vec<ConnectivityStore> = self.connectivities.lock().clone();
let connectivities: Vec<Connectivity> = stores.into_iter().map(|s| s.get_basic()).collect();
combine_connectivities(&connectivities)
}
@@ -264,12 +264,11 @@ impl Context {
let stores: Vec<_> = match sched {
InnerSchedulerState::Started(sched) => sched
.boxes()
.filter(|b| b.is_published)
.map(|b| b.conn_state.state.connectivity.clone())
.collect(),
_ => Vec::new(),
};
*self.published_connectivities.lock() = stores;
*self.connectivities.lock() = stores;
}
/// Get an overview of the current connectivity, and possibly more statistics.
@@ -331,9 +330,6 @@ impl Context {
.transport {
margin-bottom: 1em;
}
.unpublished {
opacity: 0.5;
}
.quota-list {
padding-left: 0;
}
@@ -397,28 +393,19 @@ impl Context {
let transports = self
.sql
.query_map_vec(
"SELECT id, addr, is_published FROM transports ORDER BY is_published DESC, id",
(),
|row| {
let transport_id: u32 = row.get(0)?;
let addr: String = row.get(1)?;
let is_published: bool = row.get(2)?;
Ok((transport_id, addr, is_published))
},
)
.query_map_vec("SELECT id, addr FROM transports ORDER BY id", (), |row| {
let transport_id: u32 = row.get(0)?;
let addr: String = row.get(1)?;
Ok((transport_id, addr))
})
.await?;
let quota = self.quota.read().await;
for (transport_id, transport_addr, is_published) in transports {
for (transport_id, transport_addr) in transports {
let domain = &deltachat_contact_tools::EmailAddress::new(&transport_addr)
.map_or(transport_addr.clone(), |email| email.domain);
let domain_escaped = escaper::encode_minimal(domain);
ret += if is_published {
"<li class=\"transport\">"
} else {
"<li class=\"transport unpublished\">"
};
ret += "<li class=\"transport\">";
let folders = folders_states
.iter()
.filter(|(folder_addr, ..)| *folder_addr == transport_addr);
@@ -428,18 +415,10 @@ impl Context {
ret += " <b>";
ret += &*domain_escaped;
ret += ":</b> ";
if is_published {
ret += &*escaper::encode_minimal(&detailed.to_string_imap(self));
} else {
ret += &*escaper::encode_minimal(&stock_str::phasing_out(self));
}
ret += &*escaper::encode_minimal(&detailed.to_string_imap(self));
ret += "<br />";
}
if !is_published {
ret += "</li>"; // quota is of no big interest for unpublished relays
continue;
};
let Some(quota) = quota.get(&transport_id) else {
ret += "</li>";
continue;

3
src/securejoin.rs
View File

@@ -141,9 +141,10 @@ pub async fn get_securejoin_qr(context: &Context, chat: Option<ChatId>) -> Resul
let self_addr_urlencoded = utf8_percent_encode(&self_addr, DISALLOWED_CHARACTERS).to_string();
let r_param = context
.get_published_secondary_self_addrs()
.get_self_addrs()
.await?
.into_iter()
.filter(|addr| *addr != self_addr)
.reduce(|acc, addr| {
format!(
"{acc},{}",

8
src/smtp.rs
View File

@@ -733,9 +733,12 @@ pub(crate) async fn add_self_recipients(
// Avoid sending unencrypted messages to all transports, chatmail relays won't accept
// them. Normally the user should have a non-chatmail primary transport to send unencrypted
// messages.
let from = context.get_primary_self_addr().await?;
if encrypted {
for addr in context.get_published_secondary_self_addrs().await? {
recipients.push(addr);
for addr in context.get_self_addrs().await? {
if addr != from {
recipients.push(addr);
}
}
}
// `from` must be the last addr
@@ -744,7 +747,6 @@ pub(crate) async fn add_self_recipients(
// This helps with marking messages as delivered
// if the server is slow and we never get an `OK` response
// before the connection times out.
let from = context.get_primary_self_addr().await?;
recipients.push(from);
Ok(())

32
src/sql.rs
View File

@@ -47,9 +47,6 @@ mod pool;
use pool::{Pool, WalCheckpointStats};
/// How long a hidden and unused transport should be kept in the database before being deleted.
pub const UNPUBLISHED_TRANSPORT_KEEP_TIME: i64 = 90 * 24 * 60 * 60;
/// A wrapper around the underlying Sqlite3 object.
#[derive(Debug)]
pub struct Sql {
@@ -906,12 +903,6 @@ pub async fn housekeeping(context: &Context) -> Result<()> {
.log_err(context)
.ok();
remove_unused_hidden_transports(context)
.await
.context("Failed to remove unused hidden transports")
.log_err(context)
.ok();
remove_old_pending_reactions(context)
.await
.context("Failed to remove old pending reactions")
@@ -934,28 +925,7 @@ async fn remove_old_pending_reactions(context: &Context) -> Result<usize> {
.await
}
/// Removes transports that are hidden (`is_published=0`),
/// and haven't been used to receive new messages for [`UNPUBLISHED_TRANSPORT_KEEP_TIME`] seconds.
pub(crate) async fn remove_unused_hidden_transports(context: &Context) -> Result<usize> {
let now = time();
let cutoff = now.saturating_sub(UNPUBLISHED_TRANSPORT_KEEP_TIME);
context
.sql
.execute(
"DELETE FROM transports
WHERE is_published=0
AND last_rcvd_timestamp<?1
AND add_timestamp<?1", // important, prevents immediate deletion in case of `last_rcvd_timestamp=0`
(cutoff,),
)
.await
}
/// Updates transport's `last_rcvd_timestamp`
/// with the current time.
///
/// This is used to postpone deletion of hidden transport by [`remove_unused_hidden_transports`],
/// if it is still used to receive messages.
/// Updates the transport's `last_rcvd_timestamp` with the current time.
pub(crate) async fn update_transport_last_rcvd_timestamp(
context: &Context,
transport_id: u32,

19
src/sql/migrations.rs
View File

@@ -2623,6 +2623,25 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed.
.await?;
}
inc_and_check(&mut migration_version, 165)?;
if dbversion < migration_version {
// Remove any unpublished relays and cause keyupdates.
sql.execute_migration(
"INSERT INTO removed_transports (addr, remove_timestamp)
SELECT addr, MAX(add_timestamp, unixepoch()) FROM transports
WHERE is_published=0
AND addr!=(SELECT value FROM config WHERE keyname='configured_addr')
ON CONFLICT (addr) DO UPDATE SET
remove_timestamp=MAX(excluded.remove_timestamp, remove_timestamp);
DELETE FROM transports
WHERE is_published=0
AND addr!=(SELECT value FROM config WHERE keyname='configured_addr');
DELETE FROM config WHERE keyname='keyupdate_baseline' AND changes()>0",
migration_version,
)
.await?;
}
let new_version = sql
.get_raw_config_int(VERSION_CFG)
.await?

51
src/sql/migrations/migrations_tests.rs
View File

@@ -35,17 +35,8 @@ async fn test_keyupdate_baseline_migration() -> Result<()> {
let configured = STOP_MIGRATIONS_AT
.scope(163, async move { TestContext::new_alice().await })
.await;
// An address sorting before the primary pins the seed's ORDER BY,
// an unpublished transport pins its filter.
// An address sorting before the existing one pins the seed's ORDER BY.
add_pseudo_transport(&configured, "aa@example.org").await?;
add_pseudo_transport(&configured, "unpublished@example.org").await?;
configured
.sql
.execute(
"UPDATE transports SET is_published=0 WHERE addr='unpublished@example.org'",
(),
)
.await?;
configured.sql.run_migrations(&configured).await?;
let relays = configured.get_config(Config::KeyupdateBaseline).await?;
assert_eq!(relays.as_deref(), Some("aa@example.org alice@example.org"));
@@ -53,6 +44,46 @@ async fn test_keyupdate_baseline_migration() -> Result<()> {
Ok(())
}
/// Tests that upgrading removes unpublished transports.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_unpublished_transport_migration_165() -> Result<()> {
let t = STOP_MIGRATIONS_AT
.scope(163, async move { TestContext::new_alice().await })
.await;
let skewed = tools::time() + 3600;
t.sql
.execute(
"INSERT INTO transports (addr, entered_param, configured_param, is_published, add_timestamp)
VALUES ('unpublished@example.org', '', '', 0, ?)",
(skewed,),
)
.await?;
STOP_MIGRATIONS_AT
.scope(165, async { t.sql.run_migrations(&t).await })
.await?;
assert!(
!t.sql
.exists(
"SELECT COUNT(*) FROM transports WHERE addr='unpublished@example.org'",
(),
)
.await?
);
assert_eq!(
t.sql
.query_get_value(
"SELECT remove_timestamp FROM removed_transports WHERE addr='unpublished@example.org'",
(),
)
.await?,
Some(skewed)
);
assert_eq!(t.get_config(Config::KeyupdateBaseline).await?, None);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_key_contacts_migration_autocrypt() -> Result<()> {
let t = STOP_MIGRATIONS_AT

112
src/sql/sql_tests.rs
View File

@@ -1,6 +1,5 @@
use super::*;
use crate::message::Message;
use crate::tools::SystemTime;
use crate::{EventType, test_utils::TestContext};
#[test]
@@ -370,114 +369,3 @@ async fn test_incremental_vacuum() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_remove_unused_hidden_transports() -> Result<()> {
let t = TestContext::new().await;
let now = time();
let transport_id = t
.sql
.transaction(|t| {
// published transport
t.execute(
"INSERT INTO transports
(addr,
entered_param,
configured_param,
add_timestamp)
VALUES (?, ?, ?, ?)",
("1", "", "", now),
)?;
// not published transport
let transport_id = t.query_row::<u32, _, _>(
"INSERT INTO transports
(addr,
entered_param,
configured_param,
is_published,
add_timestamp)
VALUES (?, ?, ?, ?, ?)
RETURNING id",
("2", "", "", 0, now),
|row| row.get(0),
)?;
// this transport has add_timestamp in the future,
// and should not be removed at all, since we only shift time
// by 90 days and one second.
t.execute(
"INSERT INTO transports
(addr,
entered_param,
configured_param,
is_published,
add_timestamp)
VALUES (?, ?, ?, ?, ?)",
("3", "", "", 0, now + 60),
)?;
Ok(transport_id)
})
.await?;
async fn assert_transport_removed(
context: &Context,
hidden_transport_id: u32,
should_be_removed: bool,
) -> Result<()> {
let transports_count: u32 = context
.sql
.query_get_value("SELECT count(*) FROM transports", ())
.await?
.unwrap();
assert_eq!(transports_count, if should_be_removed { 2 } else { 3 });
let hidden_transport_present: bool = context
.sql
.query_get_value(
"SELECT count(*) FROM transports WHERE id=?",
(hidden_transport_id,),
)
.await?
.unwrap();
assert_eq!(hidden_transport_present, !should_be_removed);
Ok(())
}
assert_eq!(
t.sql
.query_get_value::<i64>(
"SELECT last_rcvd_timestamp FROM transports WHERE id=?",
(transport_id,)
)
.await?
.unwrap(),
0
);
update_transport_last_rcvd_timestamp(&t, transport_id).await?;
// last_rcvd_timestamp should update
assert!(
t.sql
.query_get_value::<i64>(
"SELECT last_rcvd_timestamp FROM transports WHERE id=?",
(transport_id,)
)
.await?
.unwrap()
>= now
);
assert_transport_removed(&t, transport_id, false).await?;
remove_unused_hidden_transports(&t).await?;
// it was recently used, so nothing is deleted
assert_transport_removed(&t, transport_id, false).await?;
SystemTime::shift(Duration::from_secs(
(UNPUBLISHED_TRANSPORT_KEEP_TIME + 1).try_into()?,
));
remove_unused_hidden_transports(&t).await?;
assert_transport_removed(&t, transport_id, true).await?;
Ok(())
}

8
src/stock_str.rs
View File

@@ -434,9 +434,6 @@ https://delta.chat/donate"))]
#[strum(props(fallback = "Message pinned by %1$s."))]
MsgMessagePinnedBy = 244,
#[strum(props(fallback = "Phasing out"))]
PhasingOut = 245,
}
impl StockMessage {
@@ -1168,11 +1165,6 @@ pub(crate) fn last_msg_sent_successfully(context: &Context) -> String {
translated(context, StockMessage::LastMsgSentSuccessfully)
}
/// Stock string: `Phasing out`.
pub(crate) fn phasing_out(context: &Context) -> String {
translated(context, StockMessage::PhasingOut)
}
/// Stock string: `Error: %1$s…`.
/// `%1$s` will be replaced by a possibly more detailed, typically english, error description.
pub(crate) fn error(context: &Context, error: &str) -> String {

4
src/sync.rs
View File

@@ -66,8 +66,8 @@ pub(crate) struct TransportData {
/// Timestamp of when the transport was last time (re)configured.
pub(crate) timestamp: i64,
/// Whether the transport is published.
/// See [`Context::set_transport_unpublished`] for details.
/// Whether the transport is advertised to contacts.
/// Always `true` from this core; an older core's `false` is applied as a removal.
pub(crate) is_published: bool,
}

2
src/test_utils.rs
View File

@@ -567,7 +567,7 @@ impl TestContext {
}
}
/// Adds a published transport for `addr` without any network activity.
/// Adds a transport for `addr` without any network activity.
pub async fn add_transport(&self, addr: &str) {
add_pseudo_transport(self, addr).await.unwrap();
// A fresh `add_timestamp` makes the re-signed self key newer than the copies

197
src/transport.rs
View File

@@ -13,6 +13,7 @@ use std::sync::atomic::Ordering;
use anyhow::{Context as _, Result, bail, format_err};
use deltachat_contact_tools::{EmailAddress, addr_normalize};
use rusqlite::OptionalExtension;
use serde::{Deserialize, Serialize};
use crate::config::Config;
@@ -290,21 +291,16 @@ impl ConfiguredLoginParam {
/// Loads configured login parameters for all transports.
///
/// Returns a vector of all transport IDs
/// paired with the configured parameters for the transports and the published state.
pub(crate) async fn load_all(context: &Context) -> Result<Vec<(u32, Self, bool)>> {
/// paired with the configured parameters for the transports.
pub(crate) async fn load_all(context: &Context) -> Result<Vec<(u32, Self)>> {
context
.sql
.query_map_vec(
"SELECT id, configured_param, is_published FROM transports",
(),
|row| {
let id: u32 = row.get(0)?;
let json: String = row.get(1)?;
let param = Self::from_json(&json)?;
let is_published: bool = row.get(2)?;
Ok((id, param, is_published))
},
)
.query_map_vec("SELECT id, configured_param FROM transports", (), |row| {
let id: u32 = row.get(0)?;
let json: String = row.get(1)?;
let param = Self::from_json(&json)?;
Ok((id, param))
})
.await
}
@@ -430,15 +426,7 @@ impl ConfiguredLoginParam {
entered_param: &EnteredLoginParam,
timestamp: i64,
) -> Result<()> {
let is_published = true;
save_transport(
context,
entered_param,
&self.into(),
timestamp,
is_published,
)
.await?;
save_transport(context, entered_param, &self.into(), timestamp).await?;
Ok(())
}
@@ -510,7 +498,6 @@ pub(crate) async fn save_transport(
entered_param: &EnteredLoginParam,
configured: &ConfiguredLoginParamJson,
add_timestamp: i64,
is_published: bool,
) -> Result<bool> {
ensure_and_debug_assert!(
configured
@@ -525,23 +512,20 @@ pub(crate) async fn save_transport(
let mut modified = context
.sql
.execute(
"INSERT INTO transports (addr, entered_param, configured_param, add_timestamp, is_published)
VALUES (?, ?, ?, ?, ?)
"INSERT INTO transports (addr, entered_param, configured_param, add_timestamp)
VALUES (?, ?, ?, ?)
ON CONFLICT (addr)
DO UPDATE SET entered_param=excluded.entered_param,
configured_param=excluded.configured_param,
add_timestamp=excluded.add_timestamp,
is_published=excluded.is_published
add_timestamp=excluded.add_timestamp
WHERE entered_param != excluded.entered_param
OR configured_param != excluded.configured_param
OR add_timestamp < excluded.add_timestamp
OR is_published != excluded.is_published",
OR add_timestamp < excluded.add_timestamp",
(
&addr,
serde_json::to_string(entered_param)?,
serde_json::to_string(configured)?,
add_timestamp,
is_published,
),
)
.await?
@@ -558,7 +542,8 @@ pub(crate) async fn save_transport(
Ok(modified)
}
/// Sends a sync message to synchronize transports across devices.
/// Sends a sync message to synchronize transports across devices
/// and emits [`EventType::TransportsModified`].
pub(crate) async fn send_sync_transports(context: &Context) -> Result<()> {
info!(context, "Sending transport synchronization message.");
@@ -578,7 +563,7 @@ pub(crate) async fn send_sync_transports(context: &Context) -> Result<()> {
let transports = context
.sql
.query_map_vec(
"SELECT entered_param, configured_param, add_timestamp, is_published
"SELECT entered_param, configured_param, add_timestamp
FROM transports WHERE id>1",
(),
|row| {
@@ -587,12 +572,11 @@ pub(crate) async fn send_sync_transports(context: &Context) -> Result<()> {
let configured_json: String = row.get(1)?;
let configured: ConfiguredLoginParamJson = serde_json::from_str(&configured_json)?;
let timestamp: i64 = row.get(2)?;
let is_published: bool = row.get(3)?;
Ok(TransportData {
configured,
entered,
timestamp,
is_published,
is_published: true,
})
},
)
@@ -618,11 +602,15 @@ pub(crate) async fn send_sync_transports(context: &Context) -> Result<()> {
// Schedule the check before interrupting, so the woken SMTP loop sees it.
schedule_keyupdate_check(context).await?;
context.scheduler.interrupt_smtp().await;
context.emit_event(EventType::TransportsModified);
Ok(())
}
/// Process received data for transport synchronization.
///
/// A transport that an older core unpublished
/// arrives with `is_published: false` and is removed.
pub(crate) async fn sync_transports(
context: &Context,
transports: &[TransportData],
@@ -636,39 +624,57 @@ pub(crate) async fn sync_transports(
is_published,
} in transports
{
modified |= save_transport(context, entered, configured, *timestamp, *is_published).await?;
if !is_published {
continue;
}
let removed_later = context
.sql
.exists(
"SELECT COUNT(*) FROM removed_transports WHERE addr=? AND remove_timestamp>=?",
(&addr_normalize(&configured.addr), timestamp),
)
.await?;
if removed_later {
// Only a legacy core keeps syncing a transport that was removed here;
// it must not be resurrected, and removal wins a timestamp tie.
continue;
}
modified |= save_transport(context, entered, configured, *timestamp).await?;
}
let reelected = context
let removals: Vec<(String, i64)> = removed_transports
.iter()
.map(|removed| (removed.addr.clone(), removed.timestamp))
.chain(
transports
.iter()
.filter(|data| !data.is_published)
.map(|data| (addr_normalize(&data.configured.addr), data.timestamp)),
)
.collect();
let (deleted_ids, reelected) = context
.sql
.transaction(|transaction| {
for RemovedTransportData { addr, timestamp } in removed_transports {
let count: i64 =
transaction
.query_row("SELECT COUNT(*) FROM transports", (), |row| row.get(0))?;
if count <= 1 {
let mut deleted_ids = Vec::new();
for (addr, timestamp) in &removals {
if transport_addrs(transaction)?.len() <= 1 {
// Removing the last transport would unconfigure the account.
break;
}
modified |= transaction.execute(
"DELETE FROM transports
WHERE addr=? AND add_timestamp<=?",
(addr, timestamp),
)? > 0;
transaction.execute(
"INSERT INTO removed_transports (addr, remove_timestamp)
VALUES (?, ?)
ON CONFLICT (addr) DO
UPDATE SET remove_timestamp = excluded.remove_timestamp
WHERE excluded.remove_timestamp > remove_timestamp",
(addr, timestamp),
)?;
deleted_ids.extend(delete_transport_row(transaction, addr, *timestamp)?);
}
modified |= !deleted_ids.is_empty();
maybe_reelect_local_primary(transaction)
let reelected = maybe_update_sending_transport(transaction)?;
Ok((deleted_ids, reelected))
})
.await?;
for transport_id in &deleted_ids {
purge_transport_caches(context, *transport_id).await;
}
if let Some(new_addr) = reelected {
info!(context, "Re-elected primary transport {new_addr:?}.");
context.sql.uncache_raw_config("configured_addr").await;
@@ -691,46 +697,71 @@ 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>> {
/// Returns the addresses of all transports, newest first.
pub(crate) fn transport_addrs(transaction: &rusqlite::Transaction) -> Result<Vec<String>> {
let addrs = transaction
.prepare("SELECT addr FROM transports ORDER BY add_timestamp DESC, id DESC")?
.query_map((), |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?;
Ok(addrs)
}
/// Deletes the transport row unless it is newer than `remove_timestamp`,
/// records the tombstone, and returns the deleted row's id.
pub(crate) fn delete_transport_row(
transaction: &mut rusqlite::Transaction,
addr: &str,
remove_timestamp: i64,
) -> Result<Option<u32>> {
let deleted_id: Option<u32> = transaction
.query_row(
"DELETE FROM transports
WHERE addr=? AND add_timestamp<=?
RETURNING id",
(addr, remove_timestamp),
|row| row.get(0),
)
.optional()?;
transaction.execute(
"INSERT INTO removed_transports (addr, remove_timestamp)
VALUES (?, ?)
ON CONFLICT (addr) DO
UPDATE SET remove_timestamp = excluded.remove_timestamp
WHERE excluded.remove_timestamp > remove_timestamp",
(addr, remove_timestamp),
)?;
Ok(deleted_id)
}
/// Drops the per-process caches of a removed transport.
pub(crate) async fn purge_transport_caches(context: &Context, transport_id: u32) {
context.quota.write().await.remove(&transport_id);
context.metadata.write().await.remove(&transport_id);
}
/// Elects another transport for sending if the current one vanished.
/// Any remaining transport works and selection is anyway moving
/// to the authority of the SMTP loop, see <https://github.com/chatmail/core/pull/8619>
pub(crate) fn maybe_update_sending_transport(
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)
{
let addrs = transport_addrs(transaction)?;
if addrs.contains(&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 {
let Some(new_addr) = addrs.into_iter().next() 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,),
(&new_addr,),
)?;
Ok(Some(new_addr.clone()))
Ok(Some(new_addr))
}
/// Adds transport entry to the `transports` table with empty configuration.

210
src/transport/transport_tests.rs
View File

@@ -4,6 +4,7 @@ use std::time::Duration;
use crate::tools::SystemTime;
use super::*;
use crate::imap::ServerMetadata;
use crate::test_utils::TestContext;
use crate::test_utils::TestContextManager;
use crate::tools::time;
@@ -115,7 +116,7 @@ fn dummy_configured_login_param(addr: &str) -> ConfiguredLoginParam {
}
}
fn dummy_transport_data(addr: &str, is_published: bool) -> TransportData {
fn dummy_transport_data(addr: &str) -> TransportData {
TransportData {
configured: dummy_configured_login_param(addr).into(),
entered: EnteredLoginParam {
@@ -123,7 +124,7 @@ fn dummy_transport_data(addr: &str, is_published: bool) -> TransportData {
..Default::default()
},
timestamp: time(),
is_published,
is_published: true,
}
}
@@ -141,7 +142,7 @@ async fn add_dummy_transport(t: &TestContext, addr: &str) -> Result<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_is_published_flag() -> Result<()> {
async fn test_delete_transport() -> Result<()> {
let mut tcm = TestContextManager::new();
let alice = &tcm.alice().await;
let alice2 = &tcm.alice().await;
@@ -157,8 +158,7 @@ async fn test_is_published_flag() -> Result<()> {
bob,
Addresses {
primary: "alice@example.org",
secondary_published: &[],
secondary_unpublished: &[],
secondary: &[],
},
)
.await;
@@ -173,49 +173,35 @@ async fn test_is_published_flag() -> Result<()> {
bob,
Addresses {
primary: "alice@example.org",
secondary_published: &["alice@otherprovider.com"],
secondary_unpublished: &[],
secondary: &["alice@otherprovider.com"],
},
)
.await;
assert_eq!(
alice
.set_transport_unpublished("alice@example.org", true)
.delete_transport("unknown@example.org")
.await
.unwrap_err()
.to_string(),
"Can't set primary relay as unpublished"
"Transport does not exist"
);
// Make sure that the newly generated key has a newer timestamp,
// so that it is recognized by Bob:
SystemTime::shift(Duration::from_secs(2));
alice.evtracker.clear_events();
alice.delete_transport("alice@example.org").await?;
alice
.set_transport_unpublished("alice@otherprovider.com", true)
.await?;
sync_and_check_recipients(alice, alice2, "alice@example.org").await;
check_addrs(
alice,
alice2,
bob,
Addresses {
primary: "alice@example.org",
secondary_published: &[],
secondary_unpublished: &["alice@otherprovider.com"],
},
)
.await;
SystemTime::shift(Duration::from_secs(2));
promote_transport_and_sync(alice, alice2, "alice@otherprovider.com").await?;
alice2
.set_config(Config::ConfiguredAddr, Some("alice@otherprovider.com"))
.await?;
.evtracker
.get_matching(|e| matches!(e, EventType::TransportsModified))
.await;
assert_eq!(
alice.get_config(Config::ConfiguredAddr).await?.as_deref(),
Some("alice@otherprovider.com")
);
sync_and_check_recipients(alice, alice2, "alice@otherprovider.com").await;
check_addrs(
alice,
@@ -223,12 +209,20 @@ async fn test_is_published_flag() -> Result<()> {
bob,
Addresses {
primary: "alice@otherprovider.com",
secondary_published: &["alice@example.org"],
secondary_unpublished: &[],
secondary: &[],
},
)
.await;
assert_eq!(
alice
.delete_transport("alice@otherprovider.com")
.await
.unwrap_err()
.to_string(),
"Cannot remove the last transport"
);
Ok(())
}
@@ -257,7 +251,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 = dummy_transport_data("alice@otherprovider.com", true);
let data = dummy_transport_data("alice@otherprovider.com");
let data = std::slice::from_ref(&data);
sync_transports(alice, data, &[]).await?;
assert!(alice.restart_io_after_fetch.swap(false, Ordering::Relaxed));
@@ -307,75 +301,127 @@ 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.
/// Tests that removing the last transport keeps it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_reelect_local_primary() -> Result<()> {
async fn test_removing_last_transport() -> 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?;
let removed = RemovedTransportData {
addr: "alice@example.org".to_string(),
timestamp: time(),
};
sync_transports(alice, &[], std::slice::from_ref(&removed)).await?;
assert_eq!(alice.count_transports().await?, 1);
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.count_transports().await?, 1);
assert_eq!(
alice.get_config(Config::ConfiguredAddr).await?.as_deref(),
Some("alice@example.org")
Some("alice@otherprovider.com")
);
Ok(())
}
/// Tests which transport is elected as the local primary one.
/// Tests which transport is elected for sending.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_maybe_reelect_local_primary() -> Result<()> {
async fn test_maybe_update_sending_transport() -> 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);
assert_eq!(
t.sql.transaction(maybe_update_sending_transport).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=?",
"DELETE FROM transports WHERE addr=?",
("alice@example.org",),
)
.await?;
assert_eq!(
t.sql
.transaction(maybe_reelect_local_primary)
.transaction(maybe_update_sending_transport)
.await?
.as_deref(),
Some("alice@two.com")
Some("alice@one.com")
);
Ok(())
}
/// Tests that a transport an older core unpublished is removed.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_sync_unpublished_transport_removes_it() -> Result<()> {
let mut tcm = TestContextManager::new();
let alice = &tcm.alice().await;
add_dummy_transport(alice, "alice@otherprovider.com").await?;
let mut data = dummy_transport_data("alice@otherprovider.com");
data.is_published = false;
let transport_id: u32 = alice
.sql
.query_get_value(
"SELECT id FROM transports WHERE addr=?",
("alice@otherprovider.com",),
)
.await?
.unwrap();
alice
.metadata
.write()
.await
.insert(transport_id, ServerMetadata::default());
sync_transports(alice, std::slice::from_ref(&data), &[]).await?;
assert_eq!(alice.count_transports().await?, 1);
let tombstone: i64 = alice
.sql
.query_get_value(
"SELECT remove_timestamp FROM removed_transports WHERE addr=?",
("alice@otherprovider.com",),
)
.await?
.unwrap();
assert_eq!(tombstone, data.timestamp);
assert!(!alice.metadata.read().await.contains_key(&transport_id));
Ok(())
}
/// Tests that a stale full-list sync does not resurrect a removed transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_sync_does_not_resurrect_removed_transport() -> Result<()> {
let mut tcm = TestContextManager::new();
let alice = &tcm.alice().await;
add_dummy_transport(alice, "alice@otherprovider.com").await?;
let stale = dummy_transport_data("alice@otherprovider.com");
SystemTime::shift(Duration::from_secs(2));
alice.delete_transport("alice@otherprovider.com").await?;
assert_eq!(alice.count_transports().await?, 1);
sync_transports(alice, std::slice::from_ref(&stale), &[]).await?;
assert_eq!(alice.count_transports().await?, 1);
SystemTime::shift(Duration::from_secs(2));
let readded = dummy_transport_data("alice@otherprovider.com");
sync_transports(alice, std::slice::from_ref(&readded), &[]).await?;
assert_eq!(alice.count_transports().await?, 2);
Ok(())
}
struct Addresses {
primary: &'static str,
secondary_published: &'static [&'static str],
secondary_unpublished: &'static [&'static str],
secondary: &'static [&'static str],
}
async fn check_addrs(
@@ -391,37 +437,11 @@ async fn check_addrs(
)
}
let published_self_addrs = concat(&[addresses.secondary_published, &[addresses.primary]]);
let self_addrs = concat(&[addresses.secondary, &[addresses.primary]]);
for a in [alice2, alice] {
assert_eq(
a.get_all_self_addrs().await.unwrap(),
concat(&[
addresses.secondary_published,
addresses.secondary_unpublished,
&[addresses.primary],
]),
);
assert_eq(
a.get_published_self_addrs().await.unwrap(),
published_self_addrs.clone(),
);
assert_eq(
a.get_published_secondary_self_addrs().await.unwrap(),
concat(&[addresses.secondary_published]),
);
assert_eq(a.get_self_addrs().await.unwrap(), self_addrs.clone());
for transport in a.list_transports().await.unwrap() {
if addresses.primary == transport.param.addr
|| addresses
.secondary_published
.contains(&transport.param.addr.as_str())
{
assert_eq!(transport.is_unpublished, false);
} else if addresses
.secondary_unpublished
.contains(&transport.param.addr.as_str())
{
assert_eq!(transport.is_unpublished, true);
} else {
if !self_addrs.contains(&transport.addr.as_str()) {
panic!("Unexpected transport {transport:?}");
}
}
@@ -430,7 +450,7 @@ async fn check_addrs(
let sent = a.send_text(alice_bob_chat_id, "hi").await;
assert_eq!(
sent.recipients,
format!("bob@example.net {}", published_self_addrs.join(" ")),
format!("bob@example.net {}", self_addrs.join(" ")),
"{} is sending to the wrong set of recipients",
a.name()
);
@@ -439,7 +459,7 @@ async fn check_addrs(
let answer = bob.send_text(bob_alice_chat_id, "hi back").await;
assert_eq(
answer.recipients.split(' ').map(Into::into).collect(),
concat(&[&published_self_addrs, &["bob@example.net"]]),
concat(&[&self_addrs, &["bob@example.net"]]),
);
}
}