feat: Remove hidden relays automatically

deltachat-jsonrpc/src/api.rs

@@ -590,7 +590,7 @@ impl CommandApi {
 
     /// Immediately deletes a transport, potentially causing messages not to arrive.
     /// This must ONLY be used by the automated tests.
-    /// UI implementations must use `set_transport_unpublished()` instead.
+    /// UI implementations must use [`Self::set_transport_unpublished`] instead.
     async fn delete_transport(&self, account_id: u32, addr: String) -> Result<()> {
         let ctx = self.get_context(account_id).await?;
         ctx.delete_transport(&addr).await
@@ -605,6 +605,11 @@ impl CommandApi {
     /// 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.
+    ///
+    /// Unpublished transports that are not used to receive any new messages for a time defined by
+    /// [`HIDDEN_TRANSPORT_CLEANUP_TIME`] are automatically removed.
+    ///
+    /// [`HIDDEN_TRANSPORT_CLEANUP_TIME`]: deltachat::sql::HIDDEN_TRANSPORT_CLEANUP_TIME
     async fn set_transport_unpublished(
         &self,
         account_id: u32,

src/configure.rs

@@ -232,7 +232,7 @@ impl Context {
 
     /// Immediately deletes a transport, potentially causing messages not to arrive.
     /// This must ONLY be used by the automated tests.
-    /// UI implementations must use `set_transport_unpublished()` instead.
+    /// UI implementations must use [`Self::set_transport_unpublished`] instead.
     pub async fn delete_transport(&self, addr: &str) -> Result<()> {
         let now = time();
         let removed_transport_id = self
@@ -291,6 +291,11 @@ impl Context {
     /// 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.
+    ///
+    /// Unpublished transports that are not used to receive any new messages for a time defined by
+    /// [`HIDDEN_TRANSPORT_CLEANUP_TIME`] are automatically removed.
+    ///
+    /// [`HIDDEN_TRANSPORT_CLEANUP_TIME`]: crate::sql::HIDDEN_TRANSPORT_CLEANUP_TIME
     pub async fn set_transport_unpublished(&self, addr: &str, unpublished: bool) -> Result<()> {
         self.sql
             .transaction(|trans| {

src/receive_imf.rs

@@ -48,6 +48,7 @@ use crate::securejoin::{
 };
 use crate::simplify;
 use crate::smtp::msg_has_pending_smtp_job;
+use crate::sql;
 use crate::stats::STATISTICS_BOT_EMAIL;
 use crate::stock_str;
 use crate::sync::Sync::*;
@@ -743,6 +744,10 @@ pub(crate) async fn receive_imf_inner(
         .context("add_parts error")?
     };
 
+    sql::update_transport_last_rcvd_timestamp(context, rfc724_mid)
+        .await
+        .context("Failed to update transport's last_rcvd_timestamp value.")?;
+
     if !from_id.is_special() {
         contact::update_last_seen(context, from_id, mime_parser.timestamp_sent).await?;
     }

src/sql.rs

@@ -47,6 +47,9 @@ mod pool;
 
 use pool::{Pool, WalCheckpointStats};
 
+/// How long a hidden and unused transport should be kept in the database before being deleted.
+pub const HIDDEN_TRANSPORT_CLEANUP_TIME: i64 = 90 * 24 * 60 * 60;
+
 /// A wrapper around the underlying Sqlite3 object.
 #[derive(Debug)]
 pub struct Sql {
@@ -908,10 +911,54 @@ 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();
+
     info!(context, "Housekeeping done.");
     Ok(())
 }
 
+/// Removes transports that are hidden (`is_published=0`),
+/// and haven't been used to receive new messages for [`HIDDEN_TRANSPORT_CLEANUP_TIME`] seconds.
+pub async fn remove_unused_hidden_transports(context: &Context) -> Result<usize> {
+    let now = time();
+    let cutoff = now.saturating_sub(HIDDEN_TRANSPORT_CLEANUP_TIME);
+    context
+        .sql
+        .execute(
+            "DELETE FROM transports
+            WHERE is_published=0
+            AND last_rcvd_timestamp<?1
+            AND add_timestamp<?1",
+            (cutoff,),
+        )
+        .await
+}
+
+/// Updates the corresponding transport's `last_rcvd_timestamp`
+/// with value from message's `timestamp_rcvd`,
+/// if it is larger than the current value.
+pub async fn update_transport_last_rcvd_timestamp(
+    context: &Context,
+    rfc724_mid: &str,
+) -> Result<usize> {
+    context
+        .sql
+        .execute(
+            "UPDATE transports
+            SET last_rcvd_timestamp=MAX(msgs.timestamp_rcvd, transports.last_rcvd_timestamp)
+            FROM transports t1
+            INNER JOIN imap ON t1.id=imap.transport_id
+            INNER JOIN msgs ON msgs.rfc724_mid=imap.rfc724_mid
+            WHERE msgs.rfc724_mid=?",
+            (rfc724_mid,),
+        )
+        .await
+}
+
 /// Get the value of a column `idx` of the `row` as `Vec<u8>`.
 pub fn row_get_vec(row: &Row, idx: usize) -> rusqlite::Result<Vec<u8>> {
     row.get(idx).or_else(|err| match row.get_ref(idx)? {

src/sql/migrations.rs

@@ -2456,6 +2456,27 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed.
         .await?;
     }
 
+    inc_and_check(&mut migration_version, 155)?;
+    if dbversion < migration_version {
+        // last_rcvd_timestamp tracks timestamp of the last received message
+        // on the transport. This is used to remove hidden transports that
+        // were not used to receive messages for some predefined time.
+        //
+        // NOTE: ideally we would use `DEFAULT (unixepoch())`,
+        // but sqlite forbids non-constant default in ALTER TABLE.
+        // Instead, we need to remember to also check `add_timestamp`
+        // before removing transport
+        // (so it won't be immediately deleted if last_rcvd_timestamp=0).
+        sql.execute_migration(
+            "
+            ALTER TABLE transports
+            ADD last_rcvd_timestamp INTEGER NOT NULL DEFAULT 0
+            ",
+            migration_version,
+        )
+        .await?;
+    }
+
     let new_version = sql
         .get_raw_config_int(VERSION_CFG)
         .await?

src/sql/sql_tests.rs

@@ -1,6 +1,7 @@
 use super::*;
 use crate::message::Message;
 use crate::{EventType, test_utils::TestContext};
+use deltachat_time::SystemTimeTools;
 
 #[test]
 fn test_maybe_add_file() {
@@ -367,3 +368,138 @@ 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::<u64, _, _>(
+                "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),
+            )?;
+
+            // msgs
+            t.execute(
+                "INSERT INTO msgs
+                (rfc724_mid,
+                timestamp_rcvd,
+                mime_headers,
+                mime_in_reply_to,
+                mime_references,
+                txt_normalized)
+                VALUES (?, ?, ?, ?, ?, ?)",
+                ("smth", now, "", "", "", ""),
+            )?;
+
+            // imap
+            t.execute(
+                "INSERT INTO imap
+                (transport_id,
+                rfc724_mid,
+                folder,
+                target,
+                uid,
+                uidvalidity)
+                VALUES (?, ?, ?, ?, ?, ?)",
+                (transport_id, "smth", "", "", 0, 0),
+            )?;
+
+            Ok(transport_id)
+        })
+        .await?;
+
+    macro_rules! assert_transport_removed {
+        ($removed:expr) => {
+            let transports_count: u32 = t
+                .sql
+                .query_get_value("SELECT count(*) FROM transports", ())
+                .await?
+                .unwrap();
+            assert_eq!(transports_count, if $removed { 2 } else { 3 });
+            let hidden_transport_present: bool = t
+                .sql
+                .query_get_value(
+                    "SELECT count(*) FROM transports WHERE id=?",
+                    (transport_id,),
+                )
+                .await?
+                .unwrap();
+            assert_eq!(hidden_transport_present, !$removed);
+        };
+    }
+
+    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, "smth").await?;
+    // last_rcvd_timestamp should update
+    assert_eq!(
+        t.sql
+            .query_get_value::<i64>(
+                "SELECT last_rcvd_timestamp FROM transports WHERE id=?",
+                (transport_id,)
+            )
+            .await?
+            .unwrap(),
+        now
+    );
+
+    assert_transport_removed!(false);
+    remove_unused_hidden_transports(&t).await?;
+    // it was recently used, so nothing is deleted
+    assert_transport_removed!(false);
+
+    SystemTimeTools::shift(Duration::from_secs(
+        (HIDDEN_TRANSPORT_CLEANUP_TIME + 1).try_into()?,
+    ));
+
+    remove_unused_hidden_transports(&t).await?;
+    assert_transport_removed!(true);
+
+    Ok(())
+}