more sql fixes

src/configure/mod.rs

@@ -69,6 +69,18 @@ pub fn dc_job_do_DC_JOB_CONFIGURE_IMAP(context: &Context) {
         .unwrap()
         .imap
         .disconnect(context);
+    context
+        .sentbox_thread
+        .read()
+        .unwrap()
+        .imap
+        .disconnect(context);
+    context
+        .mvbox_thread
+        .read()
+        .unwrap()
+        .imap
+        .disconnect(context);
     context.smtp.clone().lock().unwrap().disconnect();
     info!(context, "Configure ...",);
 

src/context.rs

@@ -498,6 +498,8 @@ pub struct BobStatus {
 #[derive(Default, Debug)]
 pub struct SmtpState {
     pub idle: bool,
+    pub suspended: bool,
+    pub doing_jobs: bool,
     pub perform_jobs_needed: i32,
     pub probe_network: bool,
 }

src/job.rs

@@ -1,3 +1,4 @@
+use std::sync::{Arc, RwLock};
 use std::time::Duration;
 
 use deltachat_derive::{FromSql, ToSql};
@@ -14,6 +15,7 @@ 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;
@@ -31,6 +33,14 @@ 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,
@@ -209,8 +219,9 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_MOVE_MSG(&mut self, context: &Context) {
-        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
+    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;
 
         if let Ok(msg) = Message::load_from_db(context, MsgId::new(self.foreign_id)) {
             if context
@@ -254,8 +265,9 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_DELETE_MSG_ON_IMAP(&mut self, context: &Context) {
-        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
+    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;
 
         if let Ok(mut msg) = Message::load_from_db(context, MsgId::new(self.foreign_id)) {
             if !msg.rfc724_mid.is_empty() {
@@ -283,24 +295,26 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    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 {
+    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 {
             if let Some(mvbox_folder) = context
                 .sql
                 .get_raw_config(context, "configured_mvbox_folder")
             {
                 imap_inbox.empty_folder(context, &mvbox_folder);
             }
-        }
-        if self.foreign_id & DC_EMPTY_INBOX > 0 {
+        } else if thread == ThreadExecutor::Inbox && 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) {
-        let imap_inbox = &context.inbox_thread.read().unwrap().imap;
+    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;
 
         if let Ok(msg) = Message::load_from_db(context, MsgId::new(self.foreign_id)) {
             let folder = msg.server_folder.as_ref().unwrap();
@@ -327,14 +341,17 @@ impl Job {
     }
 
     #[allow(non_snake_case)]
-    fn do_DC_JOB_MARKSEEN_MDN_ON_IMAP(&mut self, context: &Context) {
+    fn do_DC_JOB_MARKSEEN_MDN_ON_IMAP(&mut self, context: &Context, thread: ThreadExecutor) {
         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 imap_inbox = &context.inbox_thread.read().unwrap().imap;
+
+        let t = get_mailbox_for_executor(context, thread);
+        let imap_inbox = &t.read().unwrap().imap;
+
         if imap_inbox.set_seen(context, &folder, uid) == ImapActionResult::RetryLater {
             self.try_again_later(TryAgain::StandardDelay, None);
             return;
@@ -479,12 +496,24 @@ pub fn perform_smtp_jobs(context: &Context) {
         state.probe_network = false;
         state.perform_jobs_needed = 0;
 
+        if state.suspended {
+            info!(context, "SMTP-jobs suspended.",);
+            return;
+        }
+        state.doing_jobs = true;
         probe_smtp_network
     };
 
     info!(context, "SMTP-jobs started...",);
-    job_perform(context, Thread::Smtp, probe_smtp_network);
+    job_perform(context, ThreadExecutor::Smtp, probe_smtp_network);
     info!(context, "SMTP-jobs ended.");
+
+    {
+        let &(ref lock, _) = &*context.smtp_state.clone();
+        let mut state = lock.lock().unwrap();
+
+        state.doing_jobs = false;
+    }
 }
 
 pub fn perform_smtp_idle(context: &Context) {
@@ -689,40 +718,81 @@ 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, Thread::Imap, probe_imap_network);
+    job_perform(context, ThreadExecutor::Inbox, probe_imap_network);
     info!(context, "dc_perform_inbox_jobs ended.",);
 }
 
 pub fn perform_mvbox_jobs(context: &Context) {
-    info!(context, "dc_perform_mbox_jobs EMPTY (for now).",);
+    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.",);
 }
 
 pub fn perform_sentbox_jobs(context: &Context) {
-    info!(context, "dc_perform_sentbox_jobs EMPTY (for now).",);
+    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.",);
 }
 
-fn job_perform(context: &Context, thread: Thread, probe_network: bool) {
+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();
     let query = if !probe_network {
         // processing for first-try and after backoff-timeouts:
         // process jobs in the order they were added.
-        "SELECT id, action, foreign_id, param, added_timestamp, desired_timestamp, tries \
-         FROM jobs WHERE thread=? AND desired_timestamp<=? ORDER BY action DESC, added_timestamp;"
+        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;"
     } else {
         // processing after call to dc_maybe_network():
         // process _all_ pending jobs that failed before
         // in the order of their backoff-times.
-        "SELECT id, action, foreign_id, param, added_timestamp, desired_timestamp, tries \
-         FROM jobs WHERE thread=? AND tries>0 ORDER BY desired_timestamp, action DESC;"
+        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;"
     };
 
-    let params_no_probe = params![thread as i64, time()];
-    let params_probe = params![thread as i64];
+    let params_no_probe = params![thread as i64, time(), folder];
+    let params_probe = params![thread as i64, folder];
     let params: &[&dyn rusqlite::ToSql] = if !probe_network {
         params_no_probe
     } else {
         params_probe
     };
-
     let jobs: Result<Vec<Job>, _> = context
         .sql
         .query_map(
@@ -752,20 +822,28 @@ fn job_perform(context: &Context, thread: Thread, probe_network: bool) {
     for mut job in jobs.unwrap_or_default() {
         info!(
             context,
-            "{}-job #{}, action {} started...",
-            if thread == Thread::Imap {
-                "INBOX"
-            } else {
-                "SMTP"
-            },
-            job.job_id,
-            job.action,
+            "{:?}-job #{}, action {} started...", thread, job.job_id, job.action,
         );
 
         // some configuration jobs are "exclusive":
+        // - they are always executed in the imap-thread and the smtp-thread is suspended during execution
+        // - they may change the database handle change the database handle; we do not keep old pointers therefore
         // - they can be re-executed one time AT_ONCE, but they are not save in the database for later execution
         if Action::ConfigureImap == job.action || Action::ImexImap == job.action {
             job_kill_action(context, job.action);
+            context
+                .sentbox_thread
+                .clone()
+                .read()
+                .unwrap()
+                .suspend(context);
+            context
+                .mvbox_thread
+                .clone()
+                .read()
+                .unwrap()
+                .suspend(context);
+            suspend_smtp_thread(context, true);
         }
 
         let mut tries = 0;
@@ -778,11 +856,11 @@ fn job_perform(context: &Context, thread: Thread, probe_network: bool) {
                     warn!(context, "Unknown job id found");
                 }
                 Action::SendMsgToSmtp => job.do_DC_JOB_SEND(context),
-                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::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::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) {
@@ -807,6 +885,19 @@ fn job_perform(context: &Context, thread: Thread, probe_network: bool) {
             tries += 1
         }
         if Action::ConfigureImap == job.action || Action::ImexImap == job.action {
+            context
+                .sentbox_thread
+                .clone()
+                .read()
+                .unwrap()
+                .unsuspend(context);
+            context
+                .mvbox_thread
+                .clone()
+                .read()
+                .unwrap()
+                .unsuspend(context);
+            suspend_smtp_thread(context, false);
             break;
         } else if job.try_again == TryAgain::AtOnce || job.try_again == TryAgain::StandardDelay {
             let tries = job.tries + 1;
@@ -817,18 +908,14 @@ fn job_perform(context: &Context, thread: Thread, probe_network: bool) {
                 job.update(context);
                 info!(
                     context,
-                    "{}-job #{} not succeeded on try #{}, retry in ADD_TIME+{} (in {} seconds).",
-                    if thread == Thread::Imap {
-                        "INBOX"
-                    } else {
-                        "SMTP"
-                    },
+                    "{:?}-job #{} not succeeded on try #{}, retry in ADD_TIME+{} (in {} seconds).",
+                    thread,
                     job.job_id as u32,
                     tries,
                     time_offset,
                     job.added_timestamp + time_offset - time()
                 );
-                if thread == Thread::Smtp && tries < 17 - 1 {
+                if thread == ThreadExecutor::Smtp && tries < 17 - 1 {
                     context
                         .smtp_state
                         .clone()
@@ -874,6 +961,18 @@ fn get_backoff_time_offset(c_tries: libc::c_int) -> i64 {
     seconds as i64
 }
 
+fn suspend_smtp_thread(context: &Context, suspend: bool) {
+    context.smtp_state.0.lock().unwrap().suspended = suspend;
+    if suspend {
+        loop {
+            if !context.smtp_state.0.lock().unwrap().doing_jobs {
+                return;
+            }
+            std::thread::sleep(std::time::Duration::from_micros(300 * 1000));
+        }
+    }
+}
+
 fn send_mdn(context: &Context, msg_id: MsgId) -> Result<(), Error> {
     let mut mimefactory = MimeFactory::load_mdn(context, msg_id)?;
     unsafe { mimefactory.render()? };
@@ -944,12 +1043,69 @@ pub fn job_add(
     ).ok();
 
     match thread {
-        Thread::Imap => interrupt_inbox_idle(context, false),
+        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::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

@@ -16,6 +16,7 @@ pub struct JobThread {
 pub struct JobState {
     idle: bool,
     jobs_needed: bool,
+    suspended: bool,
     using_handle: bool,
 }
 
@@ -29,6 +30,32 @@ impl JobThread {
         }
     }
 
+    pub fn suspend(&self, context: &Context) {
+        info!(context, "Suspending {}-thread.", self.name,);
+        {
+            self.state.0.lock().unwrap().suspended = true;
+        }
+        self.interrupt_idle(context);
+        loop {
+            let using_handle = self.state.0.lock().unwrap().using_handle;
+            if !using_handle {
+                return;
+            }
+            std::thread::sleep(std::time::Duration::from_micros(300 * 1000));
+        }
+    }
+
+    pub fn unsuspend(&self, context: &Context) {
+        info!(context, "Unsuspending {}-thread.", self.name);
+
+        let &(ref lock, ref cvar) = &*self.state.clone();
+        let mut state = lock.lock().unwrap();
+
+        state.suspended = false;
+        state.idle = true;
+        cvar.notify_one();
+    }
+
     pub fn interrupt_idle(&self, context: &Context) {
         {
             self.state.0.lock().unwrap().jobs_needed = true;
@@ -50,6 +77,10 @@ impl JobThread {
             let &(ref lock, _) = &*self.state.clone();
             let mut state = lock.lock().unwrap();
 
+            if state.suspended {
+                return;
+            }
+
             state.using_handle = true;
         }
 
@@ -85,7 +116,7 @@ impl JobThread {
         }
     }
 
-    fn get_watch_folder(&self, context: &Context) -> Option<String> {
+    pub(crate) 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 => {
@@ -114,6 +145,14 @@ impl JobThread {
                 return;
             }
 
+            if state.suspended {
+                while !state.idle {
+                    state = cvar.wait(state).unwrap();
+                }
+                state.idle = false;
+                return;
+            }
+
             state.using_handle = true;
 
             if !use_network {