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

If a pre-message is lost or delayed, or doesn't exist at all, download_msg() adds a new message
instead of replacing an existing one. For simplicity, move emitting IncomingMsgBunch to
Session::fetch_many_msgs() so that this event is emitted also in the mentioned case, it's not a
problem if it's emitted when a pre-message was downloaded before.

Co-authored-by: WofWca <wofwca@protonmail.com>

Cargo.lock

@@ -1375,6 +1375,7 @@ dependencies = [
  "regex",
  "rusqlite",
  "sanitize-filename",
+ "scopeguard",
  "sdp",
  "serde",
  "serde_json",

Cargo.toml

@@ -87,6 +87,7 @@ rand = { workspace = true }
 regex = { workspace = true }
 rusqlite = { workspace = true, features = ["sqlcipher"] }
 sanitize-filename = { workspace = true }
+scopeguard = "1"
 sdp = "0.17.1"
 serde_json = { workspace = true }
 serde_urlencoded = "0.7.1"

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::config::Config;
 use crate::context::Context;
 use crate::imap::session::Session;
@@ -131,9 +132,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(
@@ -209,12 +210,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,11 @@ impl Session {
         if request_uids.is_empty() {
             return Ok(());
         }
-
+        let mut received_any = scopeguard::guard(false, |v| {
+            if v {
+                context.emit_event(EventType::IncomingMsgBunch);
+            }
+        });
         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 +1399,7 @@ impl Session {
                     }
                     Ok(msg) => msg,
                 };
+                *received_any |= received_msg.is_some();
                 received_msgs_channel
                     .send((request_uid, received_msg))
                     .await?;

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,