Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-13 03:46:32 +03:00
Code Issues Packages Projects Releases Wiki Activity

Replace SendMdn job with smtp_mdns table

Browse Source 
Unlike jobs which are executed before sending normal messages, MDNs
from `smtp_mdns` table are sent after sending messages from `smtp`
table. This way normal messages have higher priority than MDNs.

There are no SMTP jobs anymore. All jobs are IMAP jobs, so
`jobs.thread` column is not used anymore.
This commit is contained in:
link2xt
2022-04-23 11:44:32 +00:00
parent 01ef053a11
commit 7bcb03f1ec
6 changed files with 254 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files

283
src/job.rs
View File

@@ -4,36 +4,22 @@
//! and job types.
use std::fmt;
use anyhow::{format_err, Context as _, Result};
use anyhow::{Context as _, Result};
use deltachat_derive::{FromSql, ToSql};
use rand::{thread_rng, Rng};
use crate::config::Config;
use crate::contact::{normalize_name, Contact, ContactId, Modifier, Origin};
use crate::contact::{normalize_name, Contact, Modifier, Origin};
use crate::context::Context;
use crate::dc_tools::time;
use crate::events::EventType;
use crate::imap::Imap;
use crate::message::{Message, MsgId};
use crate::mimefactory::MimeFactory;
use crate::param::{Param, Params};
use crate::param::Params;
use crate::scheduler::InterruptInfo;
use crate::smtp::{smtp_send, SendResult, Smtp};
use crate::sql;
// results in ~3 weeks for the last backoff timespan
const JOB_RETRIES: u32 = 17;
/// Thread IDs
#[derive(
Debug, Display, Copy, Clone, PartialEq, Eq, FromPrimitive, ToPrimitive, FromSql, ToSql,
)]
#[repr(u32)]
pub(crate) enum Thread {
Imap = 100,
Smtp = 5000,
}
/// Job try result.
#[derive(Debug, Display)]
pub enum Status {
@@ -72,7 +58,6 @@ macro_rules! job_try {
)]
#[repr(u32)]
pub enum Action {
// Jobs in the INBOX-thread, range from DC_IMAP_THREAD..DC_IMAP_THREAD+999
FetchExistingMsgs = 110,
// this is user initiated so it should have a fairly high priority
@@ -87,24 +72,6 @@ pub enum Action {
// UID synchronization is high-priority to make sure correct UIDs
// are used by message moving/deletion.
ResyncFolders = 300,
// Jobs in the SMTP-thread, range from DC_SMTP_THREAD..DC_SMTP_THREAD+999
SendMdn = 5010,
}
impl From<Action> for Thread {
fn from(action: Action) -> Thread {
use Action::*;
match action {
FetchExistingMsgs => Thread::Imap,
ResyncFolders => Thread::Imap,
UpdateRecentQuota => Thread::Imap,
DownloadMsg => Thread::Imap,
SendMdn => Thread::Smtp,
}
}
}
#[derive(Debug, Clone, PartialEq)]
@@ -159,9 +126,7 @@ impl Job {
///
/// The Job is consumed by this method.
pub(crate) async fn save(self, context: &Context) -> Result<()> {
let thread: Thread = self.action.into();
info!(context, "saving job for {}-thread: {:?}", thread, self);
info!(context, "saving job {:?}", self);
if self.job_id != 0 {
context
@@ -178,10 +143,9 @@ impl Job {
.await?;
} else {
context.sql.execute(
"INSERT INTO jobs (added_timestamp, thread, action, foreign_id, param, desired_timestamp) VALUES (?,?,?,?,?,?);",
"INSERT INTO jobs (added_timestamp, action, foreign_id, param, desired_timestamp) VALUES (?,?,?,?,?);",
paramsv![
self.added_timestamp,
thread,
self.action,
self.foreign_id,
self.param.to_string(),
@@ -193,118 +157,6 @@ impl Job {
Ok(())
}
/// Get `SendMdn` jobs with foreign_id equal to `contact_id` excluding the `job_id` job.
async fn get_additional_mdn_jobs(
&self,
context: &Context,
contact_id: ContactId,
) -> Result<(Vec<u32>, Vec<String>)> {
// Extract message IDs from job parameters
let res: Vec<(u32, MsgId)> = context
.sql
.query_map(
"SELECT id, param FROM jobs WHERE foreign_id=? AND id!=?",
paramsv![contact_id, self.job_id],
|row| {
let job_id: u32 = row.get(0)?;
let params_str: String = row.get(1)?;
let params: Params = params_str.parse().unwrap_or_default();
Ok((job_id, params))
},
|jobs| {
let res = jobs
.filter_map(|row| {
let (job_id, params) = row.ok()?;
let msg_id = params.get_msg_id()?;
Some((job_id, msg_id))
})
.collect();
Ok(res)
},
)
.await?;
// Load corresponding RFC724 message IDs
let mut job_ids = Vec::new();
let mut rfc724_mids = Vec::new();
for (job_id, msg_id) in res {
if let Ok(Message { rfc724_mid, .. }) = Message::load_from_db(context, msg_id).await {
job_ids.push(job_id);
rfc724_mids.push(rfc724_mid);
}
}
Ok((job_ids, rfc724_mids))
}
async fn send_mdn(&mut self, context: &Context, smtp: &mut Smtp) -> Status {
let mdns_enabled = job_try!(context.get_config_bool(Config::MdnsEnabled).await);
if !mdns_enabled {
// User has disabled MDNs after job scheduling but before
// execution.
return Status::Finished(Err(format_err!("MDNs are disabled")));
}
let contact_id = ContactId::new(self.foreign_id);
let contact = job_try!(Contact::load_from_db(context, contact_id).await);
if contact.is_blocked() {
return Status::Finished(Err(format_err!("Contact is blocked")));
}
let msg_id = if let Some(msg_id) = self.param.get_msg_id() {
msg_id
} else {
return Status::Finished(Err(format_err!(
"SendMdn job has invalid parameters: {}",
self.param
)));
};
// Try to aggregate other SendMdn jobs and send a combined MDN.
let (additional_job_ids, additional_rfc724_mids) = self
.get_additional_mdn_jobs(context, contact_id)
.await
.unwrap_or_default();
if !additional_rfc724_mids.is_empty() {
info!(
context,
"SendMdn job: aggregating {} additional MDNs",
additional_rfc724_mids.len()
)
}
let msg = job_try!(Message::load_from_db(context, msg_id).await);
let mimefactory =
job_try!(MimeFactory::from_mdn(context, &msg, additional_rfc724_mids).await);
let rendered_msg = job_try!(mimefactory.render(context).await);
let body = rendered_msg.message;
let addr = contact.get_addr();
let recipient = job_try!(async_smtp::EmailAddress::new(addr.to_string())
.map_err(|err| format_err!("invalid recipient: {} {:?}", addr, err)));
let recipients = vec![recipient];
// connect to SMTP server, if not yet done
if let Err(err) = smtp.connect_configured(context).await {
warn!(context, "SMTP connection failure: {:?}", err);
smtp.last_send_error = Some(err.to_string());
return Status::RetryLater;
}
match smtp_send(context, &recipients, &body, smtp, msg_id, 0).await {
SendResult::Success => {
// Remove additional SendMdn jobs we have aggregated into this one.
job_try!(kill_ids(context, &additional_job_ids).await);
Status::Finished(Ok(()))
}
SendResult::Retry => {
info!(context, "Temporary SMTP failure while sending an MDN");
Status::RetryLater
}
SendResult::Failure(err) => Status::Finished(Err(err)),
}
}
/// Read the recipients from old emails sent by the user and add them as contacts.
/// This way, we can already offer them some email addresses they can write to.
///
@@ -387,22 +239,6 @@ pub async fn kill_action(context: &Context, action: Action) -> Result<()> {
Ok(())
}
/// Remove jobs with specified IDs.
async fn kill_ids(context: &Context, job_ids: &[u32]) -> Result<()> {
if job_ids.is_empty() {
return Ok(());
}
let q = format!(
"DELETE FROM jobs WHERE id IN({})",
sql::repeat_vars(job_ids.len())
);
context
.sql
.execute(q, rusqlite::params_from_iter(job_ids))
.await?;
Ok(())
}
pub async fn action_exists(context: &Context, action: Action) -> Result<bool> {
let exists = context
.sql
@@ -461,36 +297,18 @@ async fn add_all_recipients_as_contacts(context: &Context, imap: &mut Imap, fold
pub(crate) enum Connection<'a> {
Inbox(&'a mut Imap),
Smtp(&'a mut Smtp),
}
impl<'a> fmt::Display for Connection<'a> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Connection::Inbox(_) => write!(f, "Inbox"),
Connection::Smtp(_) => write!(f, "Smtp"),
}
}
}
impl<'a> Connection<'a> {
fn inbox(&mut self) -> &mut Imap {
match self {
Connection::Inbox(imap) => imap,
_ => panic!("Not an inbox"),
}
}
fn smtp(&mut self) -> &mut Smtp {
match self {
Connection::Smtp(smtp) => smtp,
_ => panic!("Not a smtp"),
}
}
}
pub(crate) async fn perform_job(context: &Context, mut connection: Connection<'_>, mut job: Job) {
info!(context, "{}-job {} started...", &connection, &job);
info!(context, "job {} started...", &job);
let try_res = match perform_job_action(context, &mut job, &mut connection, 0).await {
Status::RetryNow => perform_job_action(context, &mut job, &mut connection, 1).await,
@@ -502,17 +320,13 @@ pub(crate) async fn perform_job(context: &Context, mut connection: Connection<'_
let tries = job.tries + 1;
if tries < JOB_RETRIES {
info!(
context,
"{} thread increases job {} tries to {}", &connection, job, tries
);
info!(context, "increase job {} tries to {}", job, tries);
job.tries = tries;
let time_offset = get_backoff_time_offset(tries, job.action);
job.desired_timestamp = time() + time_offset;
info!(
context,
"{}-job #{} not succeeded on try #{}, retry in {} seconds.",
&connection,
"job #{} not succeeded on try #{}, retry in {} seconds.",
job.job_id as u32,
tries,
time_offset
@@ -523,10 +337,7 @@ pub(crate) async fn perform_job(context: &Context, mut connection: Connection<'_
} else {
info!(
context,
"{} thread removes job {} as it exhausted {} retries",
&connection,
job,
JOB_RETRIES
"remove job {} as it exhausted {} retries", job, JOB_RETRIES
);
job.delete(context).await.unwrap_or_else(|err| {
error!(context, "failed to delete job: {}", err);
@@ -537,13 +348,10 @@ pub(crate) async fn perform_job(context: &Context, mut connection: Connection<'_
if let Err(err) = res {
warn!(
context,
"{} removes job {} as it failed with error {:#}", &connection, job, err
"remove job {} as it failed with error {:#}", job, err
);
} else {
info!(
context,
"{} removes job {} as it succeeded", &connection, job
);
info!(context, "remove job {} as it succeeded", job);
}
job.delete(context).await.unwrap_or_else(|err| {
@@ -559,13 +367,9 @@ async fn perform_job_action(
connection: &mut Connection<'_>,
tries: u32,
) -> Status {
info!(
context,
"{} begin immediate try {} of job {}", &connection, tries, job
);
info!(context, "begin immediate try {} of job {}", tries, job);
let try_res = match job.action {
Action::SendMdn => job.send_mdn(context, connection.smtp()).await,
Action::ResyncFolders => job.resync_folders(context, connection.inbox()).await,
Action::FetchExistingMsgs => job.fetch_existing_msgs(context, connection.inbox()).await,
Action::UpdateRecentQuota => match context.update_recent_quota(connection.inbox()).await {
@@ -600,19 +404,6 @@ fn get_backoff_time_offset(tries: u32, action: Action) -> i64 {
}
}
pub(crate) async fn send_mdn(context: &Context, msg_id: MsgId, from_id: ContactId) -> Result<()> {
let mut param = Params::new();
param.set(Param::MsgId, msg_id.to_u32().to_string());
add(
context,
Job::new(Action::SendMdn, from_id.to_u32(), param, 0),
)
.await?;
Ok(())
}
pub(crate) async fn schedule_resync(context: &Context) -> Result<()> {
kill_action(context, Action::ResyncFolders).await?;
add(
@@ -638,10 +429,6 @@ pub async fn add(context: &Context, job: Job) -> Result<()> {
info!(context, "interrupt: imap");
context.interrupt_inbox(InterruptInfo::new(false)).await;
}
Action::SendMdn => {
info!(context, "interrupt: smtp");
context.interrupt_smtp(InterruptInfo::new(false)).await;
}
}
}
Ok(())
@@ -649,21 +436,15 @@ pub async fn add(context: &Context, job: Job) -> Result<()> {
/// Load jobs from the database.
///
/// Load jobs for this "[Thread]", i.e. either load SMTP jobs or load
/// IMAP jobs. The `probe_network` parameter decides how to query
/// The `probe_network` parameter decides how to query
/// jobs, this is tricky and probably wrong currently. Look at the
/// SQL queries for details.
pub(crate) async fn load_next(
context: &Context,
thread: Thread,
info: &InterruptInfo,
) -> Result<Option<Job>> {
info!(context, "loading job for {}-thread", thread);
pub(crate) async fn load_next(context: &Context, info: &InterruptInfo) -> Result<Option<Job>> {
info!(context, "loading job");
let query;
let params;
let t = time();
let thread_i = thread as i64;
if !info.probe_network {
// processing for first-try and after backoff-timeouts:
@@ -671,11 +452,11 @@ pub(crate) async fn load_next(
query = r#"
SELECT id, action, foreign_id, param, added_timestamp, desired_timestamp, tries
FROM jobs
WHERE thread=? AND desired_timestamp<=?
WHERE desired_timestamp<=?
ORDER BY action DESC, added_timestamp
LIMIT 1;
"#;
params = paramsv![thread_i, t];
params = paramsv![t];
} else {
// processing after call to dc_maybe_network():
// process _all_ pending jobs that failed before
@@ -683,11 +464,11 @@ LIMIT 1;
query = r#"
SELECT id, action, foreign_id, param, added_timestamp, desired_timestamp, tries
FROM jobs
WHERE thread=? AND tries>0
WHERE tries>0
ORDER BY desired_timestamp, action DESC
LIMIT 1;
"#;
params = paramsv![thread_i];
params = paramsv![];
};
loop {
@@ -742,11 +523,10 @@ mod tests {
.sql
.execute(
"INSERT INTO jobs
(added_timestamp, thread, action, foreign_id, param, desired_timestamp)
VALUES (?, ?, ?, ?, ?, ?);",
(added_timestamp, action, foreign_id, param, desired_timestamp)
VALUES (?, ?, ?, ?, ?);",
paramsv![
now,
Thread::from(Action::DownloadMsg),
if valid {
Action::DownloadMsg as i32
} else {
@@ -768,21 +548,11 @@ mod tests {
// all jobs.
let t = TestContext::new().await;
insert_job(&t, 1, false).await; // This can not be loaded into Job struct.
let jobs = load_next(
&t,
Thread::from(Action::DownloadMsg),
&InterruptInfo::new(false),
)
.await?;
let jobs = load_next(&t, &InterruptInfo::new(false)).await?;
assert!(jobs.is_none());
insert_job(&t, 1, true).await;
let jobs = load_next(
&t,
Thread::from(Action::DownloadMsg),
&InterruptInfo::new(false),
)
.await?;
let jobs = load_next(&t, &InterruptInfo::new(false)).await?;
assert!(jobs.is_some());
Ok(())
}
@@ -793,12 +563,7 @@ mod tests {
insert_job(&t, 1, true).await;
let jobs = load_next(
&t,
Thread::from(Action::DownloadMsg),
&InterruptInfo::new(false),
)
.await?;
let jobs = load_next(&t, &InterruptInfo::new(false)).await?;
assert!(jobs.is_some());
Ok(())
}