fix: Emit IncomingMsgBunch on receipt of large message w/o pre-message (#8232)

src/download.rs

@@ -2,10 +2,11 @@
 
 use std::collections::BTreeMap;
 
-use anyhow::{Result, anyhow, bail, ensure};
+use anyhow::{Context as _, Result, anyhow, bail, ensure};
 use deltachat_derive::{FromSql, ToSql};
 use serde::{Deserialize, Serialize};
 
+use crate::chat;
 use crate::context::Context;
 use crate::imap::session::Session;
 use crate::log::warn;
@@ -130,9 +131,9 @@ impl Message {
     }
 }
 
-/// Actually downloads a message partially downloaded before if the message is available on the
-/// session transport, in which case returns `Some`. If the message is available on another
-/// transport, returns `None`.
+/// Actually downloads a message, normally partially downloaded before (if it's an encrypted chat
+/// message and the pre-message isn't lost), if the message is available on the session transport,
+/// in which case returns `Some`. If the message is available on another transport, returns `None`.
 ///
 /// Most messages are downloaded automatically on fetch instead.
 pub(crate) async fn download_msg(
@@ -207,12 +208,18 @@ impl Session {
         let (sender, receiver) = async_channel::unbounded();
         {
             let _fetch_msgs_lock_guard = context.fetch_msgs_mutex.lock().await;
-            self.fetch_many_msgs(context, folder, vec![uid], &uid_message_ids, sender)
+            Box::pin(self.fetch_many_msgs(context, folder, vec![uid], &uid_message_ids, sender))
                 .await?;
         }
-        if receiver.recv().await.is_err() {
-            bail!("Failed to fetch UID {uid}");
+        let mut received_msgs = Vec::with_capacity(1);
+        if let (_, Some(msg)) = receiver
+            .recv()
+            .await
+            .with_context(|| format!("Failed to fetch UID {uid}"))?
+        {
+            received_msgs.push(msg);
         }
+        chat::mark_old_messages_as_noticed(context, received_msgs).await?;
         Ok(())
     }
 }

src/imap.rs

@@ -754,8 +754,7 @@ impl Imap {
         };
 
         let actually_download_messages_future = async {
-            session
-                .fetch_many_msgs(context, folder, uids_fetch, &uid_message_ids, sender)
+            Box::pin(session.fetch_many_msgs(context, folder, uids_fetch, &uid_message_ids, sender))
                 .await
                 .context("fetch_many_msgs")
         };
@@ -785,11 +784,6 @@ impl Imap {
         }
 
         info!(context, "{} mails read from \"{}\".", read_cnt, folder);
-
-        if !received_msgs.is_empty() {
-            context.emit_event(EventType::IncomingMsgBunch);
-        }
-
         chat::mark_old_messages_as_noticed(context, received_msgs).await?;
 
         if fetch_res.is_ok() {
@@ -1288,7 +1282,7 @@ impl Session {
         if request_uids.is_empty() {
             return Ok(());
         }
-
+        let mut received_any = false;
         for (request_uids, set) in build_sequence_sets(&request_uids)? {
             info!(context, "Starting UID FETCH of message set \"{}\".", set);
             let mut fetch_responses = self
@@ -1401,6 +1395,7 @@ impl Session {
                     }
                     Ok(msg) => msg,
                 };
+                received_any |= received_msg.is_some();
                 received_msgs_channel
                     .send((request_uid, received_msg))
                     .await?;
@@ -1435,7 +1430,9 @@ impl Session {
                 );
             }
         }
-
+        if received_any {
+            context.emit_event(EventType::IncomingMsgBunch);
+        }
         Ok(())
     }
 

src/receive_imf.rs

@@ -462,6 +462,10 @@ async fn get_to_and_past_contact_ids(
 /// downloaded again, sets `chat_id=DC_CHAT_ID_TRASH` and returns `Ok(Some(…))`.
 /// If the message is so wrong that we didn't even create a database entry,
 /// returns `Ok(None)`.
+///
+/// The caller must emit [`EventType::IncomingMsgBunch`] if this function returned `Ok(Some)`,
+/// because [`EventType::IncomingMsg`] emitted by this function, if any, must be followed by
+/// [`EventType::IncomingMsgBunch`].
 pub(crate) async fn receive_imf_inner(
     context: &Context,
     rfc724_mid: &str,