use async-imap response handling branch

Cargo.lock

@@ -90,7 +90,7 @@ dependencies = [
 [[package]]
 name = "async-imap"
 version = "0.1.1"
-source = "git+https://github.com/async-email/async-imap#1327f678cf5515842fc309636459372e6ab40db2"
+source = "git+https://github.com/async-email/async-imap?branch=improved-response-handling#a79795381622e536f035a96c2c249272e9fbcece"
 dependencies = [
  "async-attributes 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "async-std 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -610,7 +610,7 @@ dependencies = [
 name = "deltachat"
 version = "1.0.0-beta.8"
 dependencies = [
- "async-imap 0.1.1 (git+https://github.com/async-email/async-imap)",
+ "async-imap 0.1.1 (git+https://github.com/async-email/async-imap?branch=improved-response-handling)",
  "async-std 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "async-tls 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "backtrace 0.3.40 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -3245,7 +3245,7 @@ dependencies = [
 "checksum arrayvec 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "cff77d8686867eceff3105329d4698d96c2391c176d5d03adc90c7389162b5b8"
 "checksum ascii_utils 0.9.3 (registry+https://github.com/rust-lang/crates.io-index)" = "71938f30533e4d95a6d17aa530939da3842c2ab6f4f84b9dae68447e4129f74a"
 "checksum async-attributes 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "efd3d156917d94862e779f356c5acae312b08fd3121e792c857d7928c8088423"
-"checksum async-imap 0.1.1 (git+https://github.com/async-email/async-imap)" = "<none>"
+"checksum async-imap 0.1.1 (git+https://github.com/async-email/async-imap?branch=improved-response-handling)" = "<none>"
 "checksum async-macros 2.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "644a5a8de80f2085a1e7e57cd1544a2a7438f6e003c0790999bd43b92a77cdb2"
 "checksum async-std 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "56933da6903b273923d13f4746d829f66ff9b444173f6743d831e80f4da15446"
 "checksum async-task 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "de6bd58f7b9cc49032559422595c81cbfcf04db2f2133592f70af19e258a1ced"

Cargo.toml

@@ -19,7 +19,7 @@ reqwest = { version = "0.9.15", default-features = false, features = ["rustls-tl
 num-derive = "0.2.5"
 num-traits = "0.2.6"
 lettre = { git = "https://github.com/deltachat/lettre", branch = "feat/rustls" }
-async-imap = { git = "https://github.com/async-email/async-imap", branch="master" }
+async-imap = { git = "https://github.com/async-email/async-imap", branch="improved-response-handling" }
 async-tls = "0.6"
 async-std = { version = "1.0", features = ["unstable"] }
 base64 = "0.10"

src/job.rs

@@ -1,4 +1,3 @@
-use std::sync::{Arc, RwLock};
 use std::time::Duration;
 
 use deltachat_derive::{FromSql, ToSql};
@@ -15,7 +14,6 @@ use crate::error::Error;
 use crate::events::Event;
 use crate::imap::*;
 use crate::imex::*;
-use crate::job_thread::JobThread;
 use crate::location;
 use crate::login_param::LoginParam;
 use crate::message::MsgId;
@@ -33,21 +31,6 @@ enum Thread {
     Smtp = 5000,
 }
 
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-enum ThreadExecutor {
-    Inbox,
-    Mvbox,
-    Sentbox,
-    Smtp,
-}
-
-#[derive(Debug, Display, Copy, Clone, PartialEq, Eq, FromPrimitive, ToPrimitive)]
-enum TryAgain {
-    Dont,
-    AtOnce,
-    StandardDelay,
-}
-
 impl Default for Thread {
     fn default() -> Self {
         Thread::Unknown
@@ -119,7 +102,7 @@ pub struct Job {
     pub added_timestamp: i64,
     pub tries: i32,
     pub param: Params,
-    try_again: TryAgain,
+    pub try_again: i32,
     pub pending_error: Option<String>,
 }
 
@@ -153,7 +136,7 @@ impl Job {
             let loginparam = LoginParam::from_database(context, "configured_");
             let connected = context.smtp.lock().unwrap().connect(context, &loginparam);
             if connected.is_err() {
-                self.try_again_later(TryAgain::StandardDelay, None);
+                self.try_again_later(3, None);
                 return;
             }
         }
@@ -194,7 +177,7 @@ impl Job {
                         Err(err) => {
                             smtp.disconnect();
                             warn!(context, "smtp failed: {}", err);
-                            self.try_again_later(TryAgain::AtOnce, Some(err.to_string()));
+                            self.try_again_later(-1, Some(err.to_string()));
                         }
                         Ok(()) => {
                             // smtp success, update db ASAP, then delete smtp file
@@ -213,15 +196,14 @@ impl Job {
     }
 
     // this value does not increase the number of tries
-    fn try_again_later(&mut self, try_again: TryAgain, pending_error: Option<String>) {
+    fn try_again_later(&mut self, try_again: i32, pending_error: Option<String>) {
         self.try_again = try_again;
         self.pending_error = pending_error;
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_MOVE_MSG(&mut self, context: &Context, thread: ThreadExecutor) {
-        let t = get_mailbox_for_executor(context, thread);
-        let imap_inbox = &t.read().unwrap().imap;
+    fn do_DC_JOB_MOVE_MSG(&mut self, context: &Context) {
+        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
 
         if let Ok(msg) = Message::load_from_db(context, MsgId::new(self.foreign_id)) {
             if context
@@ -248,7 +230,7 @@ impl Job {
                     &mut dest_uid,
                 ) {
                     ImapActionResult::RetryLater => {
-                        self.try_again_later(TryAgain::StandardDelay, None);
+                        self.try_again_later(3i32, None);
                     }
                     ImapActionResult::Success => {
                         message::update_server_uid(
@@ -265,9 +247,8 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_DELETE_MSG_ON_IMAP(&mut self, context: &Context, thread: ThreadExecutor) {
-        let t = get_mailbox_for_executor(context, thread);
-        let imap_inbox = &t.read().unwrap().imap;
+    fn do_DC_JOB_DELETE_MSG_ON_IMAP(&mut self, context: &Context) {
+        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
 
         if let Ok(mut msg) = Message::load_from_db(context, MsgId::new(self.foreign_id)) {
             if !msg.rfc724_mid.is_empty() {
@@ -285,7 +266,7 @@ impl Job {
                     let res =
                         imap_inbox.delete_msg(context, &mid, server_folder, &mut msg.server_uid);
                     if res == ImapActionResult::RetryLater {
-                        self.try_again_later(TryAgain::AtOnce, None);
+                        self.try_again_later(-1i32, None);
                         return;
                     }
                 }
@@ -295,32 +276,30 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_EMPTY_SERVER(&mut self, context: &Context, thread: ThreadExecutor) {
-        let t = get_mailbox_for_executor(context, thread);
-        let imap_inbox = &t.read().unwrap().imap;
-
-        if thread == ThreadExecutor::Mvbox && DC_EMPTY_MVBOX > 0 {
+    fn do_DC_JOB_EMPTY_SERVER(&mut self, context: &Context) {
+        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
+        if self.foreign_id & DC_EMPTY_MVBOX > 0 {
             if let Some(mvbox_folder) = context
                 .sql
                 .get_raw_config(context, "configured_mvbox_folder")
             {
                 imap_inbox.empty_folder(context, &mvbox_folder);
             }
-        } else if thread == ThreadExecutor::Inbox && DC_EMPTY_INBOX > 0 {
+        }
+        if self.foreign_id & DC_EMPTY_INBOX > 0 {
             imap_inbox.empty_folder(context, "INBOX");
         }
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_MARKSEEN_MSG_ON_IMAP(&mut self, context: &Context, thread: ThreadExecutor) {
-        let t = get_mailbox_for_executor(context, thread);
-        let imap_inbox = &t.read().unwrap().imap;
+    fn do_DC_JOB_MARKSEEN_MSG_ON_IMAP(&mut self, context: &Context) {
+        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
 
         if let Ok(msg) = Message::load_from_db(context, MsgId::new(self.foreign_id)) {
             let folder = msg.server_folder.as_ref().unwrap();
             match imap_inbox.set_seen(context, folder, msg.server_uid) {
                 ImapActionResult::RetryLater => {
-                    self.try_again_later(TryAgain::StandardDelay, None);
+                    self.try_again_later(3i32, None);
                 }
                 ImapActionResult::AlreadyDone => {}
                 ImapActionResult::Success | ImapActionResult::Failed => {
@@ -341,19 +320,16 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_MARKSEEN_MDN_ON_IMAP(&mut self, context: &Context, thread: ThreadExecutor) {
+    fn do_DC_JOB_MARKSEEN_MDN_ON_IMAP(&mut self, context: &Context) {
         let folder = self
             .param
             .get(Param::ServerFolder)
             .unwrap_or_default()
             .to_string();
         let uid = self.param.get_int(Param::ServerUid).unwrap_or_default() as u32;
-
-        let t = get_mailbox_for_executor(context, thread);
-        let imap_inbox = &t.read().unwrap().imap;
-
+        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
         if imap_inbox.set_seen(context, &folder, uid) == ImapActionResult::RetryLater {
-            self.try_again_later(TryAgain::StandardDelay, None);
+            self.try_again_later(3i32, None);
             return;
         }
         if 0 != self.param.get_int(Param::AlsoMove).unwrap_or_default() {
@@ -373,7 +349,7 @@ impl Job {
                 if ImapActionResult::RetryLater
                     == imap_inbox.mv(context, &folder, uid, &dest_folder, &mut dest_uid)
                 {
-                    self.try_again_later(TryAgain::StandardDelay, None);
+                    self.try_again_later(3, None);
                 }
             }
         }
@@ -505,7 +481,7 @@ pub fn perform_smtp_jobs(context: &Context) {
     };
 
     info!(context, "SMTP-jobs started...",);
-    job_perform(context, ThreadExecutor::Smtp, probe_smtp_network);
+    job_perform(context, Thread::Smtp, probe_smtp_network);
     info!(context, "SMTP-jobs ended.");
 
     {
@@ -718,81 +694,40 @@ pub fn perform_inbox_jobs(context: &Context) {
     *context.probe_imap_network.write().unwrap() = false;
     *context.perform_inbox_jobs_needed.write().unwrap() = false;
 
-    job_perform(context, ThreadExecutor::Inbox, probe_imap_network);
+    job_perform(context, Thread::Imap, probe_imap_network);
     info!(context, "dc_perform_inbox_jobs ended.",);
 }
 
 pub fn perform_mvbox_jobs(context: &Context) {
-    info!(context, "dc_perform_mbox_jobs starting.",);
-
-    let probe_imap_network = *context.probe_imap_network.clone().read().unwrap();
-    *context.probe_imap_network.write().unwrap() = false;
-    *context.perform_inbox_jobs_needed.write().unwrap() = false;
-
-    job_perform(context, ThreadExecutor::Mvbox, probe_imap_network);
-    info!(context, "dc_perform_mbox_jobs ended.",);
+    info!(context, "dc_perform_mbox_jobs EMPTY (for now).",);
 }
 
 pub fn perform_sentbox_jobs(context: &Context) {
-    info!(context, "dc_perform_sentbox_jobs starting.",);
-
-    let probe_imap_network = *context.probe_imap_network.clone().read().unwrap();
-    *context.probe_imap_network.write().unwrap() = false;
-    *context.perform_inbox_jobs_needed.write().unwrap() = false;
-
-    job_perform(context, ThreadExecutor::Sentbox, probe_imap_network);
-    info!(context, "dc_perform_sentbox_jobs ended.",);
+    info!(context, "dc_perform_sentbox_jobs EMPTY (for now).",);
 }
 
-fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
-    let folder = get_folder_for_executor(context, thread);
-
-    let fallback = if thread == ThreadExecutor::Inbox || thread == ThreadExecutor::Smtp {
-        // only select non empty ones on the inbox thread or smtp
-        "OR (j.foreign_id IS NULL)) "
-    } else {
-        ") "
-    };
-
-    let query_start = concat!(
-        "SELECT ",
-        "j.id AS id, ",
-        "j.action AS action, ",
-        "j.foreign_id AS foreign_id, ",
-        "j.param AS param, ",
-        "j.added_timestamp AS added_timestamp, ",
-        "j.desired_timestamp AS desired_timestamp, ",
-        "j.tries AS tries ",
-        "FROM jobs j ",
-        "LEFT JOIN msgs m ON j.foreign_id=m.id ",
-    )
-    .to_string();
+fn job_perform(context: &Context, thread: Thread, probe_network: bool) {
     let query = if !probe_network {
         // processing for first-try and after backoff-timeouts:
         // process jobs in the order they were added.
-        query_start
-            + "WHERE j.thread=? AND j.desired_timestamp<=? AND "
-            + "((j.foreign_id IS NOT NULL AND m.server_folder=?) "
-            + fallback
-            + "ORDER BY j.action DESC, j.added_timestamp;"
+        "SELECT id, action, foreign_id, param, added_timestamp, desired_timestamp, tries \
+         FROM jobs WHERE thread=? AND desired_timestamp<=? ORDER BY action DESC, added_timestamp;"
     } else {
         // processing after call to dc_maybe_network():
         // process _all_ pending jobs that failed before
         // in the order of their backoff-times.
-        query_start
-            + "WHERE j.thread=? AND j.tries>0 AND "
-            + "((j.foreign_id IS NOT NULL AND m.server_folder=?) "
-            + fallback
-            + "ORDER BY j.desired_timestamp, j.action DESC;"
+        "SELECT id, action, foreign_id, param, added_timestamp, desired_timestamp, tries \
+         FROM jobs WHERE thread=? AND tries>0 ORDER BY desired_timestamp, action DESC;"
     };
 
-    let params_no_probe = params![thread as i64, time(), folder];
-    let params_probe = params![thread as i64, folder];
+    let params_no_probe = params![thread as i64, time()];
+    let params_probe = params![thread as i64];
     let params: &[&dyn rusqlite::ToSql] = if !probe_network {
         params_no_probe
     } else {
         params_probe
     };
+
     let jobs: Result<Vec<Job>, _> = context
         .sql
         .query_map(
@@ -807,7 +742,7 @@ fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
                     added_timestamp: row.get(4)?,
                     tries: row.get(6)?,
                     param: row.get::<_, String>(3)?.parse().unwrap_or_default(),
-                    try_again: TryAgain::Dont,
+                    try_again: 0,
                     pending_error: None,
                 };
 
@@ -822,7 +757,14 @@ fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
     for mut job in jobs.unwrap_or_default() {
         info!(
             context,
-            "{:?}-job #{}, action {} started...", thread, job.job_id, job.action,
+            "{}-job #{}, action {} started...",
+            if thread == Thread::Imap {
+                "INBOX"
+            } else {
+                "SMTP"
+            },
+            job.job_id,
+            job.action,
         );
 
         // some configuration jobs are "exclusive":
@@ -849,18 +791,18 @@ fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
         let mut tries = 0;
         while tries <= 1 {
             // this can be modified by a job using dc_job_try_again_later()
-            job.try_again = TryAgain::Dont;
+            job.try_again = 0;
 
             match job.action {
                 Action::Unknown => {
                     warn!(context, "Unknown job id found");
                 }
                 Action::SendMsgToSmtp => job.do_DC_JOB_SEND(context),
-                Action::EmptyServer => job.do_DC_JOB_EMPTY_SERVER(context, thread),
-                Action::DeleteMsgOnImap => job.do_DC_JOB_DELETE_MSG_ON_IMAP(context, thread),
-                Action::MarkseenMsgOnImap => job.do_DC_JOB_MARKSEEN_MSG_ON_IMAP(context, thread),
-                Action::MarkseenMdnOnImap => job.do_DC_JOB_MARKSEEN_MDN_ON_IMAP(context, thread),
-                Action::MoveMsg => job.do_DC_JOB_MOVE_MSG(context, thread),
+                Action::EmptyServer => job.do_DC_JOB_EMPTY_SERVER(context),
+                Action::DeleteMsgOnImap => job.do_DC_JOB_DELETE_MSG_ON_IMAP(context),
+                Action::MarkseenMsgOnImap => job.do_DC_JOB_MARKSEEN_MSG_ON_IMAP(context),
+                Action::MarkseenMdnOnImap => job.do_DC_JOB_MARKSEEN_MDN_ON_IMAP(context),
+                Action::MoveMsg => job.do_DC_JOB_MOVE_MSG(context),
                 Action::SendMdn => job.do_DC_JOB_SEND(context),
                 Action::ConfigureImap => dc_job_do_DC_JOB_CONFIGURE_IMAP(context),
                 Action::ImexImap => match job_do_DC_JOB_IMEX_IMAP(context, &job) {
@@ -879,7 +821,7 @@ fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
                 Action::SendMdnOld => {}
                 Action::SendMsgToSmtpOld => {}
             }
-            if job.try_again != TryAgain::AtOnce {
+            if job.try_again != -1 {
                 break;
             }
             tries += 1
@@ -899,7 +841,19 @@ fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
                 .unsuspend(context);
             suspend_smtp_thread(context, false);
             break;
-        } else if job.try_again == TryAgain::AtOnce || job.try_again == TryAgain::StandardDelay {
+        } else if job.try_again == 2 {
+            // just try over next loop unconditionally, the ui typically interrupts idle when the file (video) is ready
+            info!(
+                context,
+                "{}-job #{} not yet ready and will be delayed.",
+                if thread == Thread::Imap {
+                    "INBOX"
+                } else {
+                    "SMTP"
+                },
+                job.job_id
+            );
+        } else if job.try_again == -1 || job.try_again == 3 {
             let tries = job.tries + 1;
             if tries < 17 {
                 job.tries = tries;
@@ -908,14 +862,18 @@ fn job_perform(context: &Context, thread: ThreadExecutor, probe_network: bool) {
                 job.update(context);
                 info!(
                     context,
-                    "{:?}-job #{} not succeeded on try #{}, retry in ADD_TIME+{} (in {} seconds).",
-                    thread,
+                    "{}-job #{} not succeeded on try #{}, retry in ADD_TIME+{} (in {} seconds).",
+                    if thread == Thread::Imap {
+                        "INBOX"
+                    } else {
+                        "SMTP"
+                    },
                     job.job_id as u32,
                     tries,
                     time_offset,
                     job.added_timestamp + time_offset - time()
                 );
-                if thread == ThreadExecutor::Smtp && tries < 17 - 1 {
+                if thread == Thread::Smtp && tries < 17 - 1 {
                     context
                         .smtp_state
                         .clone()
@@ -1043,69 +1001,12 @@ pub fn job_add(
     ).ok();
 
     match thread {
-        Thread::Imap => {
-            let folder: String = context
-                .sql
-                .query_get_value(
-                    context,
-                    "SELECT server_folder FROM msgs WHERE id=?",
-                    params![foreign_id],
-                )
-                .unwrap_or_default();
-
-            let mbox_thread = get_matching_mailbox(context, &folder);
-            let res = mbox_thread.try_read();
-            match res {
-                Ok(mbox_thread) => {
-                    mbox_thread.interrupt_idle(context);
-                }
-                Err(err) => {
-                    *context.perform_inbox_jobs_needed.write().unwrap() = true;
-                    warn!(context, "could not interrupt idle: {}", err);
-                }
-            }
-        }
+        Thread::Imap => interrupt_inbox_idle(context, false),
         Thread::Smtp => interrupt_smtp_idle(context),
         Thread::Unknown => {}
     }
 }
 
-fn get_folder_for_executor(context: &Context, thread: ThreadExecutor) -> String {
-    let t = get_mailbox_for_executor(context, thread);
-    let b = t.read().unwrap();
-    b.get_watch_folder(context).unwrap_or_default().to_string()
-}
-
-fn get_mailbox_for_executor(context: &Context, thread: ThreadExecutor) -> Arc<RwLock<JobThread>> {
-    match thread {
-        ThreadExecutor::Inbox => context.inbox_thread.clone(),
-        ThreadExecutor::Mvbox => context.mvbox_thread.clone(),
-        ThreadExecutor::Sentbox => context.sentbox_thread.clone(),
-        ThreadExecutor::Smtp => panic!("do not use for smtp"),
-    }
-}
-
-fn get_matching_mailbox(context: &Context, folder: &String) -> Arc<RwLock<JobThread>> {
-    let mvbox_folder = context
-        .mvbox_thread
-        .read()
-        .unwrap()
-        .get_watch_folder(context);
-    let sentbox_folder = context
-        .sentbox_thread
-        .read()
-        .unwrap()
-        .get_watch_folder(context);
-
-    if mvbox_folder.is_some() && folder == mvbox_folder.as_ref().unwrap() {
-        context.mvbox_thread.clone()
-    } else if sentbox_folder.is_some() && folder == sentbox_folder.as_ref().unwrap() {
-        context.sentbox_thread.clone()
-    } else {
-        context.inbox_thread.clone()
-    }
-}
-
 pub fn interrupt_smtp_idle(context: &Context) {
     info!(context, "Interrupting SMTP-idle...",);
 

src/job_thread.rs

@@ -116,7 +116,7 @@ impl JobThread {
         }
     }
 
-    pub(crate) fn get_watch_folder(&self, context: &Context) -> Option<String> {
+    fn get_watch_folder(&self, context: &Context) -> Option<String> {
         match context.sql.get_raw_config(context, self.folder_config_name) {
             Some(name) => Some(name),
             None => {