diff --git a/crates/bridge/src/bridge.rs b/crates/bridge/src/bridge.rs index 828816a..a9b2b57 100644 --- a/crates/bridge/src/bridge.rs +++ b/crates/bridge/src/bridge.rs @@ -367,7 +367,6 @@ impl BridgeHandle { store.clone(), local_store, backend.clone(), - sync_limit, bus_ids_for_handler, event_rx, shutdown_sync_rx.clone(), diff --git a/crates/bridge/src/event_handler.rs b/crates/bridge/src/event_handler.rs index d22e565..4c76202 100644 --- a/crates/bridge/src/event_handler.rs +++ b/crates/bridge/src/event_handler.rs @@ -44,7 +44,6 @@ pub async fn run_event_handler( store: Arc, local_store: Arc, backend: Arc, - sync_limit: usize, last_batch_ids: Arc>>, mut rx: mpsc::Receiver, mut shutdown: watch::Receiver, @@ -56,7 +55,7 @@ pub async fn run_event_handler( _ = shutdown.changed() => break, msg = rx.recv() => { let Some(msg) = msg else { break }; - process(&store, &local_store, &*backend, sync_limit, &last_batch_ids, msg).await; + process(&store, &local_store, &*backend, &last_batch_ids, msg).await; } } } @@ -67,7 +66,6 @@ async fn process( store: &MailStore, local_store: &LocalStore, backend: &dyn MailBackend, - sync_limit: usize, last_batch_ids: &Mutex>, msg: EventBusMessage, ) { @@ -82,7 +80,7 @@ async fn process( _ => return, }; - apply_batch(store, local_store, backend, sync_limit, &batch).await; + apply_batch(store, local_store, backend, &batch).await; // Advance the in-memory catch-up state and persist it. The two must stay // in sync — the in-memory map drives the next reconnect's query string, @@ -165,7 +163,6 @@ async fn apply_batch( store: &MailStore, local_store: &LocalStore, backend: &dyn MailBackend, - sync_limit: usize, batch: &EntityUpdateBatch, ) { let Bucketed { @@ -244,7 +241,7 @@ async fn apply_batch( "Event bus: fallback full sync for {} (batch {})", folder.imap_path, batch.batch_id ); - if let Err(e) = sync_folder(store, local_store, backend, folder, sync_limit).await { + if let Err(e) = sync_folder(store, local_store, backend, folder).await { warn!( "Event bus fallback sync failed for {}: {}", folder.imap_path, e diff --git a/crates/bridge/src/imap/session.rs b/crates/bridge/src/imap/session.rs index a1166aa..c04650e 100644 --- a/crates/bridge/src/imap/session.rs +++ b/crates/bridge/src/imap/session.rs @@ -1,6 +1,6 @@ use log::{debug, info}; use std::sync::Arc; -use tutasdk::entities::generated::tutanota::{Mail, MailDetails}; +use tutasdk::entities::generated::tutanota::{Mail, MailDetails, TutanotaFile}; use crate::mail::mail_to_rfc2822; use crate::mail::rfc2822::{extract_headers, format_internal_date}; @@ -19,6 +19,12 @@ struct CachedMail { mail: Mail, details: Option, rfc2822: Option, + /// `rfc2822` always carries the real headers, but for messages outside the + /// body-prefetch window it holds a placeholder body. `body_loaded` tracks + /// whether the *body* is final: `true` once we have real details (or have + /// confirmed the message has no body source), `false` while a body FETCH + /// still needs to pull it on demand. + body_loaded: bool, uid: u32, deleted: bool, } @@ -368,44 +374,77 @@ impl ImapSession { continue; } - if self.mails[idx].details.is_none() && needs_body(&items) { + // Make sure we have a renderable body before answering a body + // FETCH. The mailbox lists every message, but only the newest + // `sync_limit` bodies are pre-warmed — older ones are fetched + // here, on demand, the first time a client opens them. + if needs_body(&items) && !self.mails[idx].body_loaded { let elem_id = self.mails[idx] .mail ._id .as_ref() .map(|id| id.element_id.to_string()); - // Check store — syncer may have loaded details since our snapshot - let from_store = match (&elem_id, &self.selected_folder) { - (Some(eid), Some(folder)) => self.store.get_details(&folder.id, eid).await, - _ => None, - }; - if let Some((details, rfc)) = from_store { - self.mails[idx].details = Some(details); - self.mails[idx].rfc2822 = Some(rfc); - } else { - debug!("Details not yet synced for uid={}", self.mails[idx].uid); - } - } - - if needs_body(&items) { - if self.mails[idx].details.is_some() && self.mails[idx].rfc2822.is_none() { + if self.mails[idx].details.is_some() { + // Have details already, just render the envelope. let rfc = mail_to_rfc2822( &self.mails[idx].mail, self.mails[idx].details.as_ref(), &[], ); self.mails[idx].rfc2822 = Some(rfc); - } else if self.mails[idx].rfc2822.is_none() { - log::warn!( - "No body for uid={}, will serve headers-only placeholder", - self.mails[idx].uid, - ); + self.mails[idx].body_loaded = true; + } else if let Some((details, rfc)) = match (&elem_id, &self.selected_folder) { + // The syncer may have filled it in since our snapshot. + (Some(eid), Some(folder)) => self.store.get_details(&folder.id, eid).await, + _ => None, + } { + self.mails[idx].details = Some(details); + self.mails[idx].rfc2822 = Some(rfc); + self.mails[idx].body_loaded = true; + } else { + // Out of the prefetch window — fetch the body on demand. + let mail = self.mails[idx].mail.clone(); + match self.backend.load_mail_details(&mail).await { + Ok(Some(details)) => { + let atts = self + .backend + .load_attachments(&mail) + .await + .unwrap_or_default(); + let refs: Vec<(&TutanotaFile, &[u8])> = + atts.iter().map(|(f, d)| (f, d.as_slice())).collect(); + let rfc = mail_to_rfc2822(&mail, Some(&details), &refs); + // Share into the in-memory store so other + // connections (and stats) see it too. + if let (Some(eid), Some(folder)) = (&elem_id, &self.selected_folder) { + self.store + .update_mail_details( + &folder.id, + eid, + details.clone(), + rfc.clone(), + false, + ) + .await; + } + self.mails[idx].details = Some(details); + self.mails[idx].rfc2822 = Some(rfc); + self.mails[idx].body_loaded = true; + } + Ok(None) => { + // Legacy mail with no body reference — headers only. + // Mark loaded so we don't re-hit the network on + // every body FETCH of a message that has no body. + 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 + ), + } } - // If `rfc2822` is already populated (Phase 0 read it from - // disk), there is nothing else to do — the body will be - // served from it; `details` being None is normal and not a - // placeholder situation. } let cached = &self.mails[idx]; @@ -661,6 +700,12 @@ impl ImapSession { } let details = sm.details.or(old_details); + // The body is final if we have real details, or a previously + // rendered rfc2822 (from the store or this session) — those always + // carry a real body. With none of those, the rfc2822 we render + // below has a placeholder body, so the body still needs an + // on-demand fetch the first time a client opens the message. + let body_loaded = details.is_some() || sm.rfc2822.is_some() || old_rfc.is_some(); let rfc2822 = sm .rfc2822 .or(old_rfc) @@ -670,6 +715,7 @@ impl ImapSession { mail: sm.mail, details, rfc2822: Some(rfc2822), + body_loaded, uid, deleted: false, }); @@ -994,6 +1040,7 @@ mod tests { }, details: None, rfc2822: Some("Date: Wed, 25 Dec 2024 12:30:45 +0000\r\nFrom: sender@tuta.com\r\nSubject: Test\r\n\r\nBody\r\n".to_string()), + body_loaded: true, uid, deleted: false, } diff --git a/crates/bridge/src/store.rs b/crates/bridge/src/store.rs index 8e395e7..b2111b9 100644 --- a/crates/bridge/src/store.rs +++ b/crates/bridge/src/store.rs @@ -137,6 +137,27 @@ impl LocalStore { .is_ok() } + /// Read a value from the generic `store_meta` key-value table. + pub fn get_meta(&self, key: &str) -> Option { + let conn = self.conn.lock().unwrap(); + conn.query_row( + "SELECT value FROM store_meta WHERE key = ?1", + [key], + |row| row.get::<_, String>(0), + ) + .ok() + } + + /// Write a value into the generic `store_meta` key-value table. + pub fn set_meta(&self, key: &str, value: &str) -> Result<(), StoreError> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "INSERT OR REPLACE INTO store_meta(key, value) VALUES (?1, ?2)", + [key, value], + )?; + Ok(()) + } + pub fn reset(&self) -> Result<(), StoreError> { warn!("Resetting local store — all cached data will be deleted"); let conn = self.conn.lock().unwrap(); diff --git a/crates/bridge/src/sync.rs b/crates/bridge/src/sync.rs index d72d55b..4b7698f 100644 --- a/crates/bridge/src/sync.rs +++ b/crates/bridge/src/sync.rs @@ -281,7 +281,7 @@ impl MailStore { } } - async fn update_mail_details( + pub(crate) async fn update_mail_details( &self, folder_id: &str, element_id: &str, @@ -349,7 +349,7 @@ pub async fn run_syncer( shutdown: watch::Receiver, ) { info!( - "Mail syncer started (limit={})", + "Mail syncer started (body prefetch depth={})", if sync_limit == 0 { "all".to_string() } else { @@ -378,10 +378,15 @@ pub async fn run_syncer( } } - // Bootstrap: if we have no cached event-bus catch-up state, the on-disk - // cache may be stale or empty. Run a one-shot full list sync of every - // folder so the store reflects current server state; from then on the - // event bus drives all updates (no periodic polling). + // We run a one-shot full **metadata** sync (the whole mail list of every + // folder) when either: + // * there's no cached event-bus state (fresh install / cache wiped), or + // * we've never completed a full-metadata sync before — the marker + // below. This second case migrates existing installs that were synced + // under the old `sync_limit`-truncated model up to a complete mailbox. + // After it runs once, the event bus keeps the list current; we don't + // re-list every launch (the metadata is persisted + loaded in Phase 0). + const FULL_METADATA_MARKER: &str = "full_metadata_synced_v1"; let needs_bootstrap = match local_store.load_event_bus_state() { Ok(s) => s.is_empty(), Err(e) => { @@ -389,24 +394,49 @@ pub async fn run_syncer( true } }; - if needs_bootstrap && !folders.is_empty() { - info!("Bootstrap sync (no cached event-bus state)"); + let full_metadata_done = local_store.get_meta(FULL_METADATA_MARKER).is_some(); + let do_full_sync = (needs_bootstrap || !full_metadata_done) && !folders.is_empty(); + + if do_full_sync { + if needs_bootstrap { + info!("Bootstrap sync (no cached event-bus state)"); + } else { + info!("One-time full-metadata sync (completing the mailbox view)"); + } + let mut all_ok = true; for folder in &folders { if *shutdown.borrow() { return; } - if let Err(e) = sync_folder(&store, &local_store, &*backend, folder, sync_limit).await { - warn!("Bootstrap sync failed for {}: {}", folder.imap_path, e); + if let Err(e) = sync_folder(&store, &local_store, &*backend, folder).await { + warn!("Full-metadata sync failed for {}: {}", folder.imap_path, e); + all_ok = false; } tokio::time::sleep(INTER_FOLDER_DELAY).await; } - } else if !needs_bootstrap { - debug!("Skipping bootstrap sync — event-bus catch-up will reconcile"); + // Only set the marker if every folder synced, so a transient failure + // gets retried on the next launch rather than silently leaving the + // mailbox incomplete. + if all_ok { + if let Err(e) = local_store.set_meta(FULL_METADATA_MARKER, "1") { + warn!("Could not persist full-metadata marker: {e}"); + } + } + } else { + debug!("Skipping full-metadata sync — already complete, event bus reconciles"); } - // From here on the syncer only owns the slow body prefetch. Folder / - // new-mail refresh is driven by the event bus + `event_handler`. - prefetch_loop(&store, &local_store, &*backend, shutdown.clone()).await; + // From here on the syncer only owns the slow body prefetch (capped at + // `sync_limit` — the body-prefetch depth). Folder / new-mail refresh is + // driven by the event bus + `event_handler`. + prefetch_loop( + &store, + &local_store, + &*backend, + sync_limit, + shutdown.clone(), + ) + .await; info!("Mail syncer shutting down"); } @@ -420,6 +450,7 @@ async fn prefetch_loop( store: &MailStore, local_store: &LocalStore, backend: &dyn MailBackend, + body_prefetch_limit: usize, mut shutdown: watch::Receiver, ) { let mut store_watch = store.subscribe(); @@ -442,6 +473,7 @@ async fn prefetch_loop( local_store, backend, &folder, + body_prefetch_limit, &mut attachment_retry_history, ) .await; @@ -509,10 +541,14 @@ async fn load_cached_folder( } Some(eml) } - Ok(None) => Some(mail_to_rfc2822(&mail, None, &[])), + // No cached body — leave `rfc2822` empty rather than baking in a + // placeholder. A placeholder would look like a real body to the + // IMAP layer and suppress the on-demand fetch; `None` is what tells + // it the body still has to be pulled the first time it's opened. + Ok(None) => None, Err(e) => { warn!("Failed to read cached eml {}: {e}", meta.element_id); - Some(mail_to_rfc2822(&mail, None, &[])) + None } }; @@ -530,14 +566,19 @@ async fn load_cached_folder( Ok(count) } +/// Sync a folder's full mail-list **metadata** (subject / sender / date / +/// flags) — every message, not a `sync_limit`-capped slice. This is what +/// makes the whole mailbox visible over IMAP; message *bodies* are fetched +/// separately and lazily (see [`prefetch_details`] + on-demand FETCH). pub(crate) async fn sync_folder( store: &MailStore, local_store: &LocalStore, backend: &dyn MailBackend, folder: &FolderInfo, - limit: usize, ) -> Result<(), String> { - let new_mails = retry(|| backend.load_mail_ids_for_folder(folder, limit)).await?; + // `0` = load every entry (the SDK paginates). Metadata is cheap relative + // to bodies (no blob fetch), so we never truncate the list. + let new_mails = retry(|| backend.load_mail_ids_for_folder(folder, 0)).await?; let existing = store.get_folder(&folder.id).await; let existing_map: HashMap = existing @@ -592,11 +633,12 @@ pub(crate) async fn sync_folder( attachments_pending: existing.attachments_pending, }); } else { - let rfc2822 = mail_to_rfc2822(mail, None, &[]); + // New mail, body not fetched yet — leave `rfc2822` empty so the + // IMAP layer fetches it on demand (a placeholder would mask that). updated.push(StoredMail { mail: mail.clone(), details: None, - rfc2822: Some(rfc2822), + rfc2822: None, uid, attachments_pending: false, }); @@ -644,12 +686,24 @@ async fn prefetch_details( local_store: &LocalStore, backend: &dyn MailBackend, folder: &FolderInfo, + body_prefetch_limit: usize, attachment_retry_history: &mut HashMap, ) { let mails = store.get_folder(&folder.id).await; - // First pass — fresh mails still missing their `.eml` on disk. - let api_needed: Vec = mails + // Body prefetch is the *capped* part: eagerly pull bodies only for the + // newest `body_prefetch_limit` messages (the store is newest-first). + // `0` = no cap (prefetch every body). Bodies beyond the window are fetched + // on demand when the client FETCHes them. (Metadata for the whole folder + // is always present — see `sync_folder`.) + let prefetch_window: &[StoredMail] = if body_prefetch_limit == 0 { + &mails + } else { + &mails[..mails.len().min(body_prefetch_limit)] + }; + + // First pass — messages in the prefetch window still missing their `.eml`. + let api_needed: Vec = prefetch_window .iter() .filter(|m| m.details.is_none()) .filter_map(|m| { diff --git a/src/main.rs b/src/main.rs index cff4ff3..9d58646 100644 --- a/src/main.rs +++ b/src/main.rs @@ -192,7 +192,6 @@ async fn main() -> anyhow::Result<()> { store.clone(), local_store.clone(), backend.clone(), - cfg.sync_limit, bus_ids_for_handler, event_rx, shutdown_rx.clone(), diff --git a/ui/src/components/ConfigPanel.tsx b/ui/src/components/ConfigPanel.tsx index f25f145..bc25359 100644 --- a/ui/src/components/ConfigPanel.tsx +++ b/ui/src/components/ConfigPanel.tsx @@ -82,18 +82,23 @@ export function ConfigPanel({ config, status, loading, onSave, onRestart }: Prop />
- + + + Your whole mailbox is always listed and searchable by subject, sender + and date. This only sets how many recent message bodies are + kept ready offline — older ones load on demand when you open them. + {fetchAll ? ( - Downloads every mail from the start — can be slow on large accounts. + Downloads every body — slow + uses the most disk on large accounts. ) : ( setSyncLimit(Math.max(1, Number(e.target.value)))} - placeholder="Max mails per folder" + placeholder="Bodies to keep offline (most recent)" /> )}