Show the complete mailbox; sync_limit now caps body prefetch only

The local store was capped at sync_limit, so IMAP only ever listed the
newest N messages — a search in Thunderbird (the only search UI we have)
silently missed everything older. Now the syncer lists the *full* mailbox
metadata for every folder, and sync_limit governs only how many recent
message bodies are pre-warmed offline. Bodies outside that window are
fetched on demand the first time a client opens the message.

A one-time full-metadata sync (marker full_metadata_synced_v1) completes
the mailbox view on first launch after upgrade.

Crucially, an empty body is now stored as rfc2822 = None rather than a
rendered "(No body available)" placeholder: the placeholder looked like a
real body to the IMAP layer and suppressed the on-demand fetch. CachedMail
gains body_loaded to track whether the body (not just the headers) is final.

Validated live on a 19,322-message INBOX (26,965 mails total across
folders): full listing, on-demand body fetch (~0.1-0.4s), in-memory cache
on re-fetch.
This commit is contained in:
Anthony
2026-05-29 18:45:08 +02:00
parent c401349d24
commit 472eb7880e
7 changed files with 183 additions and 61 deletions
-1
View File
@@ -367,7 +367,6 @@ impl BridgeHandle {
store.clone(), store.clone(),
local_store, local_store,
backend.clone(), backend.clone(),
sync_limit,
bus_ids_for_handler, bus_ids_for_handler,
event_rx, event_rx,
shutdown_sync_rx.clone(), shutdown_sync_rx.clone(),
+3 -6
View File
@@ -44,7 +44,6 @@ pub async fn run_event_handler(
store: Arc<MailStore>, store: Arc<MailStore>,
local_store: Arc<LocalStore>, local_store: Arc<LocalStore>,
backend: Arc<dyn MailBackend>, backend: Arc<dyn MailBackend>,
sync_limit: usize,
last_batch_ids: Arc<Mutex<HashMap<String, String>>>, last_batch_ids: Arc<Mutex<HashMap<String, String>>>,
mut rx: mpsc::Receiver<EventBusMessage>, mut rx: mpsc::Receiver<EventBusMessage>,
mut shutdown: watch::Receiver<bool>, mut shutdown: watch::Receiver<bool>,
@@ -56,7 +55,7 @@ pub async fn run_event_handler(
_ = shutdown.changed() => break, _ = shutdown.changed() => break,
msg = rx.recv() => { msg = rx.recv() => {
let Some(msg) = msg else { break }; 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, store: &MailStore,
local_store: &LocalStore, local_store: &LocalStore,
backend: &dyn MailBackend, backend: &dyn MailBackend,
sync_limit: usize,
last_batch_ids: &Mutex<HashMap<String, String>>, last_batch_ids: &Mutex<HashMap<String, String>>,
msg: EventBusMessage, msg: EventBusMessage,
) { ) {
@@ -82,7 +80,7 @@ async fn process(
_ => return, _ => 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 // 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, // in sync — the in-memory map drives the next reconnect's query string,
@@ -165,7 +163,6 @@ async fn apply_batch(
store: &MailStore, store: &MailStore,
local_store: &LocalStore, local_store: &LocalStore,
backend: &dyn MailBackend, backend: &dyn MailBackend,
sync_limit: usize,
batch: &EntityUpdateBatch, batch: &EntityUpdateBatch,
) { ) {
let Bucketed { let Bucketed {
@@ -244,7 +241,7 @@ async fn apply_batch(
"Event bus: fallback full sync for {} (batch {})", "Event bus: fallback full sync for {} (batch {})",
folder.imap_path, batch.batch_id 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!( warn!(
"Event bus fallback sync failed for {}: {}", "Event bus fallback sync failed for {}: {}",
folder.imap_path, e folder.imap_path, e
+73 -26
View File
@@ -1,6 +1,6 @@
use log::{debug, info}; use log::{debug, info};
use std::sync::Arc; 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::mail_to_rfc2822;
use crate::mail::rfc2822::{extract_headers, format_internal_date}; use crate::mail::rfc2822::{extract_headers, format_internal_date};
@@ -19,6 +19,12 @@ struct CachedMail {
mail: Mail, mail: Mail,
details: Option<MailDetails>, details: Option<MailDetails>,
rfc2822: Option<String>, rfc2822: Option<String>,
/// `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, uid: u32,
deleted: bool, deleted: bool,
} }
@@ -368,44 +374,77 @@ impl ImapSession {
continue; 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] let elem_id = self.mails[idx]
.mail .mail
._id ._id
.as_ref() .as_ref()
.map(|id| id.element_id.to_string()); .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 { if self.mails[idx].details.is_some() {
self.mails[idx].details = Some(details); // Have details already, just render the envelope.
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() {
let rfc = mail_to_rfc2822( let rfc = mail_to_rfc2822(
&self.mails[idx].mail, &self.mails[idx].mail,
self.mails[idx].details.as_ref(), self.mails[idx].details.as_ref(),
&[], &[],
); );
self.mails[idx].rfc2822 = Some(rfc); self.mails[idx].rfc2822 = Some(rfc);
} else if self.mails[idx].rfc2822.is_none() { self.mails[idx].body_loaded = true;
log::warn!( } else if let Some((details, rfc)) = match (&elem_id, &self.selected_folder) {
"No body for uid={}, will serve headers-only placeholder", // The syncer may have filled it in since our snapshot.
self.mails[idx].uid, (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]; let cached = &self.mails[idx];
@@ -661,6 +700,12 @@ impl ImapSession {
} }
let details = sm.details.or(old_details); 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 let rfc2822 = sm
.rfc2822 .rfc2822
.or(old_rfc) .or(old_rfc)
@@ -670,6 +715,7 @@ impl ImapSession {
mail: sm.mail, mail: sm.mail,
details, details,
rfc2822: Some(rfc2822), rfc2822: Some(rfc2822),
body_loaded,
uid, uid,
deleted: false, deleted: false,
}); });
@@ -994,6 +1040,7 @@ mod tests {
}, },
details: None, 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()), 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, uid,
deleted: false, deleted: false,
} }
+21
View File
@@ -137,6 +137,27 @@ impl LocalStore {
.is_ok() .is_ok()
} }
/// Read a value from the generic `store_meta` key-value table.
pub fn get_meta(&self, key: &str) -> Option<String> {
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> { pub fn reset(&self) -> Result<(), StoreError> {
warn!("Resetting local store — all cached data will be deleted"); warn!("Resetting local store — all cached data will be deleted");
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
+77 -23
View File
@@ -281,7 +281,7 @@ impl MailStore {
} }
} }
async fn update_mail_details( pub(crate) async fn update_mail_details(
&self, &self,
folder_id: &str, folder_id: &str,
element_id: &str, element_id: &str,
@@ -349,7 +349,7 @@ pub async fn run_syncer(
shutdown: watch::Receiver<bool>, shutdown: watch::Receiver<bool>,
) { ) {
info!( info!(
"Mail syncer started (limit={})", "Mail syncer started (body prefetch depth={})",
if sync_limit == 0 { if sync_limit == 0 {
"all".to_string() "all".to_string()
} else { } else {
@@ -378,10 +378,15 @@ pub async fn run_syncer(
} }
} }
// Bootstrap: if we have no cached event-bus catch-up state, the on-disk // We run a one-shot full **metadata** sync (the whole mail list of every
// cache may be stale or empty. Run a one-shot full list sync of every // folder) when either:
// folder so the store reflects current server state; from then on the // * there's no cached event-bus state (fresh install / cache wiped), or
// event bus drives all updates (no periodic polling). // * 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() { let needs_bootstrap = match local_store.load_event_bus_state() {
Ok(s) => s.is_empty(), Ok(s) => s.is_empty(),
Err(e) => { Err(e) => {
@@ -389,24 +394,49 @@ pub async fn run_syncer(
true true
} }
}; };
if needs_bootstrap && !folders.is_empty() { let full_metadata_done = local_store.get_meta(FULL_METADATA_MARKER).is_some();
info!("Bootstrap sync (no cached event-bus state)"); 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 { for folder in &folders {
if *shutdown.borrow() { if *shutdown.borrow() {
return; return;
} }
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!("Bootstrap sync failed for {}: {}", folder.imap_path, e); warn!("Full-metadata sync failed for {}: {}", folder.imap_path, e);
all_ok = false;
} }
tokio::time::sleep(INTER_FOLDER_DELAY).await; tokio::time::sleep(INTER_FOLDER_DELAY).await;
} }
} else if !needs_bootstrap { // Only set the marker if every folder synced, so a transient failure
debug!("Skipping bootstrap sync — event-bus catch-up will reconcile"); // 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 / // From here on the syncer only owns the slow body prefetch (capped at
// new-mail refresh is driven by the event bus + `event_handler`. // `sync_limit` — the body-prefetch depth). Folder / new-mail refresh is
prefetch_loop(&store, &local_store, &*backend, shutdown.clone()).await; // driven by the event bus + `event_handler`.
prefetch_loop(
&store,
&local_store,
&*backend,
sync_limit,
shutdown.clone(),
)
.await;
info!("Mail syncer shutting down"); info!("Mail syncer shutting down");
} }
@@ -420,6 +450,7 @@ async fn prefetch_loop(
store: &MailStore, store: &MailStore,
local_store: &LocalStore, local_store: &LocalStore,
backend: &dyn MailBackend, backend: &dyn MailBackend,
body_prefetch_limit: usize,
mut shutdown: watch::Receiver<bool>, mut shutdown: watch::Receiver<bool>,
) { ) {
let mut store_watch = store.subscribe(); let mut store_watch = store.subscribe();
@@ -442,6 +473,7 @@ async fn prefetch_loop(
local_store, local_store,
backend, backend,
&folder, &folder,
body_prefetch_limit,
&mut attachment_retry_history, &mut attachment_retry_history,
) )
.await; .await;
@@ -509,10 +541,14 @@ async fn load_cached_folder(
} }
Some(eml) 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) => { Err(e) => {
warn!("Failed to read cached eml {}: {e}", meta.element_id); 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) 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( pub(crate) async fn sync_folder(
store: &MailStore, store: &MailStore,
local_store: &LocalStore, local_store: &LocalStore,
backend: &dyn MailBackend, backend: &dyn MailBackend,
folder: &FolderInfo, folder: &FolderInfo,
limit: usize,
) -> Result<(), String> { ) -> 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 = store.get_folder(&folder.id).await;
let existing_map: HashMap<String, StoredMail> = existing let existing_map: HashMap<String, StoredMail> = existing
@@ -592,11 +633,12 @@ pub(crate) async fn sync_folder(
attachments_pending: existing.attachments_pending, attachments_pending: existing.attachments_pending,
}); });
} else { } 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 { updated.push(StoredMail {
mail: mail.clone(), mail: mail.clone(),
details: None, details: None,
rfc2822: Some(rfc2822), rfc2822: None,
uid, uid,
attachments_pending: false, attachments_pending: false,
}); });
@@ -644,12 +686,24 @@ async fn prefetch_details(
local_store: &LocalStore, local_store: &LocalStore,
backend: &dyn MailBackend, backend: &dyn MailBackend,
folder: &FolderInfo, folder: &FolderInfo,
body_prefetch_limit: usize,
attachment_retry_history: &mut HashMap<String, std::time::Instant>, attachment_retry_history: &mut HashMap<String, std::time::Instant>,
) { ) {
let mails = store.get_folder(&folder.id).await; let mails = store.get_folder(&folder.id).await;
// First pass — fresh mails still missing their `.eml` on disk. // Body prefetch is the *capped* part: eagerly pull bodies only for the
let api_needed: Vec<Mail> = mails // 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<Mail> = prefetch_window
.iter() .iter()
.filter(|m| m.details.is_none()) .filter(|m| m.details.is_none())
.filter_map(|m| { .filter_map(|m| {
-1
View File
@@ -192,7 +192,6 @@ async fn main() -> anyhow::Result<()> {
store.clone(), store.clone(),
local_store.clone(), local_store.clone(),
backend.clone(), backend.clone(),
cfg.sync_limit,
bus_ids_for_handler, bus_ids_for_handler,
event_rx, event_rx,
shutdown_rx.clone(), shutdown_rx.clone(),
+9 -4
View File
@@ -82,18 +82,23 @@ export function ConfigPanel({ config, status, loading, onSave, onRestart }: Prop
/> />
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Mail to sync</label> <label>Offline message bodies</label>
<small className="field-hint">
Your whole mailbox is always listed and searchable by subject, sender
and date. This only sets how many recent message <em>bodies</em> are
kept ready offline older ones load on demand when you open them.
</small>
<label className="checkbox-field"> <label className="checkbox-field">
<input <input
type="checkbox" type="checkbox"
checked={fetchAll} checked={fetchAll}
onChange={(e) => setFetchAll(e.target.checked)} onChange={(e) => setFetchAll(e.target.checked)}
/> />
<span>Fetch all mail (entire account, kept locally)</span> <span>Keep every message body offline (full local copy)</span>
</label> </label>
{fetchAll ? ( {fetchAll ? (
<small className="field-hint"> <small className="field-hint">
Downloads every mail from the start can be slow on large accounts. Downloads every body slow + uses the most disk on large accounts.
</small> </small>
) : ( ) : (
<input <input
@@ -101,7 +106,7 @@ export function ConfigPanel({ config, status, loading, onSave, onRestart }: Prop
min={1} min={1}
value={syncLimit} value={syncLimit}
onChange={(e) => setSyncLimit(Math.max(1, Number(e.target.value)))} onChange={(e) => setSyncLimit(Math.max(1, Number(e.target.value)))}
placeholder="Max mails per folder" placeholder="Bodies to keep offline (most recent)"
/> />
)} )}
</div> </div>