diff --git a/Cargo.lock b/Cargo.lock index a5367d2..f52b98b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -297,6 +297,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b25655df2c3cdd83c5e5b293b88acd880332b2ddadd7c30ac43144fdc0033da9" + [[package]] name = "bichon-admin" version = "2.0.1-alpha.1" @@ -348,7 +354,7 @@ dependencies = [ name = "bichon-cli" version = "2.0.1-alpha.1" dependencies = [ - "base64 0.22.1", + "base64 0.23.0", "bichon-core", "chrono", "clap", @@ -371,7 +377,7 @@ name = "bichon-core" version = "2.0.1-alpha.1" dependencies = [ "async-imap", - "base64 0.22.1", + "base64 0.23.0", "bichon-blob", "bichon-memdb", "blake3", @@ -471,7 +477,7 @@ dependencies = [ name = "bichon-smtp" version = "2.0.1-alpha.1" dependencies = [ - "base64 0.22.1", + "base64 0.23.0", "bichon-core", "lettre", "rcgen", @@ -1976,9 +1982,9 @@ checksum = "e9025058dae765dee5070ec375f591e2ba14638c63feff74f13805a72e523163" [[package]] name = "http" -version = "1.4.2" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" dependencies = [ "bytes 1.12.0", "itoa", @@ -4928,9 +4934,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.52.3" +version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ "bytes 1.12.0", "libc", @@ -5382,9 +5388,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.23.4" +version = "1.24.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53" +checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" dependencies = [ "getrandom 0.4.2", "js-sys", diff --git a/Cargo.toml b/Cargo.toml index 358529f..e2d8bed 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,19 +18,19 @@ edition = "2021" [workspace.dependencies] chrono = "0.4.45" -clap = { version = "4.6.1", features = ["derive", "env"] } +clap = { version = "4.6", features = ["derive", "env"] } bichon-memdb = { path = "crates/memdb" } bichon-blob = { path = "crates/blob" } itertools = "0.15.0" ring = { version = "0.17.14", features = ["std"] } -serde = { version = "1.0.228", features = ["derive"] } -serde_json = "1.0.150" -tokio = { version = "1.52.3", features = ["full"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +tokio = { version = "1.53", features = ["full"] } tracing = "0.1.44" tracing-appender = "0.2.3" tracing-subscriber = { version = "0.3.23", features = ["env-filter", "json"] } -base64 = "0.22.1" -snafu = "0.9.1" +base64 = "0.23" +snafu = "0.9" reqwest = { version = "0.12.24", default-features = false, features = [ "json", "stream", @@ -40,10 +40,10 @@ reqwest = { version = "0.12.24", default-features = false, features = [ "socks", ] } tokio-socks = "0.5.3" -http = "1.4.2" +http = "1.5" regex = "1.13" email_address = "0.2.9" -futures = "0.3.32" +futures = "0.3" utf7-imap = "0.3.2" mail-parser = { version = '0.11', features = ["serde"] } # mail-send = "0.5.2" @@ -58,16 +58,16 @@ sysinfo = "0.39" num_cpus = "1.17.0" rand = "0.10.2" encoding_rs = "0.8.35" -webpki-roots = "1.0.8" -rustls = { version = "0.23.41", default-features = false, features = ["ring"] } -rustls-pki-types = "1.15.0" +webpki-roots = "1.0" +rustls = { version = "0.23", default-features = false, features = ["ring"] } +rustls-pki-types = "1.15" tokio-io-timeout = "1.2.1" semver = "1.0.28" governor = "0.10.4" lru = "0.18.1" mime_guess = "2.0.5" hex = "0.4.3" -time = { version = "0.3.53", features = [ +time = { version = "0.3", features = [ "formatting", "parsing", "local-offset", @@ -86,10 +86,10 @@ mail-send = "0.6.1" rcgen = "0.14.8" rustls-pemfile = "2.2.0" blake3 = "1.8.5" -uuid = { version = "1.23.4", features = ["v4", "serde"] } -fjall = { version = "3.1.6", features = ["lz4", "metrics", "bytes_1"] } +uuid = { version = "1.24", features = ["v4", "serde"] } +fjall = { version = "3.1", features = ["lz4", "metrics", "bytes_1"] } tracing-log = "0.2.0" -tokio-util = "0.7.18" +tokio-util = "0.7" indicatif = "0.18.6" [profile.release] diff --git a/crates/core/src/account/state.rs b/crates/core/src/account/state.rs index c415006..3fd606f 100644 --- a/crates/core/src/account/state.rs +++ b/crates/core/src/account/state.rs @@ -40,6 +40,10 @@ pub enum TriggerType { Manual, #[default] Scheduled, + /// Full re-sync (UID SEARCH ALL) invoked explicitly, e.g. to repair a + /// mailbox whose incremental download was interrupted. Semantically a + /// manual trigger, tracked distinctly for diagnostics. + SyncFull, } #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)] @@ -189,6 +193,47 @@ impl DownloadState { }) } + /// Moves a stale Running session into history as Cancelled. + /// + /// A Running `active_session` that survives an Idle decision means the + /// previous run was interrupted without a clean shutdown (e.g. process + /// killed mid-download). Leaving it in place makes the UI show a phantom + /// "syncing" state even though nothing is downloading. Callers invoke this + /// only when no download is actually running for the account, so a + /// legitimately active session is never touched. + /// + /// Returns `true` if a stale session was finalized (i.e. the previous sync + /// did not finish) and `false` otherwise. + pub fn finalize_stale_session(account_id: u64) -> BichonResult { + let stale = Self::get(account_id)? + .and_then(|s| s.active_session) + .map_or(false, |s| s.status == DownloadStatus::Running); + if stale { + Self::update_state(account_id, move |current| { + let mut updated = current.clone(); + if updated + .active_session + .as_ref() + .map_or(false, |s| s.status == DownloadStatus::Running) + { + if let Some(mut session) = updated.active_session.take() { + session.status = DownloadStatus::Cancelled; + session.end_time = Some(utc_now!()); + session.message = Some( + "Previous sync did not finish cleanly; marked as cancelled on startup." + .into(), + ); + updated.history.push(session); + updated.last_finished_at = Some(utc_now!()); + updated.active_session = None; + } + } + Ok(updated) + })?; + } + Ok(stale) + } + pub fn update_folder_progress( account_id: u64, folder_name: String, @@ -220,6 +265,19 @@ impl DownloadState { }) } + /// Touches only `current_folder` without rewriting folder progress. Lets + /// long-running IMAP operations (e.g. waiting on a slow server mid-batch) + /// keep the UI's "last activity" indicator fresh without spamming writes. + pub fn set_current_folder(account_id: u64, folder_name: String) -> BichonResult<()> { + Self::update_state(account_id, move |state| { + let mut updated = state.clone(); + if let Some(ref mut session) = updated.active_session { + session.current_folder = Some(folder_name); + } + Ok(updated) + }) + } + pub fn init_folder_details(account_id: u64, folders: Vec) -> BichonResult<()> { Self::update_state(account_id, move |state| { let mut updated = state.clone(); diff --git a/crates/core/src/cache/imap/download/download_type.rs b/crates/core/src/cache/imap/download/download_type.rs index 1bf691f..055a09b 100644 --- a/crates/core/src/cache/imap/download/download_type.rs +++ b/crates/core/src/cache/imap/download/download_type.rs @@ -53,6 +53,7 @@ pub async fn decide_next_download_task( let should_start = match trigger_type { TriggerType::Manual => true, + TriggerType::SyncFull => true, TriggerType::Scheduled => { let now = utc_now!(); let cooldown_ok = now - state.last_finished_at.unwrap_or(0) > 60 * 1000; @@ -73,10 +74,14 @@ pub async fn decide_next_download_task( DownloadState::start_new_session(account.id, trigger_type)?; Ok(DownloadTask::TraceFetch) } else { + // Nothing to download right now. A Running active_session at this point + // is a leftover from an interrupted run (a real download would have + // been blocked by the busy guard before reaching here), so mark it + // Cancelled instead of leaving the UI showing a phantom "syncing". + DownloadState::finalize_stale_session(account.id)?; Ok(DownloadTask::Idle) } } - fn should_trigger_next_download(last_trigger_at: i64, sync_interval_min: i64) -> bool { let now = utc_now!(); now - last_trigger_at > (sync_interval_min * 60 * 1000) diff --git a/crates/core/src/cache/imap/download/flow.rs b/crates/core/src/cache/imap/download/flow.rs index 3fc0b7b..d41f53e 100644 --- a/crates/core/src/cache/imap/download/flow.rs +++ b/crates/core/src/cache/imap/download/flow.rs @@ -33,7 +33,8 @@ use crate::{ }, error::{code::ErrorCode, BichonResult}, imap::executor::{ - compress_uid_list, generate_uid_sequence_hashset, ImapExecutor, DEFAULT_BATCH_SIZE, + compress_uid_list, generate_uid_sequence_hashset, slow_server_message, ImapExecutor, + DEFAULT_BATCH_SIZE, }, store::tantivy::envelope::ENVELOPE_MANAGER, }, @@ -44,7 +45,6 @@ use tracing::{debug, error, info, warn}; const MAX_NETWORK_RETRIES: u32 = 3; - #[derive(Clone, Debug, Eq, PartialEq)] pub enum FetchDirection { Since, @@ -163,13 +163,24 @@ pub async fn fetch_and_save_by_date( &batch.0, account.max_email_size_bytes, token.clone(), + Some(&|cumulative, avg_secs, stall_secs| { + // Per-message progress: the current batch's cumulative count + // keeps the UI moving while a slow server trickles messages. + let _ = DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + planned, + cumulative, + FolderStatus::Downloading, + slow_server_message(avg_secs, stall_secs), + ); + Ok(()) + }), ) .await { - Ok(processed) => break Ok(processed), - Err(e) - if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => - { + Ok((processed, _throttled)) => break Ok(processed), + Err(e) if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => { retries += 1; warn!( account_id, @@ -183,22 +194,16 @@ pub async fn fetch_and_save_by_date( match ImapExecutor::create_connection(account_id).await { Ok(new_session) => { session = new_session; - if let Err(e2) = session.examine(&mailbox.encoded_name()).await - { - let err_msg = format!( - "Re-examine failed after reconnect: {:#?}", - e2 - ); - DownloadState::append_session_error( - account_id, - err_msg, - )?; + if let Err(e2) = session.examine(&mailbox.encoded_name()).await { + let err_msg = + format!("Re-examine failed after reconnect: {:#?}", e2); + DownloadState::append_session_error(account_id, err_msg)?; break Err(e); } - tokio::time::sleep(Duration::from_secs( - 1 << (retries - 1), - )) - .await; + // Longer backoff than the original 1s/2s/4s: a + // throttling server needs time to recover. + let backoff = [5u64, 15, 30][(retries - 1) as usize]; + tokio::time::sleep(Duration::from_secs(backoff)).await; continue; } Err(e2) => { @@ -254,6 +259,11 @@ pub async fn fetch_and_save_by_date( /// Fetches all messages from a mailbox. /// Returns `Ok(Some(max_uid))` with the highest UID stored, or `Ok(None)` if empty. +/// +/// The mailbox is enumerated via `UID SEARCH ALL` first, then downloaded in UID +/// batches. Unlike sequence-number paging, UIDs stay stable while the download +/// runs (new arrivals only get larger UIDs), so no message is silently skipped +/// when the server changes mid-download. pub async fn fetch_and_save_full_mailbox( account: &AccountModel, mailbox: &MailBox, @@ -279,41 +289,57 @@ pub async fn fetch_and_save_full_mailbox( } }; - let total = match session.examine(&mailbox.encoded_name()).await { - Ok(mailbox) => mailbox.exists as u64, - Err(e) => { - let err_msg = format!("Failed to examine folder [{}]: {:#?}", mailbox.name, e); - DownloadState::update_folder_progress( - account_id, - mailbox.name.clone(), - mailbox.exists as u64, - 0, - FolderStatus::Failed, - Some(err_msg.clone()), - )?; + let uid_list = + match ImapExecutor::uid_search_all_mailbox(&mut session, &mailbox.encoded_name()).await { + Ok(list) => list, + Err(e) => { + let err_msg = format!("UID SEARCH failed in [{}]: {:#?}", mailbox.name, e); + DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + 0, + 0, + FolderStatus::Failed, + Some(err_msg.clone()), + )?; + DownloadState::append_session_error(account_id, err_msg)?; + session.logout().await.ok(); + return Err(e); + } + }; - DownloadState::append_session_error(account_id, err_msg)?; - session.logout().await.ok(); - return Err(raise_error!( - format!("{:#?}", e), - ErrorCode::ImapCommandFailed - )); - } - }; + let planned = uid_list.len() as u64; + if planned == 0 { + info!( + "Mailbox '{}' is empty, no emails to download.", + mailbox.name + ); + DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + 0, + 0, + FolderStatus::Success, + None, + )?; + session.logout().await.ok(); + return Ok(None); + } - let page_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE); - let total_batches = total.div_ceil(page_size as u64); + let max_uid = *uid_list.last().unwrap(); + let page_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize; + let uid_batches = generate_uid_sequence_hashset(uid_list, page_size); + let total_batches = uid_batches.len(); info!( - "Starting full mailbox download for '{}', total={}, batches={}", - mailbox.name, total, total_batches + "Starting full mailbox download for '{}', uids={}, batches={}", + mailbox.name, planned, total_batches ); let mut current_processed = 0u64; let mut has_error_or_cancel = false; - let mut max_uid: Option = None; - for page in 1..=total_batches { + for (index, batch) in uid_batches.into_iter().enumerate() { if token.is_cancelled() { DownloadState::update_session_status( account_id, @@ -323,7 +349,7 @@ pub async fn fetch_and_save_full_mailbox( DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total, + planned, current_processed, FolderStatus::Cancelled, None, @@ -332,48 +358,66 @@ pub async fn fetch_and_save_full_mailbox( break; } + // Heartbeat: keeps `current_folder` fresh so the web UI can show + // "waiting for server" while a slow IMAP server is mid-batch. + DownloadState::set_current_folder(account_id, mailbox.name.clone())?; + + // Heartbeat: keeps `current_folder` fresh so the web UI can show + // "waiting for server" while a slow IMAP server is mid-batch. + DownloadState::set_current_folder(account_id, mailbox.name.clone())?; + let mut retries = 0u32; let batch_result = loop { - match ImapExecutor::batch_retrieve_emails( + match ImapExecutor::uid_batch_retrieve_emails( &mut session, account_id, mailbox_id, - total, - page as u64, - page_size as u64, - &mailbox.encoded_name(), + &batch.0, account.max_email_size_bytes, token.clone(), - &mut max_uid, + Some(&|cumulative, avg_secs, stall_secs| { + // Per-message progress: the current batch's cumulative count + // keeps the UI moving while a slow server trickles messages. + let _ = DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + planned, + cumulative, + FolderStatus::Downloading, + slow_server_message(avg_secs, stall_secs), + ); + Ok(()) + }), ) .await { - Ok(count) => break Ok(count), - Err(e) - if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => - { + Ok((count, _throttled)) => break Ok(count), + Err(e) if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => { retries += 1; warn!( account_id, mailbox = mailbox.name, - page, + index, retries, "Network error on batch, reconnecting ({}/{})", retries, MAX_NETWORK_RETRIES ); + // Refresh the heartbeat after a reconnect too. + let _ = DownloadState::set_current_folder(account_id, mailbox.name.clone()); match ImapExecutor::create_connection(account_id).await { Ok(new_session) => { session = new_session; if let Err(e2) = session.examine(&mailbox.encoded_name()).await { - let err_msg = format!( - "Re-examine failed after reconnect: {:#?}", - e2 - ); + let err_msg = + format!("Re-examine failed after reconnect: {:#?}", e2); DownloadState::append_session_error(account_id, err_msg)?; break Err(e); } - tokio::time::sleep(Duration::from_secs(1 << (retries - 1))).await; + // Longer backoff than the original 1s/2s/4s: a + // throttling server needs time to recover. + let backoff = [5u64, 15, 30][(retries - 1) as usize]; + tokio::time::sleep(Duration::from_secs(backoff)).await; continue; } Err(e2) => { @@ -391,19 +435,19 @@ pub async fn fetch_and_save_full_mailbox( DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total, + planned, current_processed, FolderStatus::Downloading, None, )?; } Err(e) => { - let err_msg = format!("Batch {} failed: {:#?}", page, e); + let err_msg = format!("Batch {} failed: {:#?}", index, e); DownloadState::append_session_error(account_id, err_msg.clone())?; DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total, + planned, current_processed, FolderStatus::Failed, Some(err_msg), @@ -411,21 +455,21 @@ pub async fn fetch_and_save_full_mailbox( has_error_or_cancel = true; break; } - }; + } } if !has_error_or_cancel { DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total, + planned, current_processed, FolderStatus::Success, None, )?; } session.logout().await.ok(); - Ok(max_uid) + Ok(max_uid.into()) } /// Generates a synthetic UIDVALIDITY for IMAP servers that don't provide it. @@ -490,15 +534,13 @@ where Ok(None) => { warn!( attempt = attempt + 1, - max_retries, - "STATUS returned no UIDVALIDITY" + max_retries, "STATUS returned no UIDVALIDITY" ); } Err(e) => { warn!( attempt = attempt + 1, - max_retries, - "UIDVALIDITY fetch attempt failed: {:#?}", e + max_retries, "UIDVALIDITY fetch attempt failed: {:#?}", e ); } } @@ -634,9 +676,7 @@ async fn reconcile_uid_validity_change( .await .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - let batch_size = account - .download_batch_size - .unwrap_or(DEFAULT_BATCH_SIZE) as usize; + let batch_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize; let batches = generate_uid_sequence_hashset(missing_uids, batch_size); let mut downloaded = 0u64; @@ -661,10 +701,23 @@ async fn reconcile_uid_validity_change( &batch.0, account.max_email_size_bytes, token.clone(), + Some(&|cumulative, avg_secs, stall_secs| { + // Per-message progress: the current batch's cumulative count + // keeps the UI moving while a slow server trickles messages. + let _ = DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + planned, + cumulative, + FolderStatus::Downloading, + slow_server_message(avg_secs, stall_secs), + ); + Ok(()) + }), ) .await { - Ok(processed) => { + Ok((processed, _throttled)) => { downloaded += processed; DownloadState::update_folder_progress( account_id, @@ -791,18 +844,12 @@ pub async fn reconcile_mailboxes( account_id, local_mailbox.name, &local_mailbox.uid_validity, &remote_uid_validity ); - reconcile_uid_validity_change( - account, - local_mailbox, - remote_mailbox, - token.clone(), - ) - .await? + reconcile_uid_validity_change(account, local_mailbox, remote_mailbox, token.clone()) + .await? } else { perform_incremental_sync(account, local_mailbox, remote_mailbox, token.clone()) .await? }; - let mut updated = remote_mailbox.clone(); updated.highest_uid = new_highest_uid; // Update uid_validity with the resolved value (either from server or synthetic) @@ -924,33 +971,15 @@ async fn perform_incremental_sync( // Use stored highest_uid if available; otherwise fall back to Tantivy // query once (backward compatibility with pre-existing databases). let start_uid = match local_mailbox.highest_uid { - Some(uid) => { - tracing::info!( - "[account {}][mailbox {}] perform_incremental_sync: stored highest_uid={}, remote.exists={}", - account.id, - local_mailbox.name, - uid, - remote_mailbox.exists - ); - uid as u64 + 1 - } + Some(uid) => uid as u64 + 1, None => { - let local_max_uid = - ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?; - tracing::info!( - "[account {}][mailbox {}] perform_incremental_sync: highest_uid unset, Tantivy max_uid={:?}, remote.exists={}", - account.id, - local_mailbox.name, - local_max_uid, - remote_mailbox.exists - ); + let local_max_uid = ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?; match local_max_uid { Some(uid) => uid + 1, None => { info!( "No maximum UID found in index for mailbox, assuming local storage is missing." ); - let result = match &account.date_since { Some(date_since) => { fetch_and_save_by_date( @@ -974,10 +1003,8 @@ async fn perform_incremental_sync( .await? } None => { - fetch_and_save_full_mailbox( - account, remote_mailbox, token, - ) - .await? + fetch_and_save_full_mailbox(account, remote_mailbox, token) + .await? } }, }; @@ -993,6 +1020,19 @@ async fn perform_incremental_sync( .as_ref() .map(|r| r.calculate_date()) .transpose()?; + if start_uid == 1 { + // No stored highest_uid and no indexed messages: fall back to a + // full mailbox download. Tell the UI up-front so it doesn't show + // a stale "Pending" while the (possibly large) mailbox streams. + let _ = DownloadState::update_folder_progress( + account.id, + remote_mailbox.name.clone(), + 0, + 0, + FolderStatus::Downloading, + Some("Full mailbox download".into()), + )?; + } let new_max_uid = ImapExecutor::fetch_new_mail( &mut session, @@ -1034,7 +1074,10 @@ mod tests { fn test_generate_synthetic_uidvalidity_different_mailboxes() { let inbox = generate_synthetic_uidvalidity("INBOX"); let sent = generate_synthetic_uidvalidity("Sent"); - assert_ne!(inbox, sent, "different mailboxes should have different uid_validity"); + assert_ne!( + inbox, sent, + "different mailboxes should have different uid_validity" + ); } #[test] @@ -1070,10 +1113,8 @@ mod tests { // Ensure a rustls crypto provider is installed (ring). // May already be installed by production code; ignore duplicate. - rustls::crypto::CryptoProvider::install_default( - rustls::crypto::ring::default_provider(), - ) - .ok(); + rustls::crypto::CryptoProvider::install_default(rustls::crypto::ring::default_provider()) + .ok(); let tcp = TcpStream::connect((host, port)) .await @@ -1084,8 +1125,8 @@ mod tests { let timeout_stream = TimeoutStream::new(tcp); let pinned = Box::pin(timeout_stream); - let server_name = ServerName::try_from(host.to_owned()) - .map_err(|e| format!("Invalid hostname: {e}"))?; + let server_name = + ServerName::try_from(host.to_owned()).map_err(|e| format!("Invalid hostname: {e}"))?; let config = ClientConfig::builder() .with_root_certificates(rustls::RootCertStore { @@ -1208,11 +1249,7 @@ mod tests { session.logout().await.ok(); - println!( - "Call {}: UIDVALIDITY = {:?}", - i + 1, - status.uid_validity - ); + println!("Call {}: UIDVALIDITY = {:?}", i + 1, status.uid_validity); results.borrow_mut().push(status.uid_validity); } @@ -1275,13 +1312,10 @@ mod tests { let sample: Vec = all_uids.into_iter().take(5).collect(); let uid_set = compress_uid_list(sample.clone()); - let result = ImapExecutor::fetch_uid_metadata( - &mut session, - &uid_set, - CancellationToken::new(), - ) - .await - .expect("fetch_uid_metadata should succeed"); + let result = + ImapExecutor::fetch_uid_metadata(&mut session, &uid_set, CancellationToken::new()) + .await + .expect("fetch_uid_metadata should succeed"); session.logout().await.ok(); @@ -1329,8 +1363,7 @@ mod tests { #[tokio::test] async fn test_retry_first_attempt_succeeds() { - let result = fetch_uid_validity_with_retry_inner(3, mock_results(vec![Ok(Some(42))])) - .await; + let result = fetch_uid_validity_with_retry_inner(3, mock_results(vec![Ok(Some(42))])).await; assert_eq!(result.unwrap(), Some(42)); } @@ -1392,13 +1425,7 @@ mod tests { // max_retries=5, success on 5th attempt let result = fetch_uid_validity_with_retry_inner( 5, - mock_results(vec![ - Ok(None), - Ok(None), - Ok(None), - Ok(None), - Ok(Some(5)), - ]), + mock_results(vec![Ok(None), Ok(None), Ok(None), Ok(None), Ok(Some(5))]), ) .await; assert_eq!(result.unwrap(), Some(5)); @@ -1418,11 +1445,7 @@ mod tests { #[tokio::test] async fn test_retry_max_retries_zero() { // max_retries=0 means no attempts at all - let result = fetch_uid_validity_with_retry_inner( - 0, - mock_results(vec![Ok(Some(42))]), - ) - .await; + let result = fetch_uid_validity_with_retry_inner(0, mock_results(vec![Ok(Some(42))])).await; assert_eq!(result.unwrap(), None); } @@ -1431,8 +1454,8 @@ mod tests { // ============================================================ use crate::imap::mock_server::{ - examine_response, uid_fetch_metadata_response, uid_fetch_rfc822_response, - minimal_eml, MockImapServer, MockImapServerHandle, + examine_response, minimal_eml, uid_fetch_metadata_response, uid_fetch_rfc822_response, + MockImapServer, MockImapServerHandle, }; /// Build an `async_imap::Session` connected to the mock server, @@ -1453,9 +1476,11 @@ mod tests { client.read_response().await.unwrap(); // Login - let mut session = client.login("user", "pass").await.map_err(|(e, _)| { - panic!("Login failed: {e:?}") - }).unwrap(); + let mut session = client + .login("user", "pass") + .await + .map_err(|(e, _)| panic!("Login failed: {e:?}")) + .unwrap(); // Examine session.examine("INBOX").await.unwrap(); @@ -1468,37 +1493,28 @@ mod tests { let handle = MockImapServer::new() .respond("LOGIN", "{TAG} OK LOGIN done\r\n") .respond("EXAMINE", examine_response("INBOX", 3, 42, 4)) - .respond("UID FETCH", uid_fetch_metadata_response(&[ - (1, ""), - (2, ""), - (3, ""), - ])) + .respond( + "UID FETCH", + uid_fetch_metadata_response(&[ + (1, ""), + (2, ""), + (3, ""), + ]), + ) .start() .await; let mut session = mock_session(&handle).await; - let result = ImapExecutor::fetch_uid_metadata( - &mut session, - "1:3", - CancellationToken::new(), - ) - .await - .unwrap(); + let result = + ImapExecutor::fetch_uid_metadata(&mut session, "1:3", CancellationToken::new()) + .await + .unwrap(); assert_eq!(result.len(), 3); - assert_eq!( - result.get(&1).unwrap().as_deref(), - Some("msg-a@test.com") - ); - assert_eq!( - result.get(&2).unwrap().as_deref(), - Some("msg-b@test.com") - ); - assert_eq!( - result.get(&3).unwrap().as_deref(), - Some("msg-c@test.com") - ); + assert_eq!(result.get(&1).unwrap().as_deref(), Some("msg-a@test.com")); + assert_eq!(result.get(&2).unwrap().as_deref(), Some("msg-b@test.com")); + assert_eq!(result.get(&3).unwrap().as_deref(), Some("msg-c@test.com")); session.logout().await.ok(); } @@ -1523,10 +1539,7 @@ mod tests { let mut session = mock_session(&handle).await; let uids: Vec = { - let mut stream = session - .uid_fetch("1:2", "(UID FLAGS)") - .await - .unwrap(); + let mut stream = session.uid_fetch("1:2", "(UID FLAGS)").await.unwrap(); use futures::TryStreamExt; let mut uids = Vec::new(); @@ -1557,10 +1570,7 @@ mod tests { let mut session = mock_session(&handle).await; let bodies: Vec<(u32, Vec)> = { - let mut stream = session - .uid_fetch("1:1", "(UID BODY[])") - .await - .unwrap(); + let mut stream = session.uid_fetch("1:1", "(UID BODY[])").await.unwrap(); use futures::TryStreamExt; let mut bodies = Vec::new(); @@ -1591,18 +1601,12 @@ mod tests { let mut session = mock_session(&handle).await; - let result = ImapExecutor::fetch_uid_metadata( - &mut session, - "1:*", - CancellationToken::new(), - ) - .await - .unwrap(); + let result = + ImapExecutor::fetch_uid_metadata(&mut session, "1:*", CancellationToken::new()) + .await + .unwrap(); - assert!( - result.is_empty(), - "empty mailbox should return empty map" - ); + assert!(result.is_empty(), "empty mailbox should return empty map"); session.logout().await.ok(); } @@ -1610,8 +1614,7 @@ mod tests { #[tokio::test] async fn fetch_uid_metadata_missing_message_id() { // One entry has a Message-ID, the other has no header at all. - let header_with_msgid = - "From: sender@example.com\r\n\ + let header_with_msgid = "From: sender@example.com\r\n\ Date: Thu, 01 Jan 2025 00:00:00 +0000\r\n\ Message-ID: \r\n\r\n"; let header_without_msgid = "\r\n"; @@ -1637,19 +1640,13 @@ mod tests { let mut session = mock_session(&handle).await; - let result = ImapExecutor::fetch_uid_metadata( - &mut session, - "1:2", - CancellationToken::new(), - ) - .await - .unwrap(); + let result = + ImapExecutor::fetch_uid_metadata(&mut session, "1:2", CancellationToken::new()) + .await + .unwrap(); assert_eq!(result.len(), 2); - assert_eq!( - result.get(&1).unwrap().as_deref(), - Some("ok@test.com") - ); + assert_eq!(result.get(&1).unwrap().as_deref(), Some("ok@test.com")); // UID 2 has no Message-ID header → None assert_eq!(result.get(&2).unwrap().as_deref(), None); diff --git a/crates/core/src/context/executors.rs b/crates/core/src/context/executors.rs index 7970a5d..b47f952 100644 --- a/crates/core/src/context/executors.rs +++ b/crates/core/src/context/executors.rs @@ -17,6 +17,7 @@ // along with this program. If not, see . use crate::account::migration::AccountType; +use crate::account::state::DownloadState; use crate::context::Initialize; use crate::{ { @@ -25,7 +26,7 @@ use crate::{ utc_now, }; use std::sync::LazyLock; -use tracing::info; +use tracing::{info, warn}; pub static BICHON_CONTEXT: LazyLock = LazyLock::new(BichonContext::new); @@ -65,6 +66,28 @@ impl BichonContext { active_accounts.len() ); for account in active_accounts { + // A Running session surviving startup is a leftover from a previous + // interrupted run; nothing is downloading yet at this point. Mark it + // Cancelled so the UI doesn't show a phantom "syncing" state. The + // scheduler starts regardless — its first tick runs immediately, so + // the interrupted run is caught up on, and the session's trigger + // stays Scheduled rather than showing a "Manual" the user never + // initiated. + match DownloadState::finalize_stale_session(account.id) { + Ok(true) => { + info!( + "Account {}: stale sync session finalized on startup.", + account.id + ); + } + Err(e) => { + warn!( + "Failed to finalize stale session for account {}: {:#?}", + account.id, e + ); + } + Ok(false) => {} + } DOWNLOAD_CONTROLLER .trigger_schedule(account.id, account.email) .await diff --git a/crates/core/src/imap/executor.rs b/crates/core/src/imap/executor.rs index 3f5313e..c434142 100644 --- a/crates/core/src/imap/executor.rs +++ b/crates/core/src/imap/executor.rs @@ -33,6 +33,7 @@ use tracing::info; const BODY_FETCH_COMMAND: &str = "(UID INTERNALDATE RFC822.SIZE BODY.PEEK[])"; const SIZE_ONLY_FETCH: &str = "(UID RFC822.SIZE)"; +const MAX_NETWORK_RETRIES: u32 = 3; fn classify_imap_error(e: &async_imap::error::Error) -> ErrorCode { match e { @@ -97,6 +98,30 @@ impl ImapExecutor { .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e))) } + /// Enumerate every UID currently present in the mailbox via `UID SEARCH ALL`. + /// + /// This is the drift-safe way to page through a full mailbox: UIDs are stable + /// while the download runs (new arrivals only get larger UIDs), whereas + /// sequence numbers shift when mail is added or removed mid-download, which + /// silently skips messages. The caller owns the returned list and decides + /// how to batch it (see `generate_uid_sequence_hashset`). + pub async fn uid_search_all_mailbox( + session: &mut Session>, + mailbox_name: &str, + ) -> BichonResult> { + session + .examine(mailbox_name) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; + let results = session + .uid_search("ALL") + .await + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; + let mut uids: Vec = results.into_iter().collect(); + uids.sort(); + Ok(uids) + } + /// Fetches new mail for a mailbox. /// /// When `before` is `Some(date)`, a two-step approach is used: @@ -201,15 +226,33 @@ impl ImapExecutor { ErrorCode::InternalError )); } - let processed = Self::uid_batch_retrieve_emails( + let (processed, throttled) = Self::uid_batch_retrieve_emails( session, account.id, mailbox.id, &batch.0, account.max_email_size_bytes, token.clone(), + Some(&|cumulative, avg_secs, stall_secs| { + // Per-message progress: the current batch's cumulative count + // keeps the UI moving while a slow server trickles messages. + let _ = DownloadState::update_folder_progress( + account.id, + mailbox.name.clone(), + planned, + cumulative, + FolderStatus::Downloading, + slow_server_message(avg_secs, stall_secs), + ); + Ok(()) + }), ) .await?; + if throttled { + // Server appears to be rate-limiting; back off before the next + // batch so we don't hammer the limiter with back-to-back bursts. + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + } count += processed; DownloadState::update_folder_progress( account.id, @@ -233,8 +276,15 @@ impl ImapExecutor { Ok(max_uid) } - /// Direct ranged UID FETCH without date filtering. Streams results from - /// the server in a single IMAP round-trip. + /// Fetches all messages with UID >= start_uid via batched UID FETCH. + /// + /// A single ranged `UID FETCH {start}:*` can block for minutes on slow + /// servers pushing hundreds of messages, and hits the socket read timeout + /// if the server stalls, with zero progress feedback in the meantime. + /// Instead, enumerate the UIDs first, then download in small batches — + /// each batch is a short round-trip with a SIZE pre-check (oversized + /// messages are skipped without fetching their body), progress is reported + /// per batch, and the whole download stays responsive to cancellation. async fn fetch_new_mail_range( session: &mut Session>, account: &AccountModel, @@ -244,32 +294,48 @@ impl ImapExecutor { ) -> BichonResult> { let uid_range = format!("{start_uid}:*"); info!( - "[account {}][mailbox {}] fetch_new_mail: direct UID FETCH {}", + "[account {}][mailbox {}] fetch_new_mail: batched UID FETCH {}", account.id, mailbox.name, uid_range ); - let mut stream = session - .uid_fetch(&uid_range, BODY_FETCH_COMMAND) - .await - .map_err(|e| { - let err_msg = format!("UID FETCH failed in [{}]: {:#?}", mailbox.name, e); - let _ = DownloadState::append_session_error(account.id, err_msg); - raise_error!(format!("{:#?}", e), classify_imap_error(&e)) - })?; + let results = session.uid_search(&uid_range).await.map_err(|e| { + let err_msg = format!("UID SEARCH failed in [{}]: {:#?}", mailbox.name, e); + let _ = DownloadState::append_session_error(account.id, err_msg); + raise_error!(format!("{:#?}", e), classify_imap_error(&e)) + })?; + let mut uid_vec: Vec = results.into_iter().collect(); + uid_vec.sort(); + + if uid_vec.is_empty() { + DownloadState::update_folder_progress( + account.id, + mailbox.name.clone(), + 0, + 0, + FolderStatus::Success, + Some("No new emails found.".into()), + )?; + return Ok(None); + } + + let max_uid = uid_vec.last().copied(); + let planned = uid_vec.len() as u64; + let batch_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize; + let batches = generate_uid_sequence_hashset(uid_vec, batch_size); + let total_batches = batches.len(); + + DownloadState::update_folder_progress( + account.id, + mailbox.name.clone(), + planned, + 0, + FolderStatus::Downloading, + None, + )?; let mut count = 0u64; - let mut skipped = 0u64; - let mut max_uid: Option = None; - let size_limit = account - .max_email_size_bytes - .unwrap_or(DEFAULT_MAX_EMAIL_SIZE); - while let Some(fetch) = stream - .try_next() - .await - .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? - { + for (index, batch) in batches.into_iter().enumerate() { if token.is_cancelled() { - tracing::info!("Account {}: fetch_new_mail stream interrupted.", account.id); DownloadState::update_session_status( account.id, DownloadStatus::Cancelled, @@ -281,149 +347,134 @@ impl ImapExecutor { )); } - let msg_size = fetch.size.unwrap_or(0) as u64; - if msg_size > 0 && msg_size > size_limit { - tracing::warn!( - account_id = account.id, - mailbox_id = mailbox.id, - uid = fetch.uid, - size = msg_size, - limit = size_limit, - "Skipping oversized email (streaming mode)" - ); - skipped += 1; - continue; + // A slow server can stall a batch past the socket read timeout. + // Retry such batches on a fresh connection instead of failing the + // whole sync session. + let mut retries = 0u32; + let batch_result = loop { + match Self::uid_batch_retrieve_emails( + session, + account.id, + mailbox.id, + &batch.0, + account.max_email_size_bytes, + token.clone(), + Some(&|cumulative, avg_secs, stall_secs| { + // Report per-message so the UI moves even while a slow + // server trickles out the current batch. + DownloadState::update_folder_progress( + account.id, + mailbox.name.clone(), + planned, + cumulative, + FolderStatus::Downloading, + slow_server_message(avg_secs, stall_secs), + ) + }), + ) + .await + { + Ok(processed) => break Ok(processed), + Err(e) + if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => + { + retries += 1; + tracing::warn!( + account_id = account.id, + mailbox = mailbox.name, + index, + retries, + "Network error on batch, reconnecting ({}/{})", + retries, + MAX_NETWORK_RETRIES + ); + match ImapExecutor::create_connection(account.id).await { + Ok(new_session) => { + *session = new_session; + if let Err(e2) = session.examine(&mailbox.encoded_name()).await { + let err_msg = + format!("Re-examine failed after reconnect: {:#?}", e2); + DownloadState::append_session_error(account.id, err_msg)?; + break Err(e); + } + // Longer backoff than the original 1s/2s/4s: a + // throttling server needs time to recover. + let backoff = [5u64, 15, 30][(retries - 1) as usize]; + tracing::warn!( + account_id = account.id, + mailbox = mailbox.name, + "Backing off {}s before retrying batch", + backoff + ); + tokio::time::sleep(std::time::Duration::from_secs(backoff)).await; + continue; + } + Err(e2) => { + tracing::error!( + account_id = account.id, + "Reconnection failed: {:#?}", + e2 + ); + break Err(e); + } + } + } + Err(e) => break Err(e), + } + }; + match batch_result { + Ok((processed, _throttled)) => { + count += processed; + DownloadState::update_folder_progress( + account.id, + mailbox.name.clone(), + planned, + count, + FolderStatus::Downloading, + None, + )?; + tracing::debug!( + "[account {}][mailbox {}] fetch_new_mail: batch {}/{} done ({} processed)", + account.id, + mailbox.name, + index + 1, + total_batches, + processed + ); + } + Err(e) => { + DownloadState::append_session_error(account.id, format!("{:#?}", e))?; + return Err(e); + } } - - if let Some(uid) = fetch.uid { - max_uid = Some(max_uid.unwrap_or(0).max(uid)); + if count == planned { + break; } - extract_envelope_and_store_it(fetch, account.id, mailbox.id).await?; - count += 1; } - let total = count + skipped; - if total == 0 { - DownloadState::update_folder_progress( - account.id, - mailbox.name.clone(), - 0, - 0, - FolderStatus::Success, - Some("No new emails found.".into()), - )?; - } else { - DownloadState::update_folder_progress( - account.id, - mailbox.name.clone(), - total, - count, - FolderStatus::Success, - if skipped > 0 { - Some(format!("{skipped} email(s) skipped due to size limit")) - } else { - None - }, - )?; - } + DownloadState::update_folder_progress( + account.id, + mailbox.name.clone(), + count, + count, + FolderStatus::Success, + None, + )?; Ok(max_uid) } - pub async fn batch_retrieve_emails( - session: &mut Session>, - account_id: u64, - mailbox_id: u64, - total: u64, - page: u64, - page_size: u64, - encoded_mailbox_name: &str, - max_email_size_bytes: Option, - token: CancellationToken, - max_uid: &mut Option, - ) -> BichonResult { - assert!(page > 0, "Page number must be greater than 0"); - assert!(page_size > 0, "Page size must be greater than 0"); - - // Fetch messages starting from the oldest (ascending order). - let start = (page - 1) * page_size + 1; - if start > total { - return Ok(0); - } - let end = (start + page_size - 1).min(total); - - let sequence_set = format!("{}:{}", start, end); - info!( - "Fetching mailbox '{}' messages: sequence {} (page {}, page_size {})", - encoded_mailbox_name, sequence_set, page, page_size - ); - - let limit = max_email_size_bytes.unwrap_or(DEFAULT_MAX_EMAIL_SIZE); - - // PASS 1: fetch only SIZE to identify oversized messages - let acceptable_uids = { - let mut size_stream = session - .fetch(sequence_set.as_str(), SIZE_ONLY_FETCH) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; - - let mut uids: Vec = Vec::new(); - while let Some(fetch) = size_stream - .try_next() - .await - .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? - { - let uid = fetch.uid.unwrap_or(0); - let msg_size = fetch.size.unwrap_or(0) as u64; - if msg_size == 0 || msg_size <= limit { - uids.push(uid); - } else { - tracing::warn!( - account_id, - mailbox_id, - uid, - size = msg_size, - limit, - "Skipping oversized email" - ); - } - } - uids - }; - - if acceptable_uids.is_empty() { - return Ok(0); - } - - // PASS 2: fetch bodies only for acceptable UIDs - let filtered = compress_uid_list(acceptable_uids); - let mut body_stream = session - .uid_fetch(&filtered, BODY_FETCH_COMMAND) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; - - let mut count = 0; - while let Some(fetch) = body_stream - .try_next() - .await - .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? - { - if token.is_cancelled() { - tracing::info!("Account {}: UID fetch stream interrupted.", account_id); - return Err(raise_error!( - "Stream cancelled".into(), - ErrorCode::InternalError - )); - } - if let Some(uid) = fetch.uid { - *max_uid = Some((*max_uid).unwrap_or(0).max(uid)); - } - extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?; - count += 1; - } - Ok(count) - } - + /// Downloads the bodies of `uid_set` in one batch. + /// + /// `progress` (if given) is invoked after each stored message with the + /// cumulative count for the whole mailbox, the current average + /// inter-message interval in seconds (None until at least two messages + /// arrived), and the current stall duration in seconds when the server is + /// silent (None when a message just arrived). The UI can then move per + /// message — and detect a slow server — instead of only updating when a + /// batch finishes. Slow servers push a batch's messages over many seconds + /// (even minutes); without per-message updates the UI freezes and looks + /// stuck. pub async fn uid_batch_retrieve_emails( session: &mut Session>, account_id: u64, @@ -431,7 +482,10 @@ impl ImapExecutor { uid_set: &str, max_email_size_bytes: Option, token: CancellationToken, - ) -> BichonResult { + progress: Option< + &(dyn Fn(u64, Option, Option) -> BichonResult<()> + Send + Sync), + >, + ) -> BichonResult<(u64, bool)> { let limit = max_email_size_bytes.unwrap_or(DEFAULT_MAX_EMAIL_SIZE); // PASS 1: fetch only SIZE to identify oversized messages @@ -466,7 +520,7 @@ impl ImapExecutor { }; if acceptable_uids.is_empty() { - return Ok(0); + return Ok((0, false)); } // PASS 2: fetch bodies only for acceptable UIDs @@ -477,11 +531,43 @@ impl ImapExecutor { .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; let mut count = 0u64; - while let Some(fetch) = body_stream - .try_next() - .await - .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? - { + // Sliding window of recent per-message receive times, used to estimate + // how slow the server is: slow servers push messages seconds apart. + let mut recv_times: std::collections::VecDeque = + std::collections::VecDeque::with_capacity(11); + let mut last_recv = std::time::Instant::now(); + // Consecutive stall reports. A high value means the server is + // throttling us; the caller sleeps before the next batch to avoid + // hammering the rate limiter back-to-back. + let mut consecutive_stalls = 0u32; + // While the server is silent, re-report the current wait every few + // seconds so the UI shows the wait climbing instead of a stale + // average (the average only moves when a message actually arrives). + const STALL_REPORT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); + loop { + let item = + match tokio::time::timeout(STALL_REPORT_INTERVAL, body_stream.try_next()).await { + Ok(item) => item + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?, + Err(_) => { + if token.is_cancelled() { + tracing::info!("Account {}: UID fetch stream interrupted.", account_id); + return Err(raise_error!( + "Stream cancelled".into(), + ErrorCode::InternalError + )); + } + let stall_secs = last_recv.elapsed().as_secs_f64(); + consecutive_stalls += 1; + + if let Some(progress) = progress { + progress(count, None, Some(stall_secs))?; + } + continue; + } + }; + let Some(fetch) = item else { break }; + if token.is_cancelled() { tracing::info!("Account {}: UID fetch stream interrupted.", account_id); return Err(raise_error!( @@ -489,10 +575,29 @@ impl ImapExecutor { ErrorCode::InternalError )); } + let now = std::time::Instant::now(); + consecutive_stalls = 0; + last_recv = now; + recv_times.push_back(now); + if recv_times.len() > 10 { + recv_times.pop_front(); + } + let avg_secs = if recv_times.len() >= 2 { + let span = recv_times + .back() + .unwrap() + .duration_since(*recv_times.front().unwrap()); + Some(span.as_secs_f64() / (recv_times.len() as f64 - 1.0)) + } else { + None + }; extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?; count += 1; + if let Some(progress) = progress { + progress(count, avg_secs, None)?; + } } - Ok(count) + Ok((count, consecutive_stalls >= 2)) } /// Fetches the raw RFC822 body of a single message by UID. @@ -588,6 +693,46 @@ impl ImapExecutor { pub const DEFAULT_BATCH_SIZE: u32 = 30; pub const DEFAULT_MAX_EMAIL_SIZE: u64 = 100 * 1024 * 1024; +/// Average seconds between consecutive messages above which the IMAP server is +/// considered slow (a healthy server responds in milliseconds). +const SLOW_SERVER_THRESHOLD_SECS: f64 = 3.0; +/// Seconds of silence before the UI is told the server is stalling on the +/// current message (lower than SLOW_SERVER_THRESHOLD so the message flips to +/// "waiting" promptly once the server stops feeding). +const STALL_REPORT_THRESHOLD_SECS: f64 = 3.0; +/// Silence beyond this is treated as likely rate limiting (throttling), not +/// just slowness — the UI then explains what Bichon is doing about it. +const RATE_LIMIT_THRESHOLD_SECS: f64 = 10.0; + +/// Returns a user-facing hint when the IMAP server is feeding messages slowly +/// or has gone silent, so the UI can explain "this is the server, not Bichon". +/// `None` when the server is responding normally or not enough messages +/// arrived to tell. `stall_secs` (server silent on the current message) takes +/// precedence over the running average. +pub fn slow_server_message(avg_secs: Option, stall_secs: Option) -> Option { + if let Some(stall) = stall_secs { + if stall >= RATE_LIMIT_THRESHOLD_SECS { + return Some(format!( + "Possible IMAP rate limiting: server has been silent for {:.0}s. Bichon is pacing the download (pausing between batches) and will retry with backoff if the connection stalls.", + stall + )); + } + if stall >= STALL_REPORT_THRESHOLD_SECS { + return Some(format!( + "IMAP server is slow: no response for {:.0}s while fetching the next message; download is still in progress.", + stall + )); + } + } + match avg_secs { + Some(secs) if secs >= SLOW_SERVER_THRESHOLD_SECS => Some(format!( + "IMAP server is slow (avg {:.1}s between messages); download is still in progress.", + secs + )), + _ => None, + } +} + /// Compresses a sorted list of UIDs into an IMAP sequence-set string. /// Consecutive UIDs become ranges (e.g. `1:5`), non-consecutive are /// comma-separated (e.g. `1:5,10,12:15`). @@ -737,10 +882,7 @@ mod test { #[test] fn parse_message_id_lowercase() { let header = b"Message-Id: \r\n"; - assert_eq!( - parse_message_id_header(header), - Some("foo@bar.com".into()) - ); + assert_eq!(parse_message_id_header(header), Some("foo@bar.com".into())); } #[test] diff --git a/crates/core/src/settings/cli.rs b/crates/core/src/settings/cli.rs index ea4914e..97558bb 100644 --- a/crates/core/src/settings/cli.rs +++ b/crates/core/src/settings/cli.rs @@ -242,6 +242,14 @@ pub struct Settings { )] pub bichon_sync_concurrency: Option, + #[clap( + long, + env, + default_value = "90", + help = "IMAP socket read timeout in seconds (0 disables the timeout). Servers that throttle or burst slowly (e.g. Zoho) can pause for 30-60s between responses; keep this above the longest expected server silence so throttling surfaces as progress delay, not a failed sync." + )] + pub bichon_imap_timeout_seconds: u64, + #[clap( long, env, diff --git a/crates/core/src/utils/net.rs b/crates/core/src/utils/net.rs index caa449a..fe876d1 100644 --- a/crates/core/src/utils/net.rs +++ b/crates/core/src/utils/net.rs @@ -18,7 +18,7 @@ use crate::error::code::ErrorCode; use crate::raise_error; -use crate::settings::proxy::Proxy; +use crate::settings::{cli::SETTINGS, proxy::Proxy}; use crate::utils::tls::establish_tls_stream; use crate::{error::BichonResult, imap::session::SessionStream}; use base64::{engine::general_purpose, Engine as _}; @@ -90,9 +90,16 @@ pub(crate) async fn establish_tcp_connection_with_timeout( let tcp_stream = connect_with_optional_proxy(use_proxy, address).await?; let mut timeout_stream = TimeoutStream::new(tcp_stream); - // Set read and write timeouts + // Set read and write timeouts. The read timeout bounds how long the sync + // task blocks waiting for a slow IMAP server between commands; a hung + // server therefore surfaces as a network error (and retry) instead of a + // silently stuck download. 0 disables the read timeout (server decides). + let read_timeout = SETTINGS + .bichon_imap_timeout_seconds + .checked_sub(1) + .map(|seconds| Duration::from_secs(seconds.max(1))); timeout_stream.set_write_timeout(Some(Duration::from_secs(15))); - timeout_stream.set_read_timeout(Some(Duration::from_secs(30))); + timeout_stream.set_read_timeout(read_timeout); // Return the timeout-wrapped TCP stream as a Pin Ok(Box::pin(timeout_stream)) diff --git a/web/src/api/account/api.ts b/web/src/api/account/api.ts index 7be68c5..ae1604c 100644 --- a/web/src/api/account/api.ts +++ b/web/src/api/account/api.ts @@ -42,6 +42,7 @@ export enum DownloadStatus { export enum TriggerType { Manual = "Manual", Scheduled = "Scheduled", + SyncFull = "SyncFull", } export enum FolderStatus { diff --git a/web/src/features/accounts/components/running-state-action.tsx b/web/src/features/accounts/components/running-state-action.tsx index 5816aef..104f67f 100644 --- a/web/src/features/accounts/components/running-state-action.tsx +++ b/web/src/features/accounts/components/running-state-action.tsx @@ -24,7 +24,9 @@ import { useTranslation } from 'react-i18next'; import { useCurrentUser } from '@/hooks/use-current-user'; import { toast } from '@/hooks/use-toast'; import { ToastAction } from '@/components/ui/toast'; -import { AccountModel } from '@/api/account/api'; +import { useQuery } from '@tanstack/react-query' +import { download_state, AccountModel, DownloadStatus } from '@/api/account/api'; +import { Loader2 } from 'lucide-react' interface Props { row: Row @@ -44,29 +46,50 @@ export function RunningStateCellAction({ row }: Props) { } const hasPermission = require_any_permission(['system:root', 'account:read_details'], row.original.id) + // Live sync status pill. While a download session is running, poll every 5s + // so the list always reflects progress without needing the dialog open. + const { data: state } = useQuery({ + queryKey: ['running-state', row.original.id], + queryFn: () => download_state(row.original.id), + refetchInterval: (query) => { + const s = query.state.data?.active_session + return s && s.status === DownloadStatus.Running ? 5000 : false + }, + }) + const running = state?.active_session + const isRunning = !!running && running.status === DownloadStatus.Running + return ( - +
+ {isRunning && ( + + + {t('accounts.runningState.syncing')} + + )} + +
) } diff --git a/web/src/features/accounts/components/running-state-dialog.tsx b/web/src/features/accounts/components/running-state-dialog.tsx index 9b1beb3..c39763f 100644 --- a/web/src/features/accounts/components/running-state-dialog.tsx +++ b/web/src/features/accounts/components/running-state-dialog.tsx @@ -26,7 +26,8 @@ import { } from '@/components/ui/dialog' import { Button } from '@/components/ui/button' import { useQuery } from '@tanstack/react-query' -import { download_state, AccountModel, FolderProgress } from '@/api/account/api' +import { useEffect, useState } from 'react' +import { download_state, DownloadStatus, AccountModel, FolderProgress } from '@/api/account/api' import { format } from 'date-fns' import { ScrollArea } from '@/components/ui/scroll-area' import { Badge } from '@/components/ui/badge' @@ -114,11 +115,19 @@ function FolderDetailItem({ f, t }: { f: FolderProgress, t: (key: string) => str {f.message && (
-
- +
+ {f.message.toLowerCase().includes('slow') || f.message.toLowerCase().includes('limiting') ? ( + + ) : ( + + )}

{t('accounts.runningState.message')}:

-

{f.message}

+

{f.message}

@@ -133,12 +142,37 @@ export function RunningStateDialog({ currentRow, open, onOpenChange }: Props) { const { data: state, isLoading } = useQuery({ queryKey: ['running-state', currentRow.id], queryFn: () => download_state(currentRow.id), - refetchInterval: 5000, + refetchInterval: (query) => { + const s = query.state.data?.active_session + return s && s.status === DownloadStatus.Running ? 5000 : false + }, enabled: open && !!currentRow.id, }) const session = state?.active_session const history = state?.history || [] + const isRunning = !!session && session.status === DownloadStatus.Running + + // When the poll is active, tick once per second so the "elapsed" and + // "last updated Xs ago" readouts stay fresh between refetches. + const [, setTick] = useState(0) + useEffect(() => { + if (!isRunning) return + const id = setInterval(() => setTick((v) => v + 1), 1000) + return () => clearInterval(id) + }, [isRunning]) + + const elapsedSec = session && isRunning + ? Math.max(0, Math.floor((Date.now() - new Date(session.start_time).getTime()) / 1000)) + : 0 + const formatDur = (s: number) => { + const h = Math.floor(s / 3600) + const m = Math.floor((s % 3600) / 60) + const sec = s % 60 + if (h > 0) return `${h}h ${m}m ${sec}s` + if (m > 0) return `${m}m ${sec}s` + return `${sec}s` + } return ( @@ -202,6 +236,23 @@ export function RunningStateDialog({ currentRow, open, onOpenChange }: Props) {
+ {isRunning && ( +
+
+

{t('accounts.runningState.session.elapsed')}

+
+ + {formatDur(elapsedSec)} +
+
+ {session.current_folder && ( +
+

{t('accounts.runningState.session.current_folder')}

+
{session.current_folder}
+
+ )} +
+ )} diff --git a/web/src/locales/ar.json b/web/src/locales/ar.json index e6ed802..36538eb 100644 --- a/web/src/locales/ar.json +++ b/web/src/locales/ar.json @@ -293,10 +293,14 @@ }, "message": "رسالة", "session": { + "current_folder": "مجلد البريد الحالي", + "elapsed": "الوقت المنقضي", + "last_update": "آخر تحديث", "started_at": "وقت البدء", "status": "الحالة", - "trigger": "المشغّل" + "trigger": "طريقة المشغّل" }, + "syncing": "جاري المزامنة", "tabs": { "active_session": "الجلسة النشطة", "errors": "أخطاء", diff --git a/web/src/locales/da.json b/web/src/locales/da.json index 037d010..e7a91f8 100644 --- a/web/src/locales/da.json +++ b/web/src/locales/da.json @@ -293,10 +293,14 @@ }, "message": "Besked", "session": { + "current_folder": "Nuværende postmappe", + "elapsed": "Varighed", + "last_update": "Senest opdateret", "started_at": "Starttid", "status": "Status", - "trigger": "Trigger" + "trigger": "Udløsermetode" }, + "syncing": "Synkroniserer", "tabs": { "active_session": "Aktiv session", "errors": "Fejl", diff --git a/web/src/locales/de.json b/web/src/locales/de.json index aa3e939..06fe18a 100644 --- a/web/src/locales/de.json +++ b/web/src/locales/de.json @@ -293,10 +293,14 @@ }, "message": "Nachricht", "session": { + "current_folder": "Aktueller Ordner", + "elapsed": "Dauer", + "last_update": "Zuletzt aktualisiert", "started_at": "Startzeit", "status": "Status", "trigger": "Auslöser" }, + "syncing": "Synchronisieren...", "tabs": { "active_session": "Aktive Sitzung", "errors": "Fehler", diff --git a/web/src/locales/en.json b/web/src/locales/en.json index 7718131..ffa6f04 100644 --- a/web/src/locales/en.json +++ b/web/src/locales/en.json @@ -295,10 +295,14 @@ }, "message": "Message", "session": { + "current_folder": "Current mail folder", + "elapsed": "Elapsed time", + "last_update": "Last updated", "started_at": "Started At", "status": "Status", "trigger": "Trigger" }, + "syncing": "Syncing", "tabs": { "active_session": "Active Session", "errors": "Errors", diff --git a/web/src/locales/es.json b/web/src/locales/es.json index 4a9d134..8000c55 100644 --- a/web/src/locales/es.json +++ b/web/src/locales/es.json @@ -293,10 +293,14 @@ }, "message": "Mensaje", "session": { + "current_folder": "Carpeta de correo actual", + "elapsed": "Tiempo transcurrido", + "last_update": "Última actualización", "started_at": "Hora de inicio", "status": "Estado", "trigger": "Disparador" }, + "syncing": "Sincronizando", "tabs": { "active_session": "Sesión activa", "errors": "Errores", diff --git a/web/src/locales/fi.json b/web/src/locales/fi.json index fb7235c..fb03ddb 100644 --- a/web/src/locales/fi.json +++ b/web/src/locales/fi.json @@ -293,10 +293,14 @@ }, "message": "Viesti", "session": { + "current_folder": "Nykyinen postikansio", + "elapsed": "Kesto", + "last_update": "Viimeksi päivitetty", "started_at": "Aloitusaika", "status": "Tila", - "trigger": "Laukaisija" + "trigger": "Käynnistystapa" }, + "syncing": "Synkronoidaan", "tabs": { "active_session": "Aktiivinen istunto", "errors": "Virheet", diff --git a/web/src/locales/fr.json b/web/src/locales/fr.json index 72e446d..c535590 100644 --- a/web/src/locales/fr.json +++ b/web/src/locales/fr.json @@ -293,10 +293,14 @@ }, "message": "Message", "session": { + "current_folder": "Dossier de courrier actuel", + "elapsed": "Temps écoulé", + "last_update": "Dernière mise à jour", "started_at": "Heure de début", "status": "Statut", "trigger": "Déclencheur" }, + "syncing": "Synchronisation", "tabs": { "active_session": "Session active", "errors": "Erreurs", diff --git a/web/src/locales/it.json b/web/src/locales/it.json index 933220f..83be968 100644 --- a/web/src/locales/it.json +++ b/web/src/locales/it.json @@ -293,10 +293,14 @@ }, "message": "Messaggio", "session": { + "current_folder": "Cartella posta corrente", + "elapsed": "Tempo trascorso", + "last_update": "Ultimo aggiornamento", "started_at": "Ora di inizio", "status": "Stato", - "trigger": "Trigger" + "trigger": "Innesco" }, + "syncing": "Sincronizzazione in corso", "tabs": { "active_session": "Sessione attiva", "errors": "Errori", diff --git a/web/src/locales/jp.json b/web/src/locales/jp.json index 67e04fd..803146d 100644 --- a/web/src/locales/jp.json +++ b/web/src/locales/jp.json @@ -293,10 +293,14 @@ }, "message": "メッセージ", "session": { + "current_folder": "現在のメールフォルダー", + "elapsed": "経過時間", + "last_update": "最終更新", "started_at": "開始時刻", "status": "状態", "trigger": "トリガー" }, + "syncing": "同期中", "tabs": { "active_session": "実行中タスク", "errors": "エラー", diff --git a/web/src/locales/ko.json b/web/src/locales/ko.json index 6d9e1c7..078e537 100644 --- a/web/src/locales/ko.json +++ b/web/src/locales/ko.json @@ -293,10 +293,14 @@ }, "message": "메시지", "session": { + "current_folder": "현재 메일함", + "elapsed": "경과 시간", + "last_update": "최근 업데이트", "started_at": "시작 시간", "status": "상태", "trigger": "트리거" }, + "syncing": "동기화 중", "tabs": { "active_session": "현재 작업", "errors": "오류", diff --git a/web/src/locales/nl.json b/web/src/locales/nl.json index e77191e..3605b1d 100644 --- a/web/src/locales/nl.json +++ b/web/src/locales/nl.json @@ -293,10 +293,14 @@ }, "message": "Bericht", "session": { + "current_folder": "Huidige e-mailmap", + "elapsed": "Verstreken tijd", + "last_update": "Laatst bijgewerkt", "started_at": "Starttijd", "status": "Status", - "trigger": "Trigger" + "trigger": "Triggermethode" }, + "syncing": "Synchroniseren", "tabs": { "active_session": "Actieve sessie", "errors": "Fouten", diff --git a/web/src/locales/no.json b/web/src/locales/no.json index 0707517..64e7fa9 100644 --- a/web/src/locales/no.json +++ b/web/src/locales/no.json @@ -293,10 +293,14 @@ }, "message": "Melding", "session": { + "current_folder": "Nåværende e-postmappe", + "elapsed": "Varighet", + "last_update": "Sist oppdatert", "started_at": "Starttid", "status": "Status", - "trigger": "Trigger" + "trigger": "Utløser" }, + "syncing": "Synkroniserer", "tabs": { "active_session": "Aktiv økt", "errors": "Feil", diff --git a/web/src/locales/pl.json b/web/src/locales/pl.json index 1569d0f..3141ca2 100644 --- a/web/src/locales/pl.json +++ b/web/src/locales/pl.json @@ -293,10 +293,14 @@ }, "message": "Wiadomość", "session": { + "current_folder": "Bieżący folder poczty", + "elapsed": "Czas trwania", + "last_update": "Ostatnia aktualizacja", "started_at": "Czas rozpoczęcia", "status": "Status", "trigger": "Wyzwalacz" }, + "syncing": "Synchronizowanie", "tabs": { "active_session": "Aktywna sesja", "errors": "Błędy", diff --git a/web/src/locales/pt.json b/web/src/locales/pt.json index 581b5d6..6e4ff43 100644 --- a/web/src/locales/pt.json +++ b/web/src/locales/pt.json @@ -293,10 +293,14 @@ }, "message": "Mensagem", "session": { + "current_folder": "Pasta de e-mail atual", + "elapsed": "Tempo decorrido", + "last_update": "Última atualização", "started_at": "Hora de início", "status": "Status", "trigger": "Gatilho" }, + "syncing": "Sincronizando", "tabs": { "active_session": "Sessão ativa", "errors": "Erros", diff --git a/web/src/locales/ru.json b/web/src/locales/ru.json index c6178e1..de4c92c 100644 --- a/web/src/locales/ru.json +++ b/web/src/locales/ru.json @@ -293,10 +293,14 @@ }, "message": "Сообщение", "session": { + "current_folder": "Текущая почтовая папка", + "elapsed": "Прошло времени", + "last_update": "Последнее обновление", "started_at": "Время начала", "status": "Статус", "trigger": "Триггер" }, + "syncing": "Синхронизация", "tabs": { "active_session": "Активная сессия", "errors": "Ошибки", diff --git a/web/src/locales/sv.json b/web/src/locales/sv.json index 5c96fde..06b9182 100644 --- a/web/src/locales/sv.json +++ b/web/src/locales/sv.json @@ -293,10 +293,14 @@ }, "message": "Meddelande", "session": { + "current_folder": "Aktuell e-postmapp", + "elapsed": "Tidsåtgång", + "last_update": "Senast uppdaterad", "started_at": "Starttid", "status": "Status", - "trigger": "Trigger" + "trigger": "Utlösare" }, + "syncing": "Synkroniserar", "tabs": { "active_session": "Aktiv session", "errors": "Fel", diff --git a/web/src/locales/zh-tw.json b/web/src/locales/zh-tw.json index e1154d3..bbc95a5 100644 --- a/web/src/locales/zh-tw.json +++ b/web/src/locales/zh-tw.json @@ -293,10 +293,14 @@ }, "message": "訊息", "session": { + "current_folder": "目前郵件資料夾", + "elapsed": "耗時", + "last_update": "最近更新", "started_at": "開始時間", "status": "狀態", "trigger": "觸發方式" }, + "syncing": "同步中", "tabs": { "active_session": "目前任務", "errors": "錯誤", diff --git a/web/src/locales/zh.json b/web/src/locales/zh.json index ad52f98..f21e3e5 100644 --- a/web/src/locales/zh.json +++ b/web/src/locales/zh.json @@ -295,10 +295,14 @@ }, "message": "消息", "session": { + "current_folder": "当前邮件夹", + "elapsed": "耗时", + "last_update": "最近更新", "started_at": "开始时间", "status": "状态", "trigger": "触发方式" }, + "syncing": "同步中", "tabs": { "active_session": "当前任务", "errors": "错误",