mirror of
https://github.com/chatmail/core.git
synced 2026-05-10 18:36:29 +03:00
Compare commits
2 Commits
link2xt/os
...
hpk/per-re
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9f1c02d89f | ||
|
|
5e7fcdac8f |
@@ -275,7 +275,7 @@ pub unsafe extern "C" fn dc_get_config(
|
||||
.strdup()
|
||||
} else {
|
||||
match config::Config::from_str(&key)
|
||||
.with_context(|| format!("Invalid key {key:?}"))
|
||||
.with_context(|| format!("Invalid key {:?}", &key))
|
||||
.log_err(ctx)
|
||||
{
|
||||
Ok(key) => ctx
|
||||
|
||||
@@ -122,7 +122,7 @@ async fn poke_spec(context: &Context, spec: Option<&str>) -> bool {
|
||||
let name_f = entry.file_name();
|
||||
let name = name_f.to_string_lossy();
|
||||
if name.ends_with(".eml") {
|
||||
let path_plus_name = format!("{real_spec}/{name}");
|
||||
let path_plus_name = format!("{}/{}", &real_spec, name);
|
||||
println!("Import: {path_plus_name}");
|
||||
if poke_eml_file(context, Path::new(&path_plus_name))
|
||||
.await
|
||||
@@ -133,11 +133,11 @@ async fn poke_spec(context: &Context, spec: Option<&str>) -> bool {
|
||||
}
|
||||
}
|
||||
} else {
|
||||
eprintln!("Import: Cannot open directory {real_spec:?}.");
|
||||
eprintln!("Import: Cannot open directory \"{}\".", &real_spec);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
println!("Import: {read_cnt} items read from {real_spec:?}.");
|
||||
println!("Import: {} items read from \"{}\".", read_cnt, &real_spec);
|
||||
if read_cnt > 0 {
|
||||
context.emit_msgs_changed_without_ids();
|
||||
}
|
||||
@@ -179,7 +179,7 @@ async fn log_msg(context: &Context, prefix: impl AsRef<str>, msg: &Message) {
|
||||
msg.get_id(),
|
||||
if msg.get_showpadlock() { "🔒" } else { "" },
|
||||
if msg.has_location() { "📍" } else { "" },
|
||||
contact_name,
|
||||
&contact_name,
|
||||
contact_id,
|
||||
msgtext,
|
||||
if msg.has_html() { "[HAS-HTML]️" } else { "" },
|
||||
@@ -221,7 +221,7 @@ async fn log_msg(context: &Context, prefix: impl AsRef<str>, msg: &Message) {
|
||||
},
|
||||
statestr,
|
||||
downloadstate,
|
||||
temp2,
|
||||
&temp2,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -561,7 +561,7 @@ pub async fn cmdline(context: Context, line: &str, chat_id: &mut ChatId) -> Resu
|
||||
.map_or_else(String::new, |prefix| format!("{prefix}: ")),
|
||||
summary.text,
|
||||
statestr,
|
||||
timestr,
|
||||
×tr,
|
||||
if chat.is_sending_locations() {
|
||||
"📍"
|
||||
} else {
|
||||
|
||||
@@ -432,7 +432,7 @@ async fn handle_cmd(
|
||||
{
|
||||
println!("Open the following url, set mail_pw to the generated token and server_flags to 2:\n{oauth2_url}");
|
||||
} else {
|
||||
println!("OAuth2 not available for {addr}.");
|
||||
println!("OAuth2 not available for {}.", &addr);
|
||||
}
|
||||
} else {
|
||||
println!("oauth2: set addr first.");
|
||||
|
||||
@@ -794,7 +794,7 @@ impl Config {
|
||||
.with_push_subscriber(push_subscriber.clone())
|
||||
.build()
|
||||
.await
|
||||
.with_context(|| format!("failed to create context from file {dbfile:?}"))?;
|
||||
.with_context(|| format!("failed to create context from file {:?}", &dbfile))?;
|
||||
// Try to open without a passphrase,
|
||||
// but do not return an error if account is passphare-protected.
|
||||
ctx.open("".to_string()).await?;
|
||||
|
||||
18
src/chat.rs
18
src/chat.rs
@@ -2529,7 +2529,7 @@ async fn prepare_msg_blob(context: &Context, msg: &mut Message) -> Result<()> {
|
||||
// running numbers, etc.
|
||||
let filename: String = match viewtype_orig {
|
||||
Viewtype::Voice => format!(
|
||||
"voice-messsage_{}.{suffix}",
|
||||
"voice-messsage_{}.{}",
|
||||
chrono::Utc
|
||||
.timestamp_opt(msg.timestamp_sort, 0)
|
||||
.single()
|
||||
@@ -2537,9 +2537,10 @@ async fn prepare_msg_blob(context: &Context, msg: &mut Message) -> Result<()> {
|
||||
|| "YY-mm-dd_hh:mm:ss".to_string(),
|
||||
|ts| ts.format("%Y-%m-%d_%H-%M-%S").to_string()
|
||||
),
|
||||
&suffix
|
||||
),
|
||||
Viewtype::Image | Viewtype::Gif => format!(
|
||||
"image_{}.{suffix}",
|
||||
"image_{}.{}",
|
||||
chrono::Utc
|
||||
.timestamp_opt(msg.timestamp_sort, 0)
|
||||
.single()
|
||||
@@ -2547,9 +2548,10 @@ async fn prepare_msg_blob(context: &Context, msg: &mut Message) -> Result<()> {
|
||||
|| "YY-mm-dd_hh:mm:ss".to_string(),
|
||||
|ts| ts.format("%Y-%m-%d_%H-%M-%S").to_string(),
|
||||
),
|
||||
&suffix,
|
||||
),
|
||||
Viewtype::Video => format!(
|
||||
"video_{}.{suffix}",
|
||||
"video_{}.{}",
|
||||
chrono::Utc
|
||||
.timestamp_opt(msg.timestamp_sort, 0)
|
||||
.single()
|
||||
@@ -2557,6 +2559,7 @@ async fn prepare_msg_blob(context: &Context, msg: &mut Message) -> Result<()> {
|
||||
|| "YY-mm-dd_hh:mm:ss".to_string(),
|
||||
|ts| ts.format("%Y-%m-%d_%H-%M-%S").to_string()
|
||||
),
|
||||
&suffix
|
||||
),
|
||||
_ => filename,
|
||||
};
|
||||
@@ -2951,7 +2954,6 @@ WHERE id=?
|
||||
)
|
||||
.await?;
|
||||
|
||||
let chunk_size = context.get_max_smtp_rcpt_to().await?;
|
||||
let trans_fn = |t: &mut rusqlite::Transaction| {
|
||||
let mut row_ids = Vec::<i64>::new();
|
||||
|
||||
@@ -2965,12 +2967,12 @@ WHERE id=?
|
||||
"INSERT INTO smtp (rfc724_mid, recipients, mime, msg_id)
|
||||
VALUES (?1, ?2, ?3, ?4)",
|
||||
)?;
|
||||
for recipients_chunk in recipients.chunks(chunk_size) {
|
||||
let recipients_chunk = recipients_chunk.join(" ");
|
||||
if !recipients.is_empty() {
|
||||
let all_recipients = recipients.join(" ");
|
||||
if let Some(pre_msg) = &rendered_pre_msg {
|
||||
let row_id = stmt.execute((
|
||||
&pre_msg.rfc724_mid,
|
||||
&recipients_chunk,
|
||||
&all_recipients,
|
||||
&pre_msg.message,
|
||||
msg.id,
|
||||
))?;
|
||||
@@ -2978,7 +2980,7 @@ WHERE id=?
|
||||
}
|
||||
let row_id = stmt.execute((
|
||||
&rendered_msg.rfc724_mid,
|
||||
&recipients_chunk,
|
||||
&all_recipients,
|
||||
&rendered_msg.message,
|
||||
msg.id,
|
||||
))?;
|
||||
|
||||
@@ -707,7 +707,8 @@ async fn get_autoconfig(
|
||||
ctx,
|
||||
// the doc does not mention `emailaddress=`, however, Thunderbird adds it, see <https://releases.mozilla.org/pub/thunderbird/>, which makes some sense
|
||||
&format!(
|
||||
"https://{param_domain}/.well-known/autoconfig/mail/config-v1.1.xml?emailaddress={param_addr_urlencoded}"
|
||||
"https://{}/.well-known/autoconfig/mail/config-v1.1.xml?emailaddress={}",
|
||||
¶m_domain, ¶m_addr_urlencoded
|
||||
),
|
||||
¶m.addr,
|
||||
)
|
||||
@@ -720,7 +721,7 @@ async fn get_autoconfig(
|
||||
// Outlook uses always SSL but different domains (this comment describes the next two steps)
|
||||
if let Ok(res) = outlk_autodiscover(
|
||||
ctx,
|
||||
format!("https://{param_domain}/autodiscover/autodiscover.xml"),
|
||||
format!("https://{}/autodiscover/autodiscover.xml", ¶m_domain),
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -730,7 +731,10 @@ async fn get_autoconfig(
|
||||
|
||||
if let Ok(res) = outlk_autodiscover(
|
||||
ctx,
|
||||
format!("https://autodiscover.{param_domain}/autodiscover/autodiscover.xml",),
|
||||
format!(
|
||||
"https://autodiscover.{}/autodiscover/autodiscover.xml",
|
||||
¶m_domain
|
||||
),
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -741,7 +745,7 @@ async fn get_autoconfig(
|
||||
// always SSL for Thunderbird's database
|
||||
if let Ok(res) = moz_autoconfigure(
|
||||
ctx,
|
||||
&format!("https://autoconfig.thunderbird.net/v1.1/{param_domain}"),
|
||||
&format!("https://autoconfig.thunderbird.net/v1.1/{}", ¶m_domain),
|
||||
¶m.addr,
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -204,9 +204,6 @@ pub const MAX_RCVD_IMAGE_PIXELS: u32 = 50_000_000;
|
||||
// `max_smtp_rcpt_to` in the provider db.
|
||||
pub(crate) const DEFAULT_MAX_SMTP_RCPT_TO: usize = 50;
|
||||
|
||||
/// Same as `DEFAULT_MAX_SMTP_RCPT_TO`, but for chatmail relays.
|
||||
pub(crate) const DEFAULT_CHATMAIL_MAX_SMTP_RCPT_TO: usize = 999;
|
||||
|
||||
/// How far the last quota check needs to be in the past to be checked by the background function (in seconds).
|
||||
pub(crate) const DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT: u64 = 12 * 60 * 60; // 12 hours
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ use tokio::sync::{Mutex, Notify, RwLock};
|
||||
|
||||
use crate::chat::{ChatId, get_chat_cnt};
|
||||
use crate::config::Config;
|
||||
use crate::constants::{self, DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT, DC_VERSION_STR};
|
||||
use crate::constants::{DC_BACKGROUND_FETCH_QUOTA_CHECK_RATELIMIT, DC_VERSION_STR};
|
||||
use crate::contact::{Contact, ContactId};
|
||||
use crate::debug_logging::DebugLogging;
|
||||
use crate::events::{Event, EventEmitter, EventType, Events};
|
||||
@@ -587,23 +587,6 @@ impl Context {
|
||||
self.get_config_bool(Config::IsChatmail).await
|
||||
}
|
||||
|
||||
/// Returns maximum number of recipients the provider allows to send a single email to.
|
||||
pub(crate) async fn get_max_smtp_rcpt_to(&self) -> Result<usize> {
|
||||
let is_chatmail = self.is_chatmail().await?;
|
||||
let val = self
|
||||
.get_configured_provider()
|
||||
.await?
|
||||
.and_then(|provider| provider.opt.max_smtp_rcpt_to)
|
||||
.map_or_else(
|
||||
|| match is_chatmail {
|
||||
true => constants::DEFAULT_CHATMAIL_MAX_SMTP_RCPT_TO,
|
||||
false => constants::DEFAULT_MAX_SMTP_RCPT_TO,
|
||||
},
|
||||
usize::from,
|
||||
);
|
||||
Ok(val)
|
||||
}
|
||||
|
||||
/// Does a single round of fetching from IMAP and returns.
|
||||
///
|
||||
/// Can be used even if I/O is currently stopped.
|
||||
|
||||
31
src/imap.rs
31
src/imap.rs
@@ -1045,12 +1045,15 @@ impl Session {
|
||||
if target.is_empty() {
|
||||
self.delete_message_batch(context, &uid_set, rowid_set)
|
||||
.await
|
||||
.with_context(|| format!("cannot delete batch of messages {uid_set:?}"))?;
|
||||
.with_context(|| format!("cannot delete batch of messages {:?}", &uid_set))?;
|
||||
} else {
|
||||
self.move_message_batch(context, &uid_set, rowid_set, &target)
|
||||
.await
|
||||
.with_context(|| {
|
||||
format!("cannot move batch of messages {uid_set:?} to folder {target:?}",)
|
||||
format!(
|
||||
"cannot move batch of messages {:?} to folder {:?}",
|
||||
&uid_set, target
|
||||
)
|
||||
})?;
|
||||
}
|
||||
}
|
||||
@@ -1284,10 +1287,9 @@ impl Session {
|
||||
|
||||
for (request_uids, set) in build_sequence_sets(&request_uids)? {
|
||||
info!(context, "Starting UID FETCH of message set \"{}\".", set);
|
||||
let mut fetch_responses = self
|
||||
.uid_fetch(&set, BODY_FULL)
|
||||
.await
|
||||
.with_context(|| format!("fetching messages {set} from folder {folder:?}"))?;
|
||||
let mut fetch_responses = self.uid_fetch(&set, BODY_FULL).await.with_context(|| {
|
||||
format!("fetching messages {} from folder \"{}\"", &set, folder)
|
||||
})?;
|
||||
|
||||
// Map from UIDs to unprocessed FETCH results. We put unprocessed FETCH results here
|
||||
// when we want to process other messages first.
|
||||
@@ -1501,7 +1503,7 @@ impl Session {
|
||||
.get_metadata(
|
||||
mailbox,
|
||||
options,
|
||||
"(/shared/comment /shared/admin /shared/vendor/deltachat/irohrelay /shared/vendor/deltachat/turn)",
|
||||
"(/shared/comment /shared/admin /shared/vendor/deltachat/irohrelay /shared/vendor/deltachat/turn /shared/vendor/deltachat/maxsmtprecipients)",
|
||||
)
|
||||
.await?;
|
||||
for m in metadata {
|
||||
@@ -1537,6 +1539,21 @@ impl Session {
|
||||
}
|
||||
}
|
||||
}
|
||||
"/shared/vendor/deltachat/maxsmtprecipients" => {
|
||||
if let Some(value) = m.value.and_then(|v| v.parse::<u32>().ok()) {
|
||||
let transport_id = self.transport_id();
|
||||
context
|
||||
.sql
|
||||
.execute(
|
||||
"UPDATE transports \
|
||||
SET max_smtp_rcpt_to=? WHERE id=?",
|
||||
(value, transport_id),
|
||||
)
|
||||
.await
|
||||
.log_err(context)
|
||||
.ok();
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -222,7 +222,7 @@ SELECT ?1, rfc724_mid, pre_rfc724_mid, timestamp, ?, ? FROM msgs WHERE id=?1
|
||||
} else {
|
||||
msg.timestamp_sort
|
||||
});
|
||||
ret += &format!("Received: {s}");
|
||||
ret += &format!("Received: {}", &s);
|
||||
ret += "\n";
|
||||
}
|
||||
|
||||
@@ -301,7 +301,7 @@ SELECT ?1, rfc724_mid, pre_rfc724_mid, timestamp, ?, ? FROM msgs WHERE id=?1
|
||||
ret += "Type: ";
|
||||
ret += &format!("{}", msg.viewtype);
|
||||
ret += "\n";
|
||||
ret += &format!("Mimetype: {}\n", msg.get_filemime().unwrap_or_default());
|
||||
ret += &format!("Mimetype: {}\n", &msg.get_filemime().unwrap_or_default());
|
||||
}
|
||||
let w = msg.param.get_int(Param::Width).unwrap_or_default();
|
||||
let h = msg.param.get_int(Param::Height).unwrap_or_default();
|
||||
|
||||
@@ -332,7 +332,7 @@ fn inner_generate_secure_join_qr_code(
|
||||
d.attr("cx", logo_position_x + HALF_LOGO_SIZE)?;
|
||||
d.attr("cy", logo_position_y + HALF_LOGO_SIZE)?;
|
||||
d.attr("r", HALF_LOGO_SIZE)?;
|
||||
d.attr("style", format!("fill:{color}"))
|
||||
d.attr("style", format!("fill:{}", &color))
|
||||
})?;
|
||||
|
||||
let avatar_font_size = LOGO_SIZE * 0.65;
|
||||
|
||||
@@ -3560,7 +3560,12 @@ async fn create_or_lookup_mailinglist_or_broadcast(
|
||||
mime_parser.timestamp_sent,
|
||||
)
|
||||
.await
|
||||
.with_context(|| format!("failed to create mailinglist '{name}' for grpid={listid}",))?;
|
||||
.with_context(|| {
|
||||
format!(
|
||||
"failed to create mailinglist '{}' for grpid={}",
|
||||
&name, &listid
|
||||
)
|
||||
})?;
|
||||
|
||||
if chattype == Chattype::InBroadcast {
|
||||
chat::add_to_chat_contacts_table(
|
||||
|
||||
@@ -218,7 +218,7 @@ pub(crate) fn maybe_network_lost(context: &Context, stores: Vec<ConnectivityStor
|
||||
impl fmt::Debug for ConnectivityStore {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
if let Some(guard) = self.0.try_lock() {
|
||||
write!(f, "ConnectivityStore {:?}", *guard)
|
||||
write!(f, "ConnectivityStore {:?}", &*guard)
|
||||
} else {
|
||||
write!(f, "ConnectivityStore [LOCKED]")
|
||||
}
|
||||
|
||||
@@ -73,7 +73,8 @@ fn shorten_name(name: &str, length: usize) -> String {
|
||||
// We use _ rather than ... to avoid dots at the end of the URL, which would confuse linkifiers
|
||||
format!(
|
||||
"{}_",
|
||||
name.chars()
|
||||
&name
|
||||
.chars()
|
||||
.take(length.saturating_sub(1))
|
||||
.collect::<String>()
|
||||
)
|
||||
|
||||
@@ -992,7 +992,7 @@ async fn test_wrong_auth_token() -> Result<()> {
|
||||
tcm.send_recv(alice, bob, "hi").await;
|
||||
|
||||
let alice_qr = get_securejoin_qr(alice, None).await?;
|
||||
println!("{alice_qr}");
|
||||
println!("{}", &alice_qr);
|
||||
let invalid_alice_qr = alice_qr.replace("&s=", "&s=INVALIDAUTHTOKEN&someotherkey=");
|
||||
|
||||
join_securejoin(bob, &invalid_alice_qr).await?;
|
||||
|
||||
155
src/smtp.rs
155
src/smtp.rs
@@ -2,6 +2,8 @@
|
||||
|
||||
mod connect;
|
||||
pub mod send;
|
||||
#[cfg(test)]
|
||||
mod chunking_tests;
|
||||
|
||||
use anyhow::{Context as _, Error, Result, bail, format_err};
|
||||
use async_smtp::response::{Category, Code, Detail};
|
||||
@@ -10,6 +12,7 @@ use tokio::task;
|
||||
|
||||
use crate::chat::{ChatId, add_info_msg_with_cmd};
|
||||
use crate::config::Config;
|
||||
use crate::constants;
|
||||
use crate::contact::{Contact, ContactId};
|
||||
use crate::context::Context;
|
||||
use crate::events::EventType;
|
||||
@@ -34,6 +37,9 @@ pub(crate) struct Smtp {
|
||||
/// Email address we are sending from.
|
||||
from: Option<EmailAddress>,
|
||||
|
||||
/// Transport used for the current connection.
|
||||
transport_id: Option<u32>,
|
||||
|
||||
/// Timestamp of last successful send/receive network interaction
|
||||
/// (eg connect or send succeeded). On initialization and disconnect
|
||||
/// it is set to None.
|
||||
@@ -60,6 +66,7 @@ impl Smtp {
|
||||
task::spawn(async move { transport.quit().await });
|
||||
}
|
||||
self.last_success = None;
|
||||
self.transport_id = None;
|
||||
}
|
||||
|
||||
/// Return true if smtp was connected but is not known to
|
||||
@@ -89,9 +96,10 @@ impl Smtp {
|
||||
}
|
||||
|
||||
self.connectivity.set_connecting(context);
|
||||
let (_transport_id, lp) = ConfiguredLoginParam::load(context)
|
||||
let (transport_id, lp) = ConfiguredLoginParam::load(context)
|
||||
.await?
|
||||
.context("Not configured")?;
|
||||
self.transport_id = Some(transport_id);
|
||||
let proxy_config = ProxyConfig::load(context).await?;
|
||||
self.connect(
|
||||
context,
|
||||
@@ -165,6 +173,7 @@ impl Smtp {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum SendResult {
|
||||
/// Message was sent successfully.
|
||||
Success,
|
||||
@@ -176,13 +185,36 @@ pub(crate) enum SendResult {
|
||||
Retry,
|
||||
}
|
||||
|
||||
pub(crate) trait SmtpSender: Send {
|
||||
fn send_chunk<'a>(
|
||||
&'a mut self,
|
||||
context: &'a Context,
|
||||
recipients: &'a [async_smtp::EmailAddress],
|
||||
body: &'a str,
|
||||
) -> futures::future::BoxFuture<'a, SendResult>;
|
||||
}
|
||||
|
||||
struct RealSmtpSender<'a> {
|
||||
smtp: &'a mut Smtp,
|
||||
}
|
||||
|
||||
impl SmtpSender for RealSmtpSender<'_> {
|
||||
fn send_chunk<'a>(
|
||||
&'a mut self,
|
||||
context: &'a Context,
|
||||
recipients: &'a [async_smtp::EmailAddress],
|
||||
body: &'a str,
|
||||
) -> futures::future::BoxFuture<'a, SendResult> {
|
||||
Box::pin(smtp_send(context, recipients, body, self.smtp))
|
||||
}
|
||||
}
|
||||
|
||||
/// Tries to send a message.
|
||||
pub(crate) async fn smtp_send(
|
||||
context: &Context,
|
||||
recipients: &[async_smtp::EmailAddress],
|
||||
message: &str,
|
||||
smtp: &mut Smtp,
|
||||
msg_id: Option<MsgId>,
|
||||
) -> SendResult {
|
||||
if recipients.is_empty() {
|
||||
return SendResult::Success;
|
||||
@@ -310,25 +342,6 @@ pub(crate) async fn smtp_send(
|
||||
Ok(()) => SendResult::Success,
|
||||
};
|
||||
|
||||
if let SendResult::Failure(err) = &status
|
||||
&& let Some(msg_id) = msg_id
|
||||
{
|
||||
// We couldn't send the message, so mark it as failed
|
||||
match Message::load_from_db(context, msg_id).await {
|
||||
Ok(mut msg) => {
|
||||
if let Err(err) = message::set_msg_failed(context, &mut msg, &err.to_string()).await
|
||||
{
|
||||
error!(context, "Failed to mark {msg_id} as failed: {err:#}.");
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
error!(
|
||||
context,
|
||||
"Failed to load {msg_id} to mark it as failed: {err:#}."
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
status
|
||||
}
|
||||
|
||||
@@ -406,7 +419,40 @@ pub(crate) async fn send_msg_to_smtp(
|
||||
)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let status = smtp_send(context, &recipients_list, body.as_str(), smtp, Some(msg_id)).await;
|
||||
let transport_id = smtp
|
||||
.transport_id
|
||||
.context("SMTP not connected to a transport")?;
|
||||
let chunk_size = max_smtp_rcpt_to(context, transport_id).await?;
|
||||
|
||||
let mut sender = RealSmtpSender { smtp };
|
||||
let (status, start_idx) = send_smtp_chunks(
|
||||
context,
|
||||
&recipients_list,
|
||||
body.as_str(),
|
||||
chunk_size,
|
||||
&mut sender,
|
||||
)
|
||||
.await;
|
||||
|
||||
let unsent_saved = start_idx < recipients_list.len();
|
||||
if let Some(unsent) = recipients_list.get(start_idx..)
|
||||
&& !unsent.is_empty()
|
||||
{
|
||||
let unsent_str: String = unsent
|
||||
.iter()
|
||||
.map(|a| a.as_ref())
|
||||
.collect::<Vec<&str>>()
|
||||
.join(" ");
|
||||
context
|
||||
.sql
|
||||
.execute(
|
||||
"UPDATE smtp SET recipients=? WHERE id=?",
|
||||
(unsent_str, rowid),
|
||||
)
|
||||
.await
|
||||
.log_err(context)
|
||||
.ok();
|
||||
}
|
||||
|
||||
match status {
|
||||
SendResult::Retry => {}
|
||||
@@ -455,10 +501,15 @@ pub(crate) async fn send_msg_to_smtp(
|
||||
.await?;
|
||||
};
|
||||
}
|
||||
context
|
||||
.sql
|
||||
.execute("DELETE FROM smtp WHERE id=?", (rowid,))
|
||||
.await?;
|
||||
if let Some(mut msg) = Message::load_from_db_optional(context, msg_id).await? {
|
||||
message::set_msg_failed(context, &mut msg, &err.to_string()).await?;
|
||||
}
|
||||
if !unsent_saved {
|
||||
context
|
||||
.sql
|
||||
.execute("DELETE FROM smtp WHERE id=?", (rowid,))
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
@@ -470,10 +521,39 @@ pub(crate) async fn send_msg_to_smtp(
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
SendResult::Failure(err) => Err(format_err!("{err}")),
|
||||
SendResult::Failure(err) => {
|
||||
if unsent_saved {
|
||||
Err(format_err!("Retry"))
|
||||
} else {
|
||||
Err(format_err!("{err}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn max_smtp_rcpt_to(context: &Context, transport_id: u32) -> Result<usize> {
|
||||
let limit = context
|
||||
.sql
|
||||
.query_row_optional(
|
||||
"SELECT max_smtp_rcpt_to FROM transports WHERE id=?",
|
||||
(transport_id,),
|
||||
|row| row.get::<_, u32>(0),
|
||||
)
|
||||
.await?
|
||||
.unwrap_or(0);
|
||||
|
||||
if limit > 0 {
|
||||
return Ok(limit as usize);
|
||||
}
|
||||
|
||||
let val = context
|
||||
.get_configured_provider()
|
||||
.await?
|
||||
.and_then(|provider| provider.opt.max_smtp_rcpt_to)
|
||||
.map_or(constants::DEFAULT_MAX_SMTP_RCPT_TO, usize::from);
|
||||
Ok(val)
|
||||
}
|
||||
|
||||
pub(crate) async fn msg_has_pending_smtp_job(
|
||||
context: &Context,
|
||||
msg_id: MsgId,
|
||||
@@ -600,7 +680,7 @@ async fn send_mdn_rfc724_mid(
|
||||
})
|
||||
.collect();
|
||||
|
||||
match smtp_send(context, &recipients, &body, smtp, None).await {
|
||||
match smtp_send(context, &recipients, &body, smtp).await {
|
||||
SendResult::Success => {
|
||||
if !recipients.is_empty() {
|
||||
info!(context, "Successfully sent MDN for {rfc724_mid}.");
|
||||
@@ -722,3 +802,22 @@ pub(crate) async fn add_self_recipients(
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[allow(clippy::arithmetic_side_effects)]
|
||||
pub(crate) async fn send_smtp_chunks(
|
||||
context: &Context,
|
||||
recipients: &[async_smtp::EmailAddress],
|
||||
body: &str,
|
||||
chunk_size: usize,
|
||||
sender: &mut (dyn SmtpSender + Send),
|
||||
) -> (SendResult, usize) {
|
||||
for (i, chunk) in recipients.chunks(chunk_size).enumerate() {
|
||||
let status = sender.send_chunk(context, chunk, body).await;
|
||||
match status {
|
||||
SendResult::Success => continue,
|
||||
SendResult::Failure(_) => return (status, (i + 1) * chunk_size),
|
||||
SendResult::Retry => return (status, i * chunk_size),
|
||||
}
|
||||
}
|
||||
(SendResult::Success, recipients.len())
|
||||
}
|
||||
|
||||
102
src/smtp/chunking_tests.rs
Normal file
102
src/smtp/chunking_tests.rs
Normal file
@@ -0,0 +1,102 @@
|
||||
use crate::smtp::{send_smtp_chunks, SendResult, SmtpSender};
|
||||
use crate::test_utils::TestContextManager;
|
||||
use crate::context::Context;
|
||||
use anyhow::Result;
|
||||
use futures::future::{BoxFuture, FutureExt};
|
||||
|
||||
/// Result the mock should return on the designated call.
|
||||
enum MockFailure {
|
||||
Transient,
|
||||
Permanent,
|
||||
}
|
||||
|
||||
struct MockSmtpSender {
|
||||
call_count: usize,
|
||||
fail_on_call: Option<(usize, MockFailure)>,
|
||||
}
|
||||
|
||||
impl SmtpSender for MockSmtpSender {
|
||||
fn send_chunk<'a>(
|
||||
&'a mut self,
|
||||
_context: &'a Context,
|
||||
_recipients: &'a [async_smtp::EmailAddress],
|
||||
_body: &'a str,
|
||||
) -> BoxFuture<'a, SendResult> {
|
||||
self.call_count += 1;
|
||||
let count = self.call_count;
|
||||
let fail_on = self.fail_on_call.as_ref().map(|(n, _)| *n);
|
||||
let is_permanent = matches!(
|
||||
self.fail_on_call,
|
||||
Some((_, MockFailure::Permanent))
|
||||
);
|
||||
async move {
|
||||
if fail_on == Some(count) {
|
||||
if is_permanent {
|
||||
return SendResult::Failure(
|
||||
anyhow::format_err!("permanent error"),
|
||||
);
|
||||
}
|
||||
return SendResult::Retry;
|
||||
}
|
||||
SendResult::Success
|
||||
}
|
||||
.boxed()
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_send_smtp_chunks() -> Result<()> {
|
||||
let mut tcm = TestContextManager::new();
|
||||
let alice = tcm.alice().await;
|
||||
|
||||
let recipients: Vec<_> = ["r1@ex.org", "r2@ex.org", "r3@ex.org", "r4@ex.org", "r5@ex.org"]
|
||||
.iter()
|
||||
.map(|a| async_smtp::EmailAddress::new(a.to_string()).unwrap())
|
||||
.collect();
|
||||
|
||||
// All chunks succeed.
|
||||
let mut sender = MockSmtpSender { call_count: 0, fail_on_call: None };
|
||||
let (status, processed) =
|
||||
send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
|
||||
assert!(matches!(status, SendResult::Success));
|
||||
assert_eq!(processed, 5);
|
||||
assert_eq!(sender.call_count, 3); // chunks: [2, 2, 1]
|
||||
|
||||
// Second chunk gets a transient error, only first chunk's recipients are processed.
|
||||
let mut sender =
|
||||
MockSmtpSender { call_count: 0, fail_on_call: Some((2, MockFailure::Transient)) };
|
||||
let (status, processed) =
|
||||
send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
|
||||
assert!(matches!(status, SendResult::Retry));
|
||||
assert_eq!(processed, 2);
|
||||
assert_eq!(sender.call_count, 2);
|
||||
|
||||
// Last chunk gets a transient error, first two chunks' recipients are processed.
|
||||
let mut sender =
|
||||
MockSmtpSender { call_count: 0, fail_on_call: Some((3, MockFailure::Transient)) };
|
||||
let (status, processed) =
|
||||
send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
|
||||
assert!(matches!(status, SendResult::Retry));
|
||||
assert_eq!(processed, 4);
|
||||
assert_eq!(sender.call_count, 3);
|
||||
|
||||
// Second chunk gets a permanent error; processed includes the failed chunk.
|
||||
let mut sender =
|
||||
MockSmtpSender { call_count: 0, fail_on_call: Some((2, MockFailure::Permanent)) };
|
||||
let (status, processed) =
|
||||
send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
|
||||
assert!(matches!(status, SendResult::Failure(_)));
|
||||
assert_eq!(processed, 4);
|
||||
assert_eq!(sender.call_count, 2);
|
||||
|
||||
// Last chunk gets a permanent error; processed includes the failed chunk.
|
||||
let mut sender =
|
||||
MockSmtpSender { call_count: 0, fail_on_call: Some((3, MockFailure::Permanent)) };
|
||||
let (status, processed) =
|
||||
send_smtp_chunks(&alice.ctx, &recipients, "body", 2, &mut sender).await;
|
||||
assert!(matches!(status, SendResult::Failure(_)));
|
||||
assert_eq!(processed, 6); // capped at (i+1)*chunk_size, may exceed len
|
||||
assert_eq!(sender.call_count, 3);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -2385,6 +2385,15 @@ UPDATE msgs SET state=19 WHERE state=24; -- Change OutPreparing to OutFailed.
|
||||
.await?;
|
||||
}
|
||||
|
||||
inc_and_check(&mut migration_version, 153)?;
|
||||
if dbversion < migration_version {
|
||||
sql.execute_migration(
|
||||
"ALTER TABLE transports ADD COLUMN max_smtp_rcpt_to INTEGER NOT NULL DEFAULT 0",
|
||||
migration_version,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
let new_version = sql
|
||||
.get_raw_config_int(VERSION_CFG)
|
||||
.await?
|
||||
|
||||
@@ -1685,7 +1685,7 @@ async fn write_msg(context: &Context, prefix: &str, msg: &Message, buf: &mut Str
|
||||
msg.get_id(),
|
||||
if msg.get_showpadlock() { "🔒" } else { "" },
|
||||
if msg.has_location() { "📍" } else { "" },
|
||||
contact_name,
|
||||
&contact_name,
|
||||
contact_id,
|
||||
msgtext,
|
||||
if msg.get_from_id() == ContactId::SELF {
|
||||
|
||||
@@ -161,7 +161,7 @@ async fn check_that_transition_worked(
|
||||
2,
|
||||
"Group {} has members {:?}, but should have members {:?} and {:?}",
|
||||
group,
|
||||
members,
|
||||
&members,
|
||||
alice_contact_id,
|
||||
ContactId::SELF
|
||||
);
|
||||
|
||||
@@ -62,13 +62,13 @@ pub(crate) fn truncate(buf: &str, approx_chars: usize) -> Cow<'_, str> {
|
||||
if let Some(index) = buf.get(..end_pos).and_then(|s| s.rfind([' ', '\n'])) {
|
||||
Cow::Owned(format!(
|
||||
"{}{}",
|
||||
buf.get(..=index).unwrap_or_default(),
|
||||
&buf.get(..=index).unwrap_or_default(),
|
||||
DC_ELLIPSIS
|
||||
))
|
||||
} else {
|
||||
Cow::Owned(format!(
|
||||
"{}{}",
|
||||
buf.get(..end_pos).unwrap_or_default(),
|
||||
&buf.get(..end_pos).unwrap_or_default(),
|
||||
DC_ELLIPSIS
|
||||
))
|
||||
}
|
||||
|
||||
@@ -247,12 +247,12 @@ proptest! {
|
||||
assert!(
|
||||
l <= approx_chars + el_len,
|
||||
"buf: '{}' - res: '{}' - len {}, approx {}",
|
||||
buf, res, res.len(), approx_chars
|
||||
&buf, &res, res.len(), approx_chars
|
||||
);
|
||||
|
||||
if buf.chars().count() > approx_chars + el_len {
|
||||
let l = res.len();
|
||||
assert_eq!(&res[l-5..l], "[...]", "missing ellipsis in {res}");
|
||||
assert_eq!(&res[l-5..l], "[...]", "missing ellipsis in {}", &res);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -116,7 +116,7 @@ pub(crate) struct ConnectionCandidate {
|
||||
|
||||
impl fmt::Display for ConnectionCandidate {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}:{}:{}", self.host, self.port, self.security)?;
|
||||
write!(f, "{}:{}:{}", &self.host, self.port, self.security)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -131,7 +131,7 @@ pub(crate) struct ConfiguredServerLoginParam {
|
||||
|
||||
impl fmt::Display for ConfiguredServerLoginParam {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}:{}", self.connection, self.user)?;
|
||||
write!(f, "{}:{}", self.connection, &self.user)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user