feat(imap): on-demand body fetch correctness and anti-storm

needs_body() now triggers an on-demand fetch only for items that actually
need the body (BODY[] / BODY.PEEK[] / standalone RFC822). It used to fire for
ENVELOPE and every BODY[...] section, including BODY[HEADER...], so the
client's list-building did one full body+attachment download per message: on
a 19535-mail inbox that meant downloading the whole mailbox just to render
the list (the request storm, and an empty list while it ground on). Envelope,
header, size and structure items are answered from local metadata.

On a genuine fetch failure the bridge now returns a tagged NO [UNAVAILABLE]
instead of a successful response carrying a placeholder body (which made the
client cache a fake message and re-request forever). The failed mail is put
on a 30s cooldown, shared across IMAP connections via MailStore, so the
bridge does not re-hit a throttled server on every client redraw.

Tests: needs_body (skip metadata/header, trigger only on real body),
body-fetch cooldown set/expire.

crates/bridge/src/imap/session.rs

@@ -423,7 +423,17 @@ impl ImapSession {
                     self.mails[idx].rfc2822 = Some(rfc);
                     self.mails[idx].body_loaded = true;
                 } else {
-                    // Out of the prefetch window — fetch the body on demand.
+                    // Out of the prefetch window: fetch the body on demand,
+                    // unless this mail is in a post-failure cooldown (do not
+                    // re-hit a throttled server on every client redraw).
+                    if let Some(eid) = &elem_id {
+                        if self.store.body_fetch_on_cooldown(eid) {
+                            return vec![format!(
+                                "{} NO [UNAVAILABLE] message body temporarily unavailable, try again later\r\n",
+                                tag
+                            )];
+                        }
+                    }
                     let mail = self.mails[idx].mail.clone();
                     match self.backend.load_mail_details(&mail).await {
                         Ok(Some(details)) => {
@@ -459,10 +469,19 @@ impl ImapSession {
                             self.mails[idx].body_loaded = true;
                             debug!("uid={} has no body source", self.mails[idx].uid);
                         }
-                        Err(e) => log::warn!(
-                            "On-demand body fetch failed for uid={}: {e}",
-                            self.mails[idx].uid
-                        ),
+                        Err(e) => {
+                            log::warn!(
+                                "On-demand body fetch failed for uid={}: {e}",
+                                self.mails[idx].uid
+                            );
+                            if let Some(eid) = &elem_id {
+                                self.store.mark_body_fetch_failed(eid);
+                            }
+                            return vec![format!(
+                                "{} NO [UNAVAILABLE] could not fetch message body, try again later\r\n",
+                                tag
+                            )];
+                        }
                     }
                 }
             }
@@ -1038,19 +1057,31 @@ fn parse_store_args(args: &str) -> (String, String) {
     (seq, rest)
 }
 
