Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-10-25 22:37:57 +03:00
Code Issues Packages Projects Releases Wiki Activity

smtp_failure table

Browse Source
This commit is contained in:
link2xt
2026-10-09 01:25:30 +00:00
parent c9df97b908
commit 5cd7970ba5
4 changed files with 102 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/schema.sql
+13
View File
@@ -471,6 +471,19 @@ CREATE TABLE smtp_success (
FOREIGN KEY(transport_id) REFERENCES transports(id) ON DELETE CASCADE
) STRICT;
-- Similar to smtp_success, but records failures.
CREATE TABLE smtp_failure (
-- Sequentially increasing ID of the failure
-- Transport with the highest ID is to be used last.
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
-- ID of the transport that was used to attempt sending a message.
transport_id INTEGER UNIQUE NOT NULL,
-- Delete `smtp_failure` rows when the transport is deleted.
FOREIGN KEY(transport_id) REFERENCES transports(id) ON DELETE CASCADE
) STRICT;
-- Table of "sync items" to be grouped into sync messages
-- and sent to own devices.
CREATE TABLE multi_device_sync (
src/smtp.rs
+28 -11
View File
@@ -66,7 +66,10 @@ async fn sorted_transports(context: &Context) -> Result<Vec<(u32, ConfiguredLogi
.query_map_vec(
"SELECT transports.id, configured_param FROM transports
LEFT JOIN smtp_success ON smtp_success.transport_id=transports.id
ORDER BY IFNULL(smtp_success.id, 0) DESC, transports.id ASC",
LEFT JOIN smtp_failure ON smtp_failure.transport_id=transports.id
ORDER BY IFNULL(smtp_success.id, 0) DESC,
IFNULL(smtp_failure.id, 0) ASC,
transports.id ASC",
(),
|row| {
let id: u32 = row.get(0)?;
@@ -84,22 +87,36 @@ async fn record_success(context: &Context, transport_id: u32) -> Result<()> {
// if the row exists already, so it becomes the highest rowid in the table.
context
.sql
.execute(
"INSERT OR REPLACE INTO smtp_success (transport_id) VALUES (?)",
(transport_id,),
)
.transaction(|t| {
t.execute(
"INSERT OR REPLACE INTO smtp_success (transport_id) VALUES (?)",
(transport_id,),
)?;
t.execute(
"DELETE FROM smtp_failure WHERE transport_id = ?",
(transport_id,),
)?;
Ok(())
})
.await?;
Ok(())
}
/// Removes last remembered success for SMTP transport so another transport is tried next time.
async fn remove_success(context: &Context, transport_id: u32) -> Result<()> {
async fn record_failure(context: &Context, transport_id: u32) -> Result<()> {
context
.sql
.execute(
"DELETE FROM smtp_success WHERE transport_id = ?",
(transport_id,),
)
.transaction(|t| {
t.execute(
"INSERT OR REPLACE INTO smtp_failure (transport_id) VALUES (?)",
(transport_id,),
)?;
t.execute(
"DELETE FROM smtp_success WHERE transport_id = ?",
(transport_id,),
)?;
Ok(())
})
.await?;
Ok(())
}
@@ -284,7 +301,7 @@ async fn smtp_send(
if matches!(
err,
async_smtp::error::Error::Permanent(_) | async_smtp::error::Error::Transient(_)
) && let Err(err) = remove_success(context, transport_id).await
) && let Err(err) = record_failure(context, transport_id).await
{
warn!(
context,
src/smtp/smtp_tests.rs
+46 -47
View File
@@ -1,8 +1,25 @@
use anyhow::Result;
use crate::test_utils::TestContextManager;
use crate::test_utils::{TestContext, TestContextManager};
use crate::transport;
async fn sorted_transport_addrs(t: &TestContext) -> Result<String> {
let transports = super::sorted_transports(t).await?;
let addrs: Vec<String> = transports
.into_iter()
.map(|(_transport_id, transport)| transport.addr)
.collect();
Ok(addrs.join(" "))
}
async fn sorted_transport_ids(t: &TestContext) -> Result<Vec<u32>> {
let transports = super::sorted_transports(t).await?;
Ok(transports
.into_iter()
.map(|(transport_id, _transport)| transport_id)
.collect())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_smtp_candidates() -> Result<()> {
let mut tcm = TestContextManager::new();
@@ -12,60 +29,42 @@ async fn test_smtp_candidates() -> Result<()> {
transport::add_pseudo_transport(t, "bar@example.net").await?;
transport::add_pseudo_transport(t, "baz@example.net").await?;
let transports = super::sorted_transports(t).await?;
let [
(transport_id1, ref transport1),
(transport_id2, ref transport2),
(transport_id3, ref transport3),
] = transports[..]
else {
panic!("Unexpected number of transports");
};
// By default first added transport is used first.
assert_eq!(transport1.addr, "foo@example.net");
assert_eq!(transport2.addr, "bar@example.net");
assert_eq!(transport3.addr, "baz@example.net");
assert_eq!(
sorted_transport_addrs(t).await?,
"foo@example.net bar@example.net baz@example.net"
);
assert_eq!(sorted_transport_ids(t).await?, [1, 2, 3]);
super::record_success(t, transport_id3).await?;
let transports2 = super::sorted_transports(t).await?;
assert_eq!(transports2[0].0, transport_id3);
assert_eq!(transports2[1].0, transport_id1);
assert_eq!(transports2[2].0, transport_id2);
super::record_success(t, 3).await?;
assert_eq!(sorted_transport_ids(t).await?, [3, 1, 2]);
super::record_success(t, transport_id2).await?;
let transports3 = super::sorted_transports(t).await?;
assert_eq!(transports3[0].0, transport_id2);
assert_eq!(transports3[1].0, transport_id3);
assert_eq!(transports3[2].0, transport_id1);
super::record_success(t, 2).await?;
assert_eq!(sorted_transport_ids(t).await?, [2, 3, 1]);
super::record_success(t, transport_id3).await?;
let transports4 = super::sorted_transports(t).await?;
assert_eq!(transports4[0].0, transport_id3);
assert_eq!(transports4[1].0, transport_id2);
assert_eq!(transports4[2].0, transport_id1);
super::record_success(t, 3).await?;
assert_eq!(sorted_transport_ids(t).await?, [3, 2, 1]);
super::remove_success(t, transport_id3).await?;
let transports5 = super::sorted_transports(t).await?;
assert_eq!(transports5[0].0, transport_id2);
assert_eq!(transports5[1].0, transport_id1);
assert_eq!(transports5[2].0, transport_id3);
super::record_failure(t, 3).await?;
assert_eq!(sorted_transport_ids(t).await?, [2, 1, 3]);
// New transport is added to the end.
// New transport is added before all failing transports.
transport::add_pseudo_transport(t, "qux@example.net").await?;
let transports6 = super::sorted_transports(t).await?;
assert_eq!(transports6[0].0, transport_id2);
assert_eq!(transports6[1].0, transport_id1);
assert_eq!(transports6[2].0, transport_id3);
assert_eq!(transports6.len(), 4);
let transport_id4 = transports6[3].0;
super::record_success(t, transport_id4).await?;
assert_eq!(
sorted_transport_addrs(t).await?,
"bar@example.net foo@example.net qux@example.net baz@example.net"
);
assert_eq!(sorted_transport_ids(t).await?, [2, 1, 4, 3]);
let transports7 = super::sorted_transports(t).await?;
assert_eq!(transports7[0].0, transport_id4);
assert_eq!(transports7[1].0, transport_id2);
assert_eq!(transports7[2].0, transport_id1);
assert_eq!(transports7[3].0, transport_id3);
super::record_success(t, 4).await?;
assert_eq!(sorted_transport_ids(t).await?, [4, 2, 1, 3]);
super::record_failure(t, 4).await?;
assert_eq!(sorted_transport_ids(t).await?, [2, 1, 3, 4]);
super::record_failure(t, 3).await?;
assert_eq!(sorted_transport_ids(t).await?, [2, 1, 4, 3]);
super::record_failure(t, 2).await?;
assert_eq!(sorted_transport_ids(t).await?, [1, 4, 3, 2]);
Ok(())
}
src/sql/migrations.rs
+15
View File
@@ -2687,6 +2687,21 @@ CREATE TABLE smtp_success (
.await?;
}
inc_and_check(&mut migration_version, 169)?;
if dbversion < migration_version {
sql.execute_migration(
"
CREATE TABLE smtp_failure (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
transport_id INTEGER UNIQUE NOT NULL,
FOREIGN KEY(transport_id) REFERENCES transports(id) ON DELETE CASCADE
) STRICT;
",
migration_version,
)
.await?;
}
let new_version = sql
.get_raw_config_int(VERSION_CFG)
.await?