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

refactor: Don't write relay candidates into the database, instead use two databases

Browse Source
This commit is contained in:
Hocuri
2026-09-18 20:00:17 +02:00
parent 37a04a8bba
commit db6b7d96dd
5 changed files with 105 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/schema.sql
View File

@@ -631,6 +631,15 @@ CREATE TABLE broadcast_secrets(
-- Candidate chatmail relays for automatic relay management.
CREATE TABLE relay_candidates(
host TEXT PRIMARY KEY NOT NULL,
last_tried INTEGER NOT NULL DEFAULT 0 -- Deprecated 2026-09, replaced with separate relay_candidates_last_tried table.
) STRICT;
-- This table is used for storing the timestamp of the last
-- connection attempt per chatmail relay candidate.
-- This table can contain relays that were removed from the list of candidates,
-- and it does not contain the default relays.
CREATE TABLE relay_candidates_last_tried(
host TEXT PRIMARY KEY NOT NULL,
last_tried INTEGER NOT NULL DEFAULT 0 -- Timestamp of the last connection attempt.
) STRICT;

76
src/autorelay.rs
View File

@@ -2,8 +2,8 @@
//!
//! Chatmail relays create an account on first login,
//! so a profile can add further transports on its own without user interaction.
//! Candidate hosts come from the `relay_candidates` table,
//! which migrations seed with a list of known chatmail relays.
//! Candidate hosts come from the `relay_candidates` table
//! as well as the [`DEFAULT_RELAY_CANDIDATES`] list.
//!
//! Status of implementation:
//! Additions are attempted right before going into IMAP IDLE,
@@ -13,6 +13,7 @@
//! [`Config::AutorelayFinished`] is set and nothing is ever added again,
//! so deleting a transport later does not pull in a replacement.
use std::collections::BTreeSet;
use std::pin::Pin;
use anyhow::Result;
@@ -165,13 +166,7 @@ async fn maybe_add_additional_relays_inner(context: &Context, skip_network: bool
candidates.len(),
);
context
.sql
.execute(
"UPDATE relay_candidates SET last_tried=? WHERE host=?",
(now, host),
)
.await?;
set_relay_candidate_last_tried(context, host, now).await?;
let param = login_param_from_host(host);
let res = crate::configure::configure(context, &param, skip_network).await;
if let Err(e) = res {
@@ -188,37 +183,54 @@ async fn maybe_add_additional_relays_inner(context: &Context, skip_network: bool
Ok(relay_added)
}
async fn set_relay_candidate_last_tried(
context: &Context,
host: &str,
now: i64,
) -> Result<(), anyhow::Error> {
context
.sql
.execute(
"INSERT OR REPLACE INTO relay_candidates_last_tried(host, last_tried) VALUES(?, ?)",
(host, now),
)
.await?;
Ok(())
}
async fn load_relay_candidates(context: &Context, now: i64) -> Result<Vec<String>> {
let cutoff_timestamp = now.saturating_sub(BACKOFF_PERIOD_FOR_NOT_WORKING_RELAY);
let candidates: Vec<String> = context
let res = context
.sql
.transaction(|transaction| {
// Add the default relays if they are not in the database yet
let mut statement =
transaction.prepare("INSERT OR IGNORE INTO relay_candidates(host) VALUES (?)")?;
for host in DEFAULT_RELAY_CANDIDATES {
statement.execute((host,))?;
}
let mut candidates: BTreeSet<String> =
transaction.query_map_collect("SELECT host FROM relay_candidates", (), |row| {
Ok(row.get(0)?)
})?;
transaction.query_map_vec(
// This also selects candidates which have last_tried in the future,
// essentially treating them as never tried,
// so if some timestamp far in the future is accidentally stored,
// we are not stuck never trying the candidate.
// After trying the candidate, last_tried will be corrected to the current time.
"SELECT host FROM relay_candidates WHERE (last_tried<? OR last_tried>?)
AND NOT EXISTS (
SELECT 1
FROM transports
WHERE substr(addr, instr(addr, '@') + 1) = host
)",
candidates.extend(DEFAULT_RELAY_CANDIDATES.iter().map(|s| s.to_string()));
let cutoff_timestamp = now.saturating_sub(BACKOFF_PERIOD_FOR_NOT_WORKING_RELAY);
// This does not select candidates which have last_tried in the future,
// essentially treating them as never tried,
// so if some timestamp far in the future is accidentally stored,
// we are not stuck never trying the candidate.
// After trying the candidate, last_tried will be corrected to the current time.
let exclude: BTreeSet<String> = transaction.query_map_collect(
"SELECT host FROM relay_candidates_last_tried WHERE (last_tried>=? AND last_tried<=?)
UNION
SELECT substr(addr, instr(addr, '@') + 1) FROM transports",
(cutoff_timestamp, now),
|row| Ok(row.get::<_, String>(0)?),
)
|row| Ok(row.get(0)?),
)?;
Ok(candidates
.difference(&exclude)
.map(|s| s.to_string())
.collect::<Vec<String>>())
})
.await?;
Ok(candidates)
Ok(res)
}
pub(crate) fn login_param_from_host(host: &str) -> EnteredLoginParam {

79
src/autorelay/autorelay_tests.rs
View File

@@ -44,40 +44,38 @@ async fn test_load_relay_candidates_single() -> Result<()> {
enable_config(t).await;
let now = time();
// Fill the default candidates, and make sure that they
// are not used by setting last_used to now:
load_relay_candidates(t, now).await?;
t.sql
.execute("UPDATE relay_candidates SET last_tried=?", (now,))
.await?;
// This host should be returned by load_relay_candidates():
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
("never_tried.example", 0),
"INSERT INTO relay_candidates (host) VALUES (?)",
("never_tried.example",),
)
.await?;
// This host was recently tried and should not be returned:
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
("recent.example", now),
"INSERT INTO relay_candidates (host) VALUES (?)",
("recent.example",),
)
.await?;
set_relay_candidate_last_tried(t, "recent.example", now).await?;
// This host is already in use (alice@example.org) and should not be returned:
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
("example.org", 0),
"INSERT INTO relay_candidates (host) VALUES (?)",
("example.org",),
)
.await?;
let candidates = load_relay_candidates(t, now).await?;
assert_eq!(candidates, vec!["never_tried.example".to_string()]);
assert!(candidates.contains(&"never_tried.example".to_string()));
assert_eq!(candidates.contains(&"recent.example".to_string()), false);
assert_eq!(candidates.contains(&"example.org".to_string()), false);
assert_eq!(candidates.len(), DEFAULT_RELAY_CANDIDATES.len() + 1);
Ok(())
}
@@ -92,10 +90,7 @@ async fn test_load_relay_candidates_multiple() -> Result<()> {
for host in EXAMPLE_CANDIDATES {
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
(host, 0),
)
.execute("INSERT INTO relay_candidates (host) VALUES (?)", (host,))
.await?;
}
@@ -197,17 +192,21 @@ async fn test_maybe_add_additional_relays_add_one() -> Result<()> {
enable_config(t).await;
let now = time();
// Fill the default candidates, and make sure that they
// Make sure that default relay candidates
// are not used by setting last_used to now:
load_relay_candidates(t, now).await?;
t.sql
.execute("UPDATE relay_candidates SET last_tried=?", (now,))
.await?;
for candidate in DEFAULT_RELAY_CANDIDATES {
t.sql
.execute(
"INSERT INTO relay_candidates_last_tried(host, last_tried) VALUES(?,?)",
(candidate, now),
)
.await?;
}
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
("relay.example", 0),
"INSERT INTO relay_candidates (host) VALUES (?)",
("relay.example",),
)
.await?;
@@ -232,16 +231,6 @@ async fn test_maybe_add_additional_relays_add_multiple() -> Result<()> {
enable_config(t).await;
let now = time();
t.sql.execute("DELETE FROM relay_candidates", ()).await?;
for host in ["a.example", "b.example", "c.example", "d.example"] {
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
(host, 0),
)
.await?;
}
let skip_network = true;
let relay_added = maybe_add_additional_relays_inner(t, skip_network).await?;
assert!(relay_added);
@@ -261,18 +250,22 @@ async fn test_maybe_add_additional_relays_failure() -> Result<()> {
enable_config(t).await;
let now = time();
// Fill the default candidates, and make sure that they
// Make sure that default relay candidates
// are not used by setting last_used to now:
load_relay_candidates(t, now).await?;
t.sql
.execute("UPDATE relay_candidates SET last_tried=?", (now,))
.await?;
for candidate in DEFAULT_RELAY_CANDIDATES {
t.sql
.execute(
"INSERT INTO relay_candidates_last_tried(host, last_tried) VALUES(?,?)",
(candidate, now - 2),
)
.await?;
}
for i in 1..10 {
t.sql
.execute(
"INSERT INTO relay_candidates (host, last_tried) VALUES (?, ?)",
(format!("{i}.invalid.example"), 0),
"INSERT INTO relay_candidates (host) VALUES (?)",
(format!("{i}.invalid.example"),),
)
.await?;
}
@@ -295,7 +288,7 @@ async fn test_maybe_add_additional_relays_failure() -> Result<()> {
assert!(
t.sql
.exists(
"SELECT COUNT(*) FROM relay_candidates WHERE last_tried>=?",
"SELECT COUNT(*) FROM relay_candidates_last_tried WHERE last_tried>=?",
(now,)
)
.await?

12
src/sql.rs
View File

@@ -687,27 +687,29 @@ impl Sql {
pub(crate) trait TransactionExt {
/// Prepares and executes the statement and maps a function over the resulting rows.
///
/// Collects the resulting rows into a `Vec`.
fn query_map_vec<T, F>(
/// Collects the resulting rows into a collection.
fn query_map_collect<T, C, F>(
&self,
sql: &str,
params: impl rusqlite::Params + Send,
f: F,
) -> Result<Vec<T>>
) -> Result<C>
where
T: Send + 'static,
C: Send + 'static + std::iter::FromIterator<T>,
F: Send + FnMut(&rusqlite::Row) -> Result<T>;
}
impl TransactionExt for rusqlite::Transaction<'_> {
fn query_map_vec<T, F>(
fn query_map_collect<T, C, F>(
&self,
sql: &str,
params: impl rusqlite::Params + Send,
f: F,
) -> Result<Vec<T>>
) -> Result<C>
where
T: Send + 'static,
C: Send + 'static + std::iter::FromIterator<T>,
F: Send + FnMut(&rusqlite::Row) -> Result<T>,
{
let mut stmt = self.prepare(sql)?;

11
src/sql/migrations.rs
View File

@@ -2675,8 +2675,15 @@ CREATE TABLE smtp2 (
inc_and_check(&mut migration_version, 167)?;
if dbversion < migration_version {
sql.execute_migration("DELETE FROM relay_candidates;", migration_version)
.await?;
sql.execute_migration(
"DELETE FROM relay_candidates;
CREATE TABLE relay_candidates_last_tried(
host TEXT PRIMARY KEY NOT NULL,
last_tried INTEGER NOT NULL DEFAULT 0
) STRICT",
migration_version,
)
.await?;
}
let new_version = sql