+/// Whether a FETCH item list requires the real message body to be loaded.
+///
+/// Only full-body content does: `BODY[]` / `BODY.PEEK[]`, a numbered MIME part
+/// (`BODY[1]`), `BODY[TEXT]`, or a standalone `RFC822` / `RFC822.TEXT`. Items the
+/// bridge answers from metadata alone must NOT trigger an on-demand fetch:
+/// `ENVELOPE`, `BODY[HEADER...]`, `BODYSTRUCTURE`, `RFC822.SIZE`, `FLAGS`, etc.
+/// Treating header/envelope requests as body requests made the client's
+/// list-building (one such fetch per message) download the entire mailbox.
+/// Whether a FETCH item list requires the real message body to be loaded.
+///
+/// The bridge serves the full message for `BODY[]` / `BODY.PEEK[]` and a
+/// standalone `RFC822`; only those need the body. Envelope, header, size and
+/// structure items (`ENVELOPE`, `BODY[HEADER...]`, `RFC822.SIZE`,
+/// `BODYSTRUCTURE`, `FLAGS`, ...) are answered from metadata and must NOT
+/// trigger an on-demand fetch. Treating header/envelope requests as body
+/// requests made the client's list-building (one such fetch per message)
+/// download the entire mailbox.
 fn needs_body(items: &str) -> bool {
     let u = items.to_uppercase();
-    if u.contains("BODY[") || u.contains("BODY.PEEK[") || u.contains("ENVELOPE") {
+    if u.contains("BODY[]") || u.contains("BODY.PEEK[]") {
         return true;
     }
-    // Match "RFC822" as a standalone fetch item but not "RFC822.SIZE" or "RFC822.HEADER"
-    for token in u.split_whitespace() {
-        let token = token.trim_matches(|c| c == '(' || c == ')');
-        if token == "RFC822" {
-            return true;
-        }
-    }
-    false
+    // Standalone RFC822 (a full message), but not RFC822.SIZE / .HEADER / .TEXT.
+    u.split(|c: char| c == '(' || c == ')' || c.is_whitespace())
+        .any(|token| token == "RFC822")
 }
 
 fn parse_login_args(args: &str) -> (String, String) {
@@ -1360,7 +1391,7 @@ mod tests {
         assert!(needs_body("(BODY[])"));
         assert!(needs_body("(BODY.PEEK[])"));
         assert!(needs_body("(RFC822)"));
-        assert!(needs_body("(ENVELOPE)"));
+        assert!(!needs_body("(ENVELOPE)"));
         assert!(needs_body("(FLAGS BODY[])"));
         assert!(!needs_body("(FLAGS)"));
         assert!(!needs_body("(FLAGS UID INTERNALDATE)"));
@@ -1567,6 +1598,31 @@ mod tests {
         assert!(!resp.contains("\"\"Important\"\""));
     }
 
+    // --- needs_body: only real body content triggers an on-demand fetch ---
+
+    #[test]
+    fn needs_body_skips_metadata_and_header_fetches() {
+        // List-building items are answered from metadata, no fetch.
+        assert!(!needs_body("(FLAGS UID)"));
+        assert!(!needs_body("(ENVELOPE)"));
+        assert!(!needs_body("(FLAGS UID RFC822.SIZE ENVELOPE)"));
+        assert!(!needs_body(
+            "(BODY.PEEK[HEADER.FIELDS (FROM TO SUBJECT DATE)])"
+        ));
+        assert!(!needs_body("(UID RFC822.SIZE BODYSTRUCTURE FLAGS)"));
+        assert!(!needs_body("(RFC822.HEADER)"));
+        assert!(!needs_body("(BODY.PEEK[1.MIME])"));
+    }
+
+    #[test]
+    fn needs_body_triggers_for_body_content() {
+        assert!(needs_body("(BODY[])"));
+        assert!(needs_body("(BODY.PEEK[])"));
+        assert!(needs_body("(RFC822)"));
+        assert!(needs_body("(FLAGS BODY.PEEK[])"));
+        assert!(needs_body("BODY[]"));
+    }
+
     // =================================================================
     // Integration tests with MockBackend
     // =================================================================

crates/bridge/src/sync.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
-use std::sync::Arc;
-use std::time::Duration;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 
 use log::{debug, info, warn};
 use tokio::sync::{watch, RwLock};
@@ -13,6 +13,10 @@ use crate::tuta::{FolderInfo, MailBackend};
 const INTER_REQUEST_DELAY: Duration = Duration::from_millis(150);
 const INTER_FOLDER_DELAY: Duration = Duration::from_millis(300);
 const MAX_RETRIES: u32 = 3;
+/// How long a mail whose on-demand body fetch just failed is skipped before the
+/// bridge will try the API again for it. Stops a client from re-triggering a
+/// fetch for the same message on every redraw while the server is throttling.
+const BODY_FETCH_COOLDOWN: Duration = Duration::from_secs(30);
 
 #[derive(Clone)]
 pub struct StoredMail {
@@ -36,6 +40,9 @@ pub struct MailStore {
     folder_list: RwLock<Vec<FolderInfo>>,
     generation: watch::Sender<u64>,
     gen_counter: std::sync::atomic::AtomicU64,
+    /// element_id -> instant until which an on-demand body fetch is skipped,
+    /// armed after a fetch failure. Shared across IMAP connections.
+    body_fetch_cooldown: Mutex<HashMap<String, Instant>>,
 }
 
 impl MailStore {
@@ -46,9 +53,31 @@ impl MailStore {
             folder_list: RwLock::new(Vec::new()),
             generation: tx,
             gen_counter: std::sync::atomic::AtomicU64::new(0),
+            body_fetch_cooldown: Mutex::new(HashMap::new()),
         })
     }
 
+    /// True if a recent on-demand body fetch for `element_id` failed and its
+    /// cooldown has not yet elapsed. Expired entries are dropped lazily.
+    pub(crate) fn body_fetch_on_cooldown(&self, element_id: &str) -> bool {
+        let mut map = crate::util::lock_recover(&self.body_fetch_cooldown);
+        match map.get(element_id) {
+            Some(&until) if until > Instant::now() => true,
+            Some(_) => {
+                map.remove(element_id);
+                false
+            }
+            None => false,
+        }
+    }
+
+    /// Arm a cooldown after a failed on-demand body fetch so the bridge does not
+    /// immediately re-hit a throttled or erroring server for the same mail.
+    pub(crate) fn mark_body_fetch_failed(&self, element_id: &str) {
+        crate::util::lock_recover(&self.body_fetch_cooldown)
+            .insert(element_id.to_string(), Instant::now() + BODY_FETCH_COOLDOWN);
+    }
+
     pub fn subscribe(&self) -> watch::Receiver<u64> {
         self.generation.subscribe()
     }
@@ -1127,6 +1156,22 @@ mod tests {
         }
     }
 
+    #[test]
+    fn body_fetch_cooldown_blocks_then_expires() {
+        let store = MailStore::new();
+        assert!(!store.body_fetch_on_cooldown("mail-a"));
+        store.mark_body_fetch_failed("mail-a");
+        assert!(store.body_fetch_on_cooldown("mail-a"));
+        // A different mail is unaffected.
+        assert!(!store.body_fetch_on_cooldown("mail-b"));
+        // An elapsed entry is treated as expired and dropped lazily.
+        crate::util::lock_recover(&store.body_fetch_cooldown).insert(
+            "mail-a".to_string(),
+            Instant::now() - Duration::from_secs(1),
+        );
+        assert!(!store.body_fetch_on_cooldown("mail-a"));
+    }
+
     #[tokio::test]
     async fn refresh_mail_in_place_updates_metadata_in_every_folder() {
         let store = MailStore::new